RabbitMQ 实战(四)消费者 ack 以及 生产者 confirms


声明:本文转载自https://my.oschina.net/u/3523423/blog/1620885,转载目的在于传递更多信息,仅供学习交流之用。如有侵权行为,请联系我,我会及时删除。

插播一条广告→2021 ByteDance字节跳动内推←各城市、各方向的岗位都有,大量招人!


这篇文章主要讲 RabbitMQ 中 消费者 ack 以及 生产者 confirms。

如上图,生产者把消息发送到 RabbitMQ,然后 RabbitMQ 再把消息投递到消费者。

生产者和 RabbitMQ,以及 RabbitMQ 和消费者都是通过 TCP 连接,但是他们之间是通过信道(Channel)传递数据的。多个线程共享一个连接,但是每个线程拥有独自的信道。

消费者 ack

  • 问题:怎么保证 RabbitMQ 投递的消息被成功投递到了消费者?

    RabbitMQ 投递的消息,刚投递一半,产生了网络抖动,就有可能到不了消费者。

  • 解决办法:

    RabbitMQ 对消费者说:“如果你成功接收到了消息,给我说确认收到了,不然我就当你没有收到,我还会重新投递”

在 RabbitMQ 中,有两种 acknowledgement 模式。

自动 acknowledgement 模式

这也称作发后即忘模式

在这种模式下,RabbitMQ 投递了消息,在投递成功之前,如果消费者的 TCP 连接 或者 channel 关闭了,这条消息就会丢失。

会有丢失消息问题。

手动 acknowledgement 模式

在这种模式下,RabbitMQ 投递了消息,在投递成功之前,如果消费者的 TCP 连接 或者 channel 关闭了,导致这条消息没有被 acked,RabbitMQ 会自动把当前消息重新入队,再次投递。

会有重复投递消息的问题,所以消费者得准备好处理重复消息的问题,就是所谓的:幂等性。

为了启用 手动 ack 模式,消费者需要实现 ChannelAwareMessageListener 接口。

@Component public class Consumer implements ChannelAwareMessageListener {      @Autowired     private MessageConverter messageConverter;      @Override     public void onMessage(Message message, Channel channel) throws Exception {         MessageProperties messageProperties = message.getMessageProperties();          // 代表投递的标识符,唯一标识了当前信道上的投递,通过 deliveryTag ,消费者就可以告诉 RabbitMQ 确认收到了当前消息,见下面的方法         long deliveryTag = messageProperties.getDeliveryTag();          // 如果是重复投递的消息,redelivered 为 true         Boolean redelivered = messageProperties.getRedelivered();          // 获取生产者发送的原始消息         Object originalMessage = messageConverter.fromMessage(message);          Console.log("consume message = {} , deliveryTag = {} , redelivered = {}"                 , originalMessage, deliveryTag, redelivered);          // 代表消费者确认收到当前消息,第二个参数表示一次是否 ack 多条消息         channel.basicAck(deliveryTag, false);          // 代表消费者拒绝一条或者多条消息,第二个参数表示一次是否拒绝多条消息,第三个参数表示是否把当前消息重新入队 //        channel.basicNack(deliveryTag, false, false);          // 代表消费者拒绝当前消息,第二个参数表示是否把当前消息重新入队 //        channel.basicReject(deliveryTag,false);      } } 
  • channel.basicAck

    代表消费者确认收到当前消息,语义上表示消费者成功处理了当前消息。

  • channel.basicNack

    代表消费者拒绝一条或者多条消息。basicNack 算是 basicReject 的一个扩展,因为 basicReject 不能一次拒绝多条消息。

  • channel.basicReject

    代表消费者拒绝这条消息,语义上表示消费者没有处理当前消息。

    对于 basicNack 和 basicReject ,如果参数 boolean requeue 传入 false,消息还是会从队列里面删除。这三个方法只是语义上的不同。

  • deliveryTag

