RabbitMQ学习(十七)：消息确认之消费者确认模式 II

╰半夏微凉° 2021-11-11 04:48 408阅读 0赞

# 说明 #

在上篇博文[《RabbitMQ学习(十六)：消息确认之消费者确认模式 I》][RabbitMQ_ I]中对消息确认的必要性和确认相关的传输标签，消费者确认方法等内容进行了翻译学习，本篇博文我将继续翻译学习消息者确认模式的剩余内容，主要包含了积极消极确认的方式，批量确认消息，通道Prefetch值的设置以及它和确认模式对吞吐量的影响，还有异常时的自动重新入队，客户端常见错误等内容。

# 正文 #

## 积极确认 ##

在客户端包中对消息进行确认的api方法通常被当作为是通道上的操作来提供。java客户端使用Channel\#basicAck和Channel\#basicNack方法来分别代表basic.ack和basic.nack方法。以下是使用java客户端进行积极确认的示例：

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();
                 // positively acknowledge a single delivery, the message will
                 // be discarded
                 channel.basicAck(deliveryTag, false);
             }
         });

--------------------

## 一次确定多个消息 ##

手动确认的方式可以进行批量确认来减少网络传输。**批量确认可以通过设置确认方法的multiple参数值为true实现。注意，basic.reject方法没有该参数，这就是为什么RabbitMQ要引进basic.nack方法作为协议的补充。**

当multiple的值设置为true时，RabbitMQ将确认指定传输标签以及之前所有未被确认的消息。与单个确认相同，批量确认的作用域为每个通道。例如：通道Ch上有四个未被确认的消息，标签分别为5，6，7，8；当一个delivery\_tag值为8并且multiple值为true的确认消息到达通道时，所有5到8的标签都会被确认。如果multiple值设置为false，标签为5，6，7的消息将不会被确认。

以下是使用java客户端，将Channel\#basicAck方法的multiple值设置true进行批量确认的示例：

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();
                 // positively acknowledge all deliveries up to
                 // this delivery tag
                 channel.basicAck(deliveryTag, true);
             }
         });

--------------------

## 消息的消极确认和重新入队 ##

有时一个消费者不能立即处理一个发送的消息，当时其他消费者可以，这种情况下就需要将消息重新入队让其他消费者接收处理。**RabbitMQ提供了basic.reject和basic.nack两种协议方法实现消息的重新入队。**

这两个方法一般被用来对一个消息进行消极确认，例如，消息需要被服务器丢弃或者重新入队。**这个操作通过requeue这个参数进行控制，当这个参数被设置为true时，服务器将把指定的传输标签的消息重新入队。**

这些方法与积极确认一样都是通道上的操作，java客户端使用Channel\#basicReject和Channel\#basicNack方法分别代表basic.reject和basic.nack方法。以下是进行丢弃消息和重新入队的示例：

丢弃消息：

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();
                 // negatively acknowledge, the message will
                 // be discarded
                 channel.basicReject(deliveryTag, false);
             }
         });

重新入队：

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();
                 // requeue the delivery
                 channel.basicReject(deliveryTag, true);
             }
         });

**当消息被重新入队时，如果可能消息会被放置在之前在队列的原始位置，如果不能（由于多个消费者共同消费一个队列，会进行同时传输和确认），消息将被尽可能地放到接近队头的位置。**

重新入队的消息可能会被立即发送，但这个取决于它在队列中的位置，及活跃消费者在通道上设置的Prefetch的值。这意味着如果所有的消息者因为某些原因不能处理消息而全部将消息重新入队，这个将导致重新入队和重新发送的循环。这样的循环将不断消耗网络带宽和CPU资源。消费者可以追踪消息的重新发送次数和并合理的拒绝消息（丢弃它们），或者在延迟一段时间后再重新入队。

**可能存在需要一次拒绝或重新入队多个消息的情况，这种情况可以使用basic.nack方法。这就是basic.nack方法和basic.reject的不同之处，该方法有一个可选参数multiple来进行批量操作**。以下是使用java客户端进行消息批量重新入队。

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();
                 // requeue all unacknowledged deliveries up to
                 // this delivery tag
                 channel.basicNack(deliveryTag, true, true);
             }
         });

