RabbitMQ 消费者确认（Consumer Acknowledgements)

妖狐艹你老母 2023-05-29 03:40 13阅读 0赞

## 摘要 ##

RabbitMQ关于数据安全有两个特性：[发布者确认(Publisher Confirms][Publisher Confirms])和消费者确认(Consumer Acknowledegements)。前者用于MQ服务器(broker)告诉发布者传递消息的结果，它是消息协议的拓展内容；后者用于消费者告诉MQ服务器消息传递的结果，是消息协议包含的定义。本文主要介绍RabbitMQ对消费者确认的实现。

## 一条消息传递的标识：传递标签(delivery tags) ##

消息的传递是如何被标识的呢？  
当一个消费者被注册后，RabbitMQ通过basic.delivery方法传递消息，该方法会携带一个传递标签，它唯一的标识了通道上某条消息的传递。因此传递标签是按通道分的。  
传递标签是单调递增的正整数，消费者客户端确认消息的方法将会将传递标签作为参数。因为传递标签是按照通道分的，如果消息A是从通道1传递到消费者服务器的，那么消息A的确认也必须通过通道1，如果错误的通过通道2确认，RabbitMQ会抛出协议异常并关闭通道2。

## 消费者确认的模式 ##

当一个MQ服务器节点传递一条消息到消费者服务器，它需要决定什么时候认为这条消息已经被消费者成功处理了。消息协议通常会提供一个确认机制，允许消费者向他们连接的MQ服务器发送确认。确认机制是否启用一般在消费者订阅的时候决定。

取决于使用的确认模式，RabbitMQ可以在消息从MQ服务器发送出（写入TCP Socket）后立刻就将其当做已成功处理，或者当收到来自消费者显示的（手工的）的确认后。

#### 手动确认模式 ####

消费者手工发送的确认可以是以下一种协议方法：

*  `basic.ack(deliveryTag,multiple)` 用来确认成功消息(positive acknowledgements)
 *  `basic.nack(deliveryTag,requeue,multiple)` 用来确认失败的消息(negative acknowledgements)
 *  `basic.reject(deliveryTa,requeue)` 用来确认失败的消息

确认成功简单的令rabbitMQ将消息记录为已发送并丢弃。使用`basic.reject`  
方法令RabbitMQ记录消息为发送失败，但仍然需要丢弃。

#### 自动确认模式 ####

在自动确认模式(automatic acknowledgement)中，消息在发出去后就被认为是已成功处理。这种模式损失数据安全性来换取消息的高吞吐量，如果与消费者的TCP连接或者消息通道在成功发送前关闭了，则MQ服务器发送出的消息就丢失了。因此自动确认模式应该被认为是非数据安全的，应谨慎使用。

如果要使用自动确认模式，还有个要注意的地方是消费者的负载。手动确认模式典型的使用会在消费者端定义一个有限大小的队列(prefetch)用于存放未处理状态的消息。然而自动确认模式根本无此限制，因此有可能造成消费者服务器超出负荷，导致堆内存不够用而被操作系统终止进程。因此，只有在消费者能够已稳定并高效的处理消息的前提下，才建议使用确认模式。

## RabbitMQ手动确认示例 ##

#### 1，确认一条消息传递成功 ####

JAVA库使用 `Channel#basicAck` 和 `Channel#basicNack`方法分别来实现协议中定义的 `basic.ack` 和 `basic.nack`方法。示例如下：

// 假设已存在channel实例
    
    boolean autoAck = false;
    channel.basicConsume(queueName, autoAck, "a-consumer-tag",
         new DefaultConsumer(channel) { 
             @Override
             public void handleDelivery(String consumerTag,
                                        Envelope envelope,
                                        AMQP.BasicProperties properties,
                                        byte[] body)
                 throws IOException
             { 
                 long deliveryTag = envelope.getDeliveryTag();
                 // 确认一条消息成功传递，消息将会被RabbitMQ丢弃
                 channel.basicAck(deliveryTag, false);
             }
         });

#### 2，一次确认多条消息传递成功 ####

手工确认模式可以通过批次确认来减少网络流程，通过将 确认方法的的`multiple`参数设置为`true`。注意，`basic.reject`是没有这个参数的，RabbitMQ引入`basic.nack`方法带这个参数，该方法作为拓展协议的一部分。

当`multiple`参数被设为`true`。RabbitMQ将会确认所有传递标签小于给定数值的消息。比如通道`Ch`上有未确认消息，它们的传递标签是5,6,7,8，如果有确认带的传递标签是8，且`multiple`参数被设为`true`，则5-8消息都会被确认；如果`multiple`参数被设为`false`，则5,6,7消息仍未被确认。示例如下：

// 假设已存在channel实例
    
    boolean autoAck = false;
    channel.basicConsume(queueName, autoAck, "a-consumer-tag",
         new DefaultConsumer(channel) { 
             @Override
             public void handleDelivery(String consumerTag,
                                        Envelope envelope,
                                        AMQP.BasicProperties properties,
                                        byte[] body)
                 throws IOException
             { 
                 long deliveryTag = envelope.getDeliveryTag();
                 //确认所有传递标签小于deliveryTag的消息已成功传递，并丢弃它们
                 channel.basicAck(deliveryTag, true);
             }
         });

#### 3, 确认消息失败并重新入队消息 ####

