标签:可靠性 bit param returns ttl routing dlx alt letter
一.消息的可靠投递
在使用RabbitMq的时候,作为消息发送方希望杜绝任何消息丢失或者投递失败的场景。RabbitMQ为我们提供了两种方式用来控制消息的投递可靠性
rabbitMQ 整个消息投递过程为:
producer -> rabbitMQ broker -> exchange -> queue ->consumer
1.confirm 确认模式
消息从producer 到exchange 会返回一个 confirmCallback
2.return 退回模式
消息从exchange到queue投递失败则会返回一个returnCallbak
Confirm确认模式:
1.在配置文件中开启确认模式 publisher-confimrs :true
spring: rabbitmq: host: 10.211.55.4 virtual-host: local port: 5672 username: admin password: admin publisher-confirms: true
2.在rabbitTemplate定义ConfirmCallBack回调函数
@RequestMapping("/producer") public void producer(){ rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() { @Override /** * ack exchange交换机是否收到了消息 * cause 失败原因 */ public void confirm(CorrelationData correlationData, boolean ack, String cause) { if(ack){ System.out.println("confirm方法 成功了"); }else{ System.out.println("confirm方法 失败了: " + cause); } } }); rabbitTemplate.convertAndSend(RabbitMQConfig.EXCHANGE_NAME,"boot.test","spring boot rabbit mq"); }
Return退回模式:
1.开启退回模式
spring: rabbitmq: host: 10.211.55.4 virtual-host: local port: 5672 username: admin password: admin publisher-returns: true
2.设置returnCallBack
@RequestMapping("/return") public void returncallback(){ //设置交换机处理失败消息的模式 rabbitTemplate.setMandatory(true); //设置return call back rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() { /** * * @param message 消息对象 * @param replyCode 错误码 * @param replyText 错误信息 * @param exchange 交换机 * @param routingKey 路由键 */ @Override public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) { System.out.println("return 执行了...."); System.out.println(message); System.out.println(replyCode); System.out.println(replyText); System.out.println(exchange); System.out.println(routingKey); //处理 } }); rabbitTemplate.convertAndSend(RabbitMQConfig.EXCHANGE_NAME,"zxcasd","return call back"); }
二、Consumer ACK 确认机制
表示消费端收到消息后到确认方式
有三种确认方式:
1.自动确认 acknowledge="none" 默认
2.手动确认 acknowledge="manual"
3.根据异常情况确认: acknowledge="auto"
1.开启手动确认
spring: rabbitmq: host: 10.211.55.4 virtual-host: local port: 5672 username: admin password: admin publisher-returns: true listener: simple: acknowledge-mode: manual
2.成功调用channel.basicAck 异常调用channel.basicNack
@Component public class RabbitMQListener implements ChannelAwareBatchMessageListener { @Override @RabbitListener(queues = RabbitMQConfig.QUEUE_NAME) public void onMessage(Message message, Channel channel) throws Exception { long deliveryTag = message.getMessageProperties().getDeliveryTag(); try{ System.out.println(new String(message.getBody())); int a = 1/0; //手动确认 channel.basicAck(deliveryTag,true); }catch (Exception ex){ //发生异常 拒绝确认 //第三个参数,重回队列 channel.basicNack(deliveryTag,true,true); } } }
三、消费端 限流
1.确保ack机制为手动确认
2.设置prefetch=1 每次从rabbitmq取几条,ack确认后再取下一条
server: port: 9999 spring: rabbitmq: host: 10.211.55.4 virtual-host: local port: 5672 username: admin password: admin publisher-returns: true listener: simple: acknowledge-mode: manual direct: prefetch: 1
四、TTL
Time To Live 存活时间/过期时间,当消息到达一定时间后,还没有被消费,会被自动清除。
RabbitMQ可以对消息设置过期时间,也可以对整个队列设置过期时间。
1.queue 队列设置过期时间
@Configuration public class RabbitMQConfig { public static final String EXCHANGE_NAME = "exchange_ttl"; public static final String QUEUE_NAME = "queue_ttl"; public static final String Routing_Key = "ttl.#"; // 1.交换机 @Bean("bootExchange") public Exchange bootExchange() { return ExchangeBuilder.topicExchange(EXCHANGE_NAME).durable(true).build(); } //2.Queue队列 @Bean("bootQueue") public Queue bootQueue() { return QueueBuilder.durable(QUEUE_NAME).withArgument("x-message-ttl", 5000).build(); } //3.绑定交换机和队列 @Bean public Binding bindQueueExchange(@Qualifier("bootQueue") Queue queue, @Qualifier("bootExchange") Exchange exchange) { return BindingBuilder.bind(queue).to(exchange).with(Routing_Key).noargs(); } }
2.消息设置过期时间
@RequestMapping("/producer") public void producer(){ MessagePostProcessor messagePostProcessor = new MessagePostProcessor() { @Override public Message postProcessMessage(Message message) throws AmqpException { message.getMessageProperties().setExpiration("5000"); // 消息过期时间 return message; } }; for(int i = 0;i<10;i++){ if(i>5){ rabbitTemplate.convertAndSend(RabbitMQConfig.EXCHANGE_NAME,"ttl.test","spring boot rabbit mq"+i,messagePostProcessor); }else{ rabbitTemplate.convertAndSend(RabbitMQConfig.EXCHANGE_NAME,"ttl.test","spring boot rabbit mq"+i); } } }
五、死信队列
DLX、Dead Letter Exchange (死信交换机),当消息成为Dead Messag后,可以被重新发送到另一台交换机中。
消息成为死信到三种情况:
1.队列消息长度达到限制,
2.消费者拒接消费信息,basicNack/basicReject 并且不把消息重新放入原目标队列,参数requeue=false。
3.原队列存在消息过期设置,消息到达超时时间未被消费
设定参数 x-dead-letter-exchange 和 x-dead-letter-routing-key
标签:可靠性 bit param returns ttl routing dlx alt letter
原文地址:https://www.cnblogs.com/baidawei/p/13306124.html