RabbitMQ之消费者ACK 功能

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

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

由上图可知
有发送方确认、接收方确认。
注意：
其中发送方确认又分为：生产者到交换器到确认、交换器到队列的确认
本节主要介绍 消费者消息确认
方式一，自动确认。
方式二，手动确认。
对于自动确认的方式，RabbitMQ Broker 只要将消息写入到 TCP Socket 中成功，就认为该消息投递成功，而无需 Consumer 手动确认。

对于手动确认的方式，RabbitMQ Broker 将消息发送给 Consumer 之后，由 Consumer 手动确认之后，才任务消息投递成功。

实际场景下，因为自动确认存在可能丢失消息的情况，所以在对可靠性有要求的场景下，我们基本采用手动确认。当然，如果允许消息有一定的丢失，对性能有更高的产经下，我们可以考虑采用自动确认。

进行实战演练
yml配置类

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

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

================》Direct Exchange 配置
@Configuration
public class DirectExchangeConfiguration {
    /**
     * 创建一个 Queue
     *
     * @return Queue
     */
    @Bean
    public Queue queue05() {
        // Queue:名字 | durable: 是否持久化 | exclusive: 是否排它 | autoDelete: 是否自动删除
        return new Queue(
                Message05.QUEUE,
                true,
                false,
                false);
    }

    /**
     * 创建 Direct Exchange
     *
     * @return DirectExchange
     */
    @Bean
    public DirectExchange exchange05() {
        // name: 交换机名字 | durable: 是否持久化 | exclusive: 是否排它
        return new DirectExchange(Message05.EXCHANGE,
                true,
                false);
    }

    /**
     * 创建 Binding
     * Exchange：Message05.EXCHANGE
     * Routing key：Message05.ROUTING_KEY
     * Queue：Message05.QUEUE
     *
     * @return Binding
     */
    @Bean
    public Binding binding05() {
        return BindingBuilder
                .bind(queue05()).to(exchange05())
                .with(Message05.ROUTING_KEY);
    }
=======================================》 direct 类型的消息对象
@Data
public class Message05 implements Serializable {

    public static final String QUEUE = "QUEUE_05";

    public static final String EXCHANGE = "EXCHANGE_05";

    public static final String ROUTING_KEY = "ROUTING_KEY_05";

    private String id;
}
=======================================》 生产者逻辑
@Component
public class Producer05 {
    @Resource
    private RabbitTemplate rabbitTemplate;

    public void syncSend(String id) {
        // 创建 Message05 消息
        Message05 message = new Message05();
        message.setId(id);
        // 同步发送消息
        rabbitTemplate.convertAndSend(Message05.EXCHANGE, Message05.ROUTING_KEY, message);
    }
}
=======================================》 消费者逻辑
@Component
@RabbitListener(queues = Message05.QUEUE)
@Slf4j
public class Consumer05 {

    /**
     *  ack 模式 ，
     * 这里需要和 application.yml 中的 acknowledge-mode: manual  对应一起使用
     */
    @RabbitHandler
    public void onMessageAck(Message05 message01, Message message, Channel channel) throws IOException {
        try {
            log.info("[Consumer05 onMessageAck][线程编号:{} 消息内容：{}]", Thread.currentThread().getId(), message01);
            //  如果手动ACK,消息会被监听消费,但是消息在队列中依旧存在,如果 未配置 acknowledge-mode 默认是会在消费完毕后自动ACK掉
            long deliveryTag = message.getMessageProperties().getDeliveryTag();
            // 取当前时间，达到一个随机效果，测试的话可以多跑几次试试
            if (System.currentTimeMillis() % 2 == 1) {
                // 通知 MQ 消息已被成功消费,可以ACK了
                // 第二个参数 multiple ，用于批量确认消息，为了减少网络流量，手动确认可以被批处。
                // 1. 当 multiple 为 true 时，则可以一次性确认 deliveryTag 小于等于传入值的所有消息
                // 2. 当 multiple 为 false 时，则只确认当前 deliveryTag 对应的消息
                channel.basicAck(deliveryTag, false);
                log.info("[Consumer05 onMessageAck][正常ack:{}]", message01);
            } else {
                log.info("[Consumer05 onMessageAck][未ack:{}]", message01);
                throw new RuntimeException("手动异常");
            }
        } catch (Exception e) {
            // 处理失败,重新压入MQ
            channel.basicRecover();
            log.info("[Consumer05 onMessageAck][消息重新压入MQ:{}]", message01);
        }
    }
=======================================》测试类
 @Test
    void syncSend() {
        String id = UUID.randomUUID().toString();
        producer05.syncSend(id);
        log.info("[test producer05 syncSend][id:{}] 发送成功", id);

        TimeUnit.SECONDS.sleep(2);
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114

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