RabbitMQ实战-消费端ACK、NACK及重回队列机制

当连接失败时，消息可能还在客户端和服务器之间传输 - 它们可能处于两侧的解码或编码的中间过程，在 TCP 堆栈缓冲区中，或在电线上飞行。 在这种情况下，传输中的信息将无法正常投递 - 它们需要被重新投递。Acknowledgements机制让服务器和客户端知道何时需要重新投递。

根据定义，使用消息代理（如RabbitMQ）的系统是分布式的。由于发送的协议方法（消息）不能保证到达协作方或由其成功处理，因此发布者和消费者都需要一个投递和处理确认的机制。

• 从消费者到 RabbitMQ 的递送处理确认，在消息协议中称为acknowledgements
• broker对publishers的确认是一个协议扩展，称为publisher confirms

这两个功能都启发于 TCP。它们对于从publish到MQ节点和从MQ节点到消费者的可靠投递都至关重要。即对于数据安全至关重要，应用程序对数据安全的责任与MQ节点一样多。

当 RabbitMQ 向 Con 传递消息时，它需要知道何时考虑该消息才能成功发送。什么样的逻辑是最佳的取决于系统。因此，它主要是应用决定的。在 AMQP 0-9-1 中，当 Con：

• 使用basicConsume方法进行注册

• 或使用basicGet 方法按需获取消息

就会进行。

Consumer Acknowledgement Modes and Data Safety Considerations

当节点向消费者传递消息时，它必须决定该消息是否应由消费者考虑处理（或至少接收）。由于多种内容（客户端连接、消费者应用等）可能会失败，因此此决定是数据安全问题。消息传递协议通常提供一个确认机制，允许消费者确认交付到他们连接到的节点。是否使用该机制由消费者订阅时决定。根据使用的确认模式，RabbitMQ 可以考虑在消息发出后立即成功传递（写入 TCP 插座）或收到明确（‘手册’）客户确认时。手动发送的确认可能是正面的或负面的，并使用以下协议方法之一：

basic.ack

积极地确认

basic.nack

消极地确认。

basicReject

消极地确认，但还有一个limitation

Delivery Identifiers: Delivery Tags

如何确定投递（确认表明他们各自的投递）。当一个 Con（订阅）被注册，MQ将使用basic.deliver方法发送（推送）消息。该方法带有delivery tag，该tag可唯一标识channel上的投递。因此，Delivery tags作用域在每个 channel 内。交

Delivery Tags是单调增长的正整数，并由客户库提供。客户端库方法，承认交付以交付标签作为参数。由于每个通道的递送标签范围很广，因此必须在接收的同一通道上确认交付。在不同的通道上确认将导致’未知交货标签’协议异常并关闭通道。

Positively Acknowledging Deliveries

用于交付确认的 API 方法通常暴露为客户库中通道上的操作。Java 客户端用户将使用channel：

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

Acknowledging Multiple Deliveries at Once

Manual确认模式可批量进行，以减少网络流量。这是通过将acknowledgement方法的multiple字段设置为true来实现的。

basicReject在历史上都没有该字段，这就是为什么basicNack被MQ引入，作为协议的扩展。

当multiple=true，MQ 将确认所有未完成的delivery tag，并包括确认中指定的tag。 与确认相关的其他所有内容一样，这个作用域是channel内。比如，假定channel Ch 上有未确认的delivery tag 5、6、7 和 8，当一个delivery tag=8、multiple=true的acknowledgement frame到达该channel时，则从 5 到 8 的所有投递都将被确认。 若multiple=false，则仍不确认投递 5、6 和 7。

要确认与MQ 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);
         }
     });

Negative Acknowledgement and Requeuing of Deliveries

有时，消费者无法及时处理投递，但其他实例可能能够处理。这时可能更想让它重新入队，让其他Con接收和处理它。 basicReject和basicNack就是用于实现这种想法的两个协议方法。这些方法通常用于消极地确认投递。

此类投递可被Broker丢弃或重新入队。此行为由requeue字段控制：

• 当字段设置为true，Broker将用指定的delivery tag重新入队投递（或多个投递）。

这两个方法通常暴露作为客户端库中channel上的操作。Java 客户端用户可以调用：

• Channel#basicReject
• Channel#basicNack

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);
         }
     });

RabbitMQ ACK 机制的意义

ACK机制可以保证Con拉取到了消息，若处理失败了，则队列中还有这个消息，仍然可以给Con处理。

ack机制是 Con 告诉 Broker 当前消息是否成功消费，至于 Broker 如何处理 NACK，取决于 Con 是否设置了 requeue：如果 requeue=false， 则NACK 后 Broker 还是会删除消息的。

但一般处理消息失败都是因为代码逻辑出bug，即使队列中后来仍然保留该消息，然后再给Con消费，依旧报错。 当然，若一台机器宕机，消息还有，还可以给另外机器消费，这种情景下 ACK 很有用。

如果不使用 ACK 机制，直接把出错消息存库，便于日后查bug或重新执行。 参考 Quartz 定时任务调度，Quartz可以让失败的任务重新执行一次，或者不管，或者怎么怎么样，但是 RabbitMQ 好像缺了这一点。

1 ACK和NACK

当设置autoACK=false 时，就可以使用手工ACK。 其实手工方式包括了手工ACK、手工NACK。

• 手工 ACK 时，会发送给Broker一个应答，代表消息处理成功，Broker就可回送响应给Pro
• NACK 则表示消息处理失败，如果设置了重回队列，Broker端就会将没有成功处理的消息重新发送

使用方式

Con消费时，若由于业务异常，可手工 NACK 记录日志，然后进行补偿

void basicNack(long deliveryTag, 
			   boolean multiple,
			   boolean requeue)

如果由于服务器宕机等严重问题，就需要手工 ACK 保障Con消费成功

void basicAck(long deliveryTag, boolean multiple)

2 消费端的重回队列

重回队列针对没有处理成功的消息，将消息重新投递给Broker。 重回队列会把消费失败的消息重新添加到队列尾端，供Con重新消费。 一般在实际应用中，都会关闭重回队列，即设置为false。

3 代码实战

• Con，关闭自动签收功能

• 自定义Con，对第一条消息(序号0)进行NACK，并设置重回队列

• Pro 对消息设置序号,以便区分

• 启动Con，查看管控台

• 启动Pro,这里第一条消息由于我们调用了NACK，并且设置了重回队列，所以会导致该条消息一直重复发送，消费端就会一直循环消费