--------------------

## 通道Prefetch值的设置（QoS） ##

由于消息被发送到客户端是异步的，所以在任何时刻通道中都会有多个消息正在被传递。另外，客户端的异步确认也是异步的。**所以对于通道中的未被确认的传输标签有个滑动窗口，开发者通常设置窗口的大小来避免在消费者出现无边界缓冲区的问题。这个限制可以通过使用basic.qos方法设置一个“prefetch count”值来实现。一旦未被确认的消息数到达设定的值，RabbitMQ将停止往该通道传输更多的消息，除非通道中至少有一个消息被确认。**

例如，在通道Ch中有四个未被确认的消息，传输标签分别为5，6，7，8；并且通道的prefetch值设置为4，这时RabbitMQ将不再往该通道推送更多的消息，直到其中最少一个消息被确认。当一个delivery\_tag值为5（或者是6，7，8）的确认消息到达该通道时，RabbitMQ将注意到未被确认的数量已经降到限制值以下，并且发送消息到该通道。当一次确认多个消息时，将使得更多的消息可以被发送。

我们必须注意到消息的发送和客户端的手动确认完全都是异步的，因此如果在传输过程中prefetch的值发生了改变，这出现自然的竞争条件，此时通道会暂时出现未确认消息的数量大于设定的值。

Qos prefetch值的设置不会对使用basic.get方式获取的消息造成影响，即使是使用手动确认模式。

--------------------

## 对吞吐量的影响 ##

**消息确认模式和Qos prefetch值对消费者的吞吐量有明显的影响。通常，增加prefetch值将提高消息到消息者的发送速率，自动确认模式可以得到最好的发送速率。然而，消息的发送数量和未被确认的消息数量也会同时增长，这些将增加消费者的内存消耗。**

自动确认模式或者手动确认模式在使用时如果没有设置prefetch值时，需要十分小心。消费者消费大量的消息并且没有进行确认时，将消费者所在节点的内存消耗增长。**寻找一个合适的prefetch值时十分重要的，这需要不断的测试，并且该值随着工作负载的不同而变化。一般地，该值在100到300之间可以提供较优的吞吐量并且不会明显增加消费者内存溢出的风险。注意，值越高，收益越低。**

**prefetch值为1是最保守的策略，在特定环境下如消费者连接延迟较高时，会明显地降低吞吐量。对于大多数程序来说，一个较高的值是合适和最优的。**

--------------------

## 自动重新入队 ##

**当使用手动确认时，当在传输时发生了通道或者连接关闭，任何未被确认的消息都会自动重新入队。这包括了客户端的TCP连接丢失，消息者程序的崩溃和通道协议异常的发生。**

注意，发生不可用的客户端需要耗费一段时间。

由于重新入队，消费者必须准备处理重新发送的消息，否则要考虑处理上的幂等性。**重新发送有一个特殊的Boolean属性—redeliver,被RabbitMQ设置为true**。在第一次发送时，该值被设置为false。注意，一个消费者可能会接收到之前被发送到其他消费者的消息。

--------------------

## 客户端错误：双重确认和未知标签 ##

如果客户端多次确认相同的标签，RabbitMQ将导致通道异常，如PRECONDITION\_FAILED - unknown delivery tag 100。同样地，如果使用了一个未知的传输标签，相同的通道异常将被抛出。

服务器因未知标签抛出异常的另一种情况是：**客户端尝试在与接收到的消息不同的通道上进行确认，不管是积极确认还是消极确认。因此，传输必须在同一个通道上确认**。

原文地址：[https://www.rabbitmq.com/confirms.html][https_www.rabbitmq.com_confirms.html]

[RabbitMQ_ I]: https://blog.csdn.net/sinat_36553913/article/details/97525032
[https_www.rabbitmq.com_confirms.html]: https://www.rabbitmq.com/confirms.html