有时消费者无法立即处理传递来的消息，但其他消费者有能力处理。在这种情况下，可能需要让消息重新入队并让另一个消费者接收和处理它。`basic.reject`和`basic.nack`是用于此的两种协议方法。  
上面两个方法通常被用于确认消息传递失败，MQ服务器可以丢弃这些消息或重新入队。可通过`requeue`参数来控制，当设为`true`时MQ服务器会将指定传递标签的消息重新入队。示例如下：

// 假设已存在channel实例
    
    boolean autoAck = false;
    channel.basicConsume(queueName, autoAck, "a-consumer-tag",
         new DefaultConsumer(channel) { 
             @Override
             public void handleDelivery(String consumerTag,
                                        Envelope envelope,
                                        AMQP.BasicProperties properties,
                                        byte[] body)
                 throws IOException
             { 
                 long deliveryTag = envelope.getDeliveryTag();
                 // 确认传递标签为deliveryTag的消息传递失败，并丢弃它
                 channel.basicReject(deliveryTag, false);
             }
         });

// 假设已存在channel实例
    
    boolean autoAck = false;
    channel.basicConsume(queueName, autoAck, "a-consumer-tag",
         new DefaultConsumer(channel) { 
             @Override
             public void handleDelivery(String consumerTag,
                                        Envelope envelope,
                                        AMQP.BasicProperties properties,
                                        byte[] body)
                 throws IOException
             { 
                 long deliveryTag = envelope.getDeliveryTag();
                 // 传递失败，重新入队
                 channel.basicReject(deliveryTag, true);
             }
         });

如果可能RabbitMQ会将重新入队的消息还放在它原来的位置，否则就放到尽可能离队首近的位置。假设某一瞬间出现，所有消费者的预取队列(prefetch)都已经满了(无法再接收消息)，则会出现一个重新入队/重新传递的循环，造成网络带宽和内存资源的消耗。消费者需要追踪重新传递的数量，丢弃确认失败的消息，或经过一定时延后再重新入队。

使用 `basic.nack`可以同时入队多条消息，它比`basic.reject`方法多了一个`multiple`参数，示例如下：

// 假设已存在channel实例
    
    boolean autoAck = false;
    channel.basicConsume(queueName, autoAck, "a-consumer-tag",
         new DefaultConsumer(channel) { 
             @Override
             public void handleDelivery(String consumerTag,
                                        Envelope envelope,
                                        AMQP.BasicProperties properties,
                                        byte[] body)
                 throws IOException
             { 
                 long deliveryTag = envelope.getDeliveryTag();
                 // 确认所有传递标签小于deliveryTag的消息传递失败，并重新入队它们
                 channel.basicNack(deliveryTag, true, true);
             }
         });

在手动确认模式下，如果传递的通道关闭或失去连接，则通道上未收到确认的消息会自动重新入队。这包括客户端的TCP连接丢失，消费者应用程序（进程）故障以及通道级协议异常。

考虑到消息会重新入队，消费者需要保证对消息操作的幂等性。被重发的消息，携带的`redeliver`会被RabbitMQ设置为`true`，这个参数在消息首次发送时是`false`。注意，一个消费者可能会收到上一次被其他消费者收到过的消息。

## 消费者常见错误：重复确认和未知标签 ##

如果消费者对某个传递标签重复确认，会导致RabbitMQ报通道异常:`PRECONDITION_FAILED - unknown delivery tag 100`。如果使用不存在的传递标签页会报相同异常。

其他会报`unknown delivery tag`的场景是，确认消息的通道与接收消息的通道不同。谨记，消息传递确认和消息传递必须是同一通道。

## 通道预取设置(QoS) ##

因为消息是异步发送（推送）到客户端的，所以在任何给定时刻，通常有一个以上的消息“正在运行”。另外，来自客户端的手动确认本质上也本质上是异步的。因此，存在一个未确认的传递标签滑动窗口。开发人员通常会希望限制此窗口的大小，以避免在用户端出现无限制的缓冲区问题。消息协议通过使用`basic.qos`方法设置“预取计数”值来实现 。该值定义通道上允许的未确认交付的最大数量。一旦数量达到配置的数量，RabbitMQ将停止在通道上传递更多消息，除非已确认至少一个未处理的消息。

例如，假设在通道`Ch`上有未确认的传递标签5、6、7和8，并且通道 `Ch`的预取计数设置为4，RabbitMQ将不会在`Ch`上传递任何消息，除非至少有一个未完成的传递被确认。当消费者在通道`Ch`上发送确认，`deliveryTag`设置为5，RabbitMQ会注意到并再传递一条消息。

#### 预取计数对消息吞吐量的影响 ####

通常，增加预取将提高向消费者传递消息的速度，但传递但尚未处理的消息的数量也会增加，从而增加了消费者的RAM消耗。找到合适的预取值是一个反复试验的问题，并且会因工作负载而异。100到300范围内的值通常可提供最佳吞吐量，并且不会带来压倒消费者的巨大风险。更高的取值经常会碰到[收益递减的规律][Link 1]。

## 参考资料 ##

[Consumer Acknowledgements and Publisher Confirms][]

[Publisher Confirms]: https://blog.csdn.net/Lamb_IT/article/details/102649484
[Link 1]: https://www.rabbitmq.com/blog/2014/04/14/finding-bottlenecks-with-rabbitmq-3-3/
[Consumer Acknowledgements and Publisher Confirms]: https://www.rabbitmq.com/confirms.html#automatic-requeueing