消息队列手动确认Ack

小灰灰 2023-06-30 11:54 25阅读 0赞

以RabbitMQ为例，默认情况下 RabbitMQ 是自动ACK机制，就意味着 MQ 会在消息发送完毕后，自动帮我们去ACK，然后删除消息的信息。  
这样依赖就存在这样一个问题：  
如果消费者处理消息需要较长时间，最好的做法是消费端处理完之后手动去确认。

**1、配置文件：**

rabbitmq:
        host: ${yun.activity.rabbitmq.host}
        port: ${yun.activity.rabbitmq.port}
        username: ${yun.activity.rabbitmq.username}
        password: ${yun.activity.rabbitmq.password}
        virtual-host: ${yun.activity.rabbitmq.virtual-host}
        publisher-confirms: true
        publisher-returns: true
        template:
          mandatory: true
        listener:
          simple:
            acknowledge-mode: manual

***通过acknowledge-mode: manual 设置为手动设置Ack模式。***

**2、消费者：**

@RabbitListener(queues = "activity-eleven-first-notify", containerFactory = "activityElevenCainerFactory")
        public void processNormalOrder(Message message, Channel channel) {
            try {
                dealInfo(message);
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                try {
                    channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }

以上代码通过，***channel.basicAck去手动回复消息处理状态***，

除了basicAck之外，RabbitMQ还提供了其他几个方法进行设置Ack，如下：

另外，消息的确认类型：

1）channel.basicAck(deliveryTag, multiple);  
consumer处理成功后，通知broker删除队列中的消息，如果设置multiple=true，表示支持批量确认机制以减少网络流量。  
例如：有值为5,6,7,8 deliveryTag的投递  
如果此时channel.basicAck(8, true);则表示前面未确认的5,6,7投递也一起确认处理完毕。  
如果此时channel.basicAck(8, false);则仅表示deliveryTag=8的消息已经成功处理。

2）channel.basicNack(deliveryTag, multiple, requeue);  
consumer处理失败后，例如：有值为5,6,7,8 deliveryTag的投递。  
如果channel.basicNack(8, true, true);表示deliveryTag=8之前未确认的消息都处理失败且将这些消息重新放回队列中。  
如果channel.basicNack(8, true, false);表示deliveryTag=8之前未确认的消息都处理失败且将这些消息直接丢弃。  
如果channel.basicNack(8, false, true);表示deliveryTag=8的消息处理失败且将该消息重新放回队列。  
如果channel.basicNack(8, false, false);表示deliveryTag=8的消息处理失败且将该消息直接丢弃。

3）channel.basicReject(deliveryTag, requeue);  
相比channel.basicNack，除了没有multiple批量确认机制之外，其他语义完全一样。  
如果channel.basicReject(8, true);表示deliveryTag=8的消息处理失败且将该消息重新放回队列。  
如果channel.basicReject(8, false);表示deliveryTag=8的消息处理失败且将该消息直接丢弃。

由此可见，手动Ack如果处理方式不对会发生一些问题。  
1.没有及时ack，或者程序出现bug，所有的消息将被存在unacked中，消耗内存  
如果忘记了ack，那么后果很严重。当Consumer退出时，Message会重新分发。然后RabbitMQ会占用越来越多的内存，由于 RabbitMQ会长时间运行，因此这个“内存泄漏”是致命的。  
2.如果使用BasicNack，将消费失败的消息重新塞进队列的头部，则会造成死循环。  
（解决basicNack造成的消息循环循环消费的办法是为队列设置“回退队列”，设置回退队列和阀值，如设置队列为q1，阀值为2，则在rollback两次后将消息转入q1）

综上，手动ack需要注意的是：  
1.在消费者端一定要进行ack，或者是nack，可以放在try方法块的finally中执行  
2.可以对消费者的异常状态进行捕捉，根据异常类型选择ack，或者nack抛弃消息，nack再次尝试  
3.对于nack的再次尝试，是进入到队列头的，如果一直是失败的状态，将会造成阻塞。所以最好是专门投递到“死信队列”，