    deliveryTag 是 64 bit long 值,从 1 开始,不停的递增 1。不同的 channel 有独立的 deliveryTag。比如有两个消费者,你会发现,都是从 1 开始递增,互不影响。

由于上面创建的消费者,没有指明监听那个队列,所以还需要创建一个 MessageListenerContainer

@Bean public SimpleMessageListenerContainer container(ConnectionFactory connectionFactory, ChannelAwareMessageListener listener) {     SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();     container.setConnectionFactory(connectionFactory);      // 指定消费者     container.setMessageListener(listener);     // 指定监听的队列     container.setQueueNames(QUEUE_NAME);      // 设置消费者的 ack 模式为手动确认模式     container.setAcknowledgeMode(AcknowledgeMode.MANUAL);      container.setPrefetchCount(300);      return container; } 

这样就开启了消费者手动 ack 模式。

注意

如果开启了消费者手动 ack 模式,但是又没有调用手动确认方法(比如:channel.basicAck),那问题就大了,RabbitMQ 会在当前 channel 上一直阻塞,等待消费者 ack。

生产者 confirms

  • 问题:怎么保证生产者发送的消息被 RabbitMQ 成功接收?

    生产者发送的消息,刚发送一半,产生了网络抖动,就有可能到不了 RabbitMQ。

  • 解决办法:

    生产者对 RabbitMQ 说:“如果你成功接收到了消息,给我说确认收到了,不然我就当你没有收到”

自定义消息元数据

/**  * 自定义消息元数据  */ @NoArgsConstructor @Data public class RabbitMetaMessage implements Serializable{      /**      * 是否是 returnCallback      */     private boolean returnCallback;      /**      * 承载原始消息数据数据      */     private Object payload;      public RabbitMetaMessage(Object payload) {         this.payload = payload;     } } 
  • returnCallback 标记当前消息是否触发了 returnCallback(后面会解释)
  • payload 保存原始消息数据

生产者

先把消息存储到 redis,再发送到 rabbitmq

@RestController public class ProducerController {      @Autowired     private RabbitTemplate rabbitTemplate;      @Autowired     private RedisTemplate redisTemplate;      @Autowired     private DefaultKeyGenerator keyGenerator;      @GetMapping("/sendMessage")     public Object sendMessage() {         new Thread(() -> {             HashOperations hashOperations = redisTemplate.opsForHash();             for (int i = 0; i < 1; i++) {                 String id = keyGenerator.generateKey() + "";                 String value = "message " + i;                 RabbitMetaMessage rabbitMetaMessage = new RabbitMetaMessage(value);                  // 先把消息存储到 redis                 hashOperations.put(RedisConfig.RETRY_KEY, id, rabbitMetaMessage);                  Console.log("send message = {}", value);                  // 再发送到 rabbitmq                 rabbitTemplate.convertAndSend(RabbitMQConfig.EXCHANGE_NAME, RabbitMQConfig.ROUTING_KEY, value, (message) -> {                     message.getMessageProperties().setMessageId(id);                     return message;                 }, new CorrelationData(id));             }         }).start();         return "ok";     }  } 

配置 ConnectionFactory

@Bean public ConnectionFactory connectionFactory() {     CachingConnectionFactory connectionFactory = new CachingConnectionFactory("192.168.238.132", 5672);     connectionFactory.setUsername("guest");     connectionFactory.setPassword("guest");      // 设置 生产者 confirms     connectionFactory.setPublisherConfirms(true);      // 设置 生产者 Returns     connectionFactory.setPublisherReturns(true);      return connectionFactory; } 

配置 RabbitTemplate

@Bean public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {     RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);      // 必须设置为 true,不然当 发送到交换器成功,但是没有匹配的队列,不会触发 ReturnCallback 回调     // 而且 ReturnCallback 比 ConfirmCallback 先回调,意思就是 ReturnCallback 执行完了才会执行 ConfirmCallback     rabbitTemplate.setMandatory(true);      // 设置 ConfirmCallback 回调     rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {         Console.log("ConfirmCallback , correlationData = {} , ack = {} , cause = {} ", correlationData, ack, cause);         // 如果发送到交换器都没有成功(比如说删除了交换器),ack 返回值为 false         // 如果发送到交换器成功,但是没有匹配的队列(比如说取消了绑定),ack 返回值为还是 true (这是一个坑,需要注意)         if (ack) {             String messageId = correlationData.getId();             RabbitMetaMessage rabbitMetaMessage = (RabbitMetaMessage) redisTemplate.opsForHash().get(RedisConfig.RETRY_KEY, messageId);             Console.log("rabbitMetaMessage = {}", rabbitMetaMessage);             if (!rabbitMetaMessage.isReturnCallback()) {                  // 到这一步才能完全保证消息成功发送到了 rabbitmq                 // 删除 redis 里面的消息                 redisTemplate.opsForHash().delete(RedisConfig.RETRY_KEY, messageId);             }         }      });      // 设置 ReturnCallback 回调     // 如果发送到交换器成功,但是没有匹配的队列,就会触发这个回调     rabbitTemplate.setReturnCallback((message, replyCode, replyText,                                       exchange, routingKey) -> {         Console.log("ReturnCallback unroutable messages, message = {} , replyCode = {} , replyText = {} , exchange = {} , routingKey = {} ", message, replyCode, replyText, exchange, routingKey);          // 从 redis 取出消息,设置 returnCallback 设置为 true         String messageId = message.getMessageProperties().getMessageId();         RabbitMetaMessage rabbitMetaMessage = (RabbitMetaMessage) redisTemplate.opsForHash().get(RedisConfig.RETRY_KEY, messageId);         rabbitMetaMessage.setReturnCallback(true);         redisTemplate.opsForHash().put(RedisConfig.RETRY_KEY, messageId, rabbitMetaMessage);     });     return rabbitTemplate; } 

ReturnCallback 回调

必须 rabbitTemplate.setMandatory(true),不然当 发送到交换器成功,但是没有匹配的队列,不会触发 ReturnCallback 回调。而且 ReturnCallback 比 ConfirmCallback 先回调。

如何模拟 发送到交换器成功,但是没有匹配的队列,先把项目启动,然后再把队列解绑,再发送消息,就会触发 ReturnCallback 回调,而且发现消息也丢失了,没有到任何队列。

这样就解绑了。

运行项目,然后打开浏览器,输入 http://localhost:9999/sendMessage

控制台打出如下日志

这样就触发了 ReturnCallback 回调 ,从 redis 取出消息,设置 returnCallback 设置为 true。你会发现 ConfirmCallback 的 ack 返回值还是 true。

ConfirmCallback 回调

这里有个需要注意的地方,如果发送到交换器成功,但是没有匹配的队列(比如说取消了绑定),ack 返回值为还是 true (这是一个坑,需要注意,就像上面那种情况!!!)。所以不能单靠这个来判断消息真的发送成功了。这个时候会先触发 ReturnCallback 回调,我们把 returnCallback 设置为 true,所以还得判断 returnCallback 是否为 true,如果为 ture,表示消息发送不成功,false 才能完全保证消息成功发送到了 rabbitmq。

如何模拟 ack 返回值为 false,先把项目启动,然后再把交换器删除,就会发现 ConfirmCallback 的 ack 为 false。

运行项目,然后打开浏览器,输入 http://localhost:9999/sendMessage

控制台打出如下日志

你会发现 ConfirmCallback 的 ack 返回值才是 false。

注意

不能单单依靠 ConfirmCallback 的 ack 返回值为 true,就断定当前消息发送成功了。

源码地址

参考资料

Consumer Acknowledgements and Publisher Confirms

结语

由于本人知识和能力有限,文中如有没说清楚的地方,希望大家能在评论区指出,以帮助我将博文写得更好。

本文发表于2018年02月11日 00:31
(c)注:本文转载自https://my.oschina.net/u/3523423/blog/1620885,转载目的在于传递更多信息,并不代表本网赞同其观点和对其真实性负责。如有侵权行为,请联系我们,我们会及时删除.

阅读 1075 讨论 0 喜欢 0

抢先体验

扫码体验
趣味小程序
文字表情生成器

闪念胶囊

你要过得好哇,这样我才能恨你啊,你要是过得不好,我都不知道该恨你还是拥抱你啊。

直抵黄龙府,与诸君痛饮尔。

那时陪伴我的人啊,你们如今在何方。

不出意外的话,我们再也不会见了,祝你前程似锦。

这世界真好,吃野东西也要留出这条命来看看

快捷链接
网站地图
提交友链
Copyright © 2016 - 2021 Cion.
All Rights Reserved.
京ICP备2021004668号-1