RabbitMQ之消费者ACK 功能

什么是消费者ACK？
简单说就是 消息确认机制 mq要包管消息能可靠的到达消费者。消费者在消费是是可以指定autoAck参数（自动/手动）的方式告诉mq本身是否确认收到消息。
当 autoAck 参数为 false 时， 队列中的消息分成了两部门： 一部门是等待投递给消费者的消息；一部门是已经投递给消费者，但是还没有收到消费者确认信号的消息。 若 RabbitMQ 服务器端不停没有收到消费者简直认信号，而且消费此消息的消费者已经断开连接， 则服务器端会安排该消息重新进入队列，等待投递给下一个消费者（也可能还是原来的那个消费者）。
如下图

可以看到当前队列中的 “Ready” 状态和 “Unacked” 状态的消息数，分别对应等待投递给消费者的消息数和已经投递给消费者但是未收到确认信号的消息数。
上述分析 消息必要确认后消息将会被释放掉 从磁盘/内存中。
所以消息确认分类有两大类 如下图

由上图可知
有发送方确认、接收方确认。
注意：
其中发送方确认又分为：生产者到互换器到确认、互换器到队列简直认
本节主要介绍 消费者消息确认
方式一，自动确认。
方式二，手动确认。
对于自动确认的方式，RabbitMQ Broker 只要将消息写入到 TCP Socket 中乐成，就认为该消息投递乐成，而无需 Consumer 手动确认。
对于手动确认的方式，RabbitMQ Broker 将消息发送给 Consumer 之后，由 Consumer 手动确认之后，才任务消息投递乐成。
实际场景下，因为自动确认存在可能丢失消息的情况，所以在对可靠性有要求的场景下，我们基本采用手动确认。当然，假如允许消息有一定的丢失，对性能有更高的产经下，我们可以思量采用自动确认。
进行实战演练
yml配置类

1. spring:
2.   rabbitmq:
3.           #host为一般模式 若集群模式 将key换成addresses的形式
4.     host: 192.168.9.104
5.     port: 5672
6.     #账号密码自行替换
7.     username: admin
8.     password: admin
9.     # 以下手动提交消息
10.     listener:
11.    
12.       simple:
13.         acknowledge-mode: manual
14.       direct:
15.         acknowledge-mode: manual

复制代码

以上ackoneledge-mode的值有如下：
NONE 对应 Consumer 的自动确认
MANUAL 对应 Consumer 的手动确认，由开发者在消费逻辑中，手动进行确认。
AUTO 对应 Consumer 的手动确认，在消费消息完成（包括正常返回、和抛出非常）后，由 Spring-AMQP 框架来“自动”进行确认。
以下为逻辑代码演示

1. ================》Direct Exchange 配置
2. @Configuration
3. public class DirectExchangeConfiguration {
4.     /**
5.      * 创建一个 Queue
6.      *
7.      * @return Queue
8.      */
9.     @Bean
10.     public Queue queue05() {
11.         // Queue:名字 | durable: 是否持久化 | exclusive: 是否排它 | autoDelete: 是否自动删除
12.         return new Queue(
13.                 Message05.QUEUE,
14.                 true,
15.                 false,
16.                 false);
17.     }
19.     /**
20.      * 创建 Direct Exchange
21.      *
22.      * @return DirectExchange
23.      */
24.     @Bean
25.     public DirectExchange exchange05() {
26.         // name: 交换机名字 | durable: 是否持久化 | exclusive: 是否排它
27.         return new DirectExchange(Message05.EXCHANGE,
28.                 true,
29.                 false);
30.     }
32.     /**
33.      * 创建 Binding
34.      * Exchange：Message05.EXCHANGE
35.      * Routing key：Message05.ROUTING_KEY
36.      * Queue：Message05.QUEUE
37.      *
38.      * @return Binding
39.      */
40.     @Bean
41.     public Binding binding05() {
42.         return BindingBuilder
43.                 .bind(queue05()).to(exchange05())
44.                 .with(Message05.ROUTING_KEY);
45.     }
46. =======================================》 direct 类型的消息对象
47. @Data
48. public class Message05 implements Serializable {
50.     public static final String QUEUE = "QUEUE_05";
52.     public static final String EXCHANGE = "EXCHANGE_05";
54.     public static final String ROUTING_KEY = "ROUTING_KEY_05";
56.     private String id;
57. }
58. =======================================》 生产者逻辑
59. @Component
60. public class Producer05 {
61.     @Resource
62.     private RabbitTemplate rabbitTemplate;
64.     public void syncSend(String id) {
65.         // 创建 Message05 消息
66.         Message05 message = new Message05();
67.         message.setId(id);
68.         // 同步发送消息
69.         rabbitTemplate.convertAndSend(Message05.EXCHANGE, Message05.ROUTING_KEY, message);
70.     }
71. }
72. =======================================》 消费者逻辑
73. @Component
74. @RabbitListener(queues = Message05.QUEUE)
75. @Slf4j
76. public class Consumer05 {
78.     /**
79.      *  ack 模式 ，
80.      * 这里需要和 application.yml 中的 acknowledge-mode: manual  对应一起使用
81.      */
82.     @RabbitHandler
83.     public void onMessageAck(Message05 message01, Message message, Channel channel) throws IOException {
84.         try {
85.             log.info("[Consumer05 onMessageAck][线程编号:{} 消息内容：{}]", Thread.currentThread().getId(), message01);
86.             //  如果手动ACK,消息会被监听消费,但是消息在队列中依旧存在,如果 未配置 acknowledge-mode 默认是会在消费完毕后自动ACK掉
87.             long deliveryTag = message.getMessageProperties().getDeliveryTag();
88.             // 取当前时间，达到一个随机效果，测试的话可以多跑几次试试
89.             if (System.currentTimeMillis() % 2 == 1) {
90.                 // 通知 MQ 消息已被成功消费,可以ACK了
91.                 // 第二个参数 multiple ，用于批量确认消息，为了减少网络流量，手动确认可以被批处。
92.                 // 1. 当 multiple 为 true 时，则可以一次性确认 deliveryTag 小于等于传入值的所有消息
93.                 // 2. 当 multiple 为 false 时，则只确认当前 deliveryTag 对应的消息
94.                 channel.basicAck(deliveryTag, false);
95.                 log.info("[Consumer05 onMessageAck][正常ack:{}]", message01);
96.             } else {
97.                 log.info("[Consumer05 onMessageAck][未ack:{}]", message01);
98.                 throw new RuntimeException("手动异常");
99.             }
100.         } catch (Exception e) {
101.             // 处理失败,重新压入MQ
102.             channel.basicRecover();
103.             log.info("[Consumer05 onMessageAck][消息重新压入MQ:{}]", message01);
104.         }
105.     }
106. =======================================》测试类
107. @Test
108.     void syncSend() {
109.         String id = UUID.randomUUID().toString();
110.         producer05.syncSend(id);
111.         log.info("[test producer05 syncSend][id:{}] 发送成功", id);
113.         TimeUnit.SECONDS.sleep(2);
114.     }

复制代码

以上的是消费者ACK实现的代码 若不相识rabbitmq的基本使用 建议先看看我前面对应的文章 文章链接：点我—>let’s go
若需完整代码 可识别二维码后 给您发代码。

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。