中间件系列十 RabbitMQ之消费者端的消息确认机制

￡神魔★判官ぃ 2022-05-30 02:06 321阅读 0赞

## 概述 ##

在RabbitMQ中，即使将queue，exchange, message等都设置了持久化之后，还是不能保证100%保证数据不丢失了。为了实现消息不丢失，我们需要从Consumer端和Productor端同时进行处理。本篇文章先介绍Consumer端，在AMPQ-0-9-1中有定义从消费者到RabbitMQ的消息确认机制，通过此机制可以保证消息能够从RabbitMQ正确到达消费者端。本文介绍在RabbitMQ中如何实现消费者端的消息确认机制，包括如下内容

*  1 消费者的实现机制
 *  2 消费者端的代码实现
 *  3 使用wireshark对消息确认的关键包进行转包，并进行分析
 *  4 在使用消息确认机制的注意点

## 消费者端投递确认机制 ##

### 在消费者端确认的方式 ###

RabbitMQ中的两种确认方式：

*  1 自动确认方式：RabbitMQ成功将消息发出（即将消息成功写入TCP Socket）中立即认为本次投递已经被正确处理，不管消费者端是否成功处理本次投递
    
    > 在自动确认模式下，消息发送后即被认为成功投递，又称为”fire-and-forget”  
    > **优点**：这种模式下吞吐量非常高。  
    > **缺点**：A. 有可能出现投递丢失的情况，不同于手动确认模式，如果消费者的TCP连接或通道在消息成功交互之前关闭，则此消息会丢失 B. 消费者端过载的问题。在手动确认模式中，可以设置一次最多同时处理多少消息，而自动模式不能设置此值。因此，消费者有可能因为消息无法及时处理，堆积中内存中，内存耗尽而奔溃 C. 此种模式只推荐在消费者可以快速且稳定处理投递的消息的场景中使用
 *  2 手动处理方式：消费者收到消息后，手动调用basic.ack/basic.nack/basic.reject后，RabbitMQ收到这些消息后，才认为本次投递成功
    
    > 手动消息确认方法有：  
    > § **basic.ack**用于肯定确认  
    > § **basic.nack**用于否定确认（注意：这是AMQP 0-9-1的RabbitMQ扩展）  
    > § **basic.reject**用于否定确认，但与basic.nack相比有一个限制:一次只能拒绝单条消息  
    > 消费者端以上的3个方法都表示消息已经被正确投递，但是basic.ack表示消息已经被正确处理，但是basic.nack,basic.reject表示没有被正确处理，但是RabbitMQ中仍然需要删除这条消息。  
    > 手动的确认模式的投递效率略低于自动，但是可以弥补自动确认模式的不足。

### 批量手动投递确认 ###

消息手动除了一次确认一条，也可以一次确认多条。为了减少网络流量，可以批量手动确认。在应答时，设置basic.nack的multiple 字段为true，可以同时对delivery\_tag和比delivery\_tag值小的投递消息进行确认  
例如，假设在通道上没有确认消息的delivery\_tag是5,6,7和8，当basic.nack中delivery\_tag被设置为8并且multiple 被设置为true时，方法执行成功后，从5到8的所有消息将被确认。 如果multiple 设置为false，那么交货5,6和7仍然是未确认的。

### 投递唯一码: Delivery Tags ###

当消费者向RabbitMQ注册后，RabbitMQ使用basic.deliver向消费者投递消息时，消息体上会带上delivery tag，这个值会唯一标识本次投递，在同一通道上，此值是唯一的。delivery tag值有64位长度，值从1开始，每发送一次消息值递增1，最大值为9223372036854775807。消费者端在应答消息时，带上此参数，告诉RabbitMQ某次投递已经正确应答。

## 消费者端消息投递确认代码 ##

### 工程： ###

工程：  
代码路径：

### 消费者代码： ###

代码：关键点  
a. channel.basicConsume设置接收非自动确认  
b. 在处理完消息后，调用channel.basicAck进行手动消息确认

// 默认消费者实现
    Consumer consumer = new DefaultConsumer(channel) {
        @Override
        public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body)
                throws IOException {
            String message = new String(body, "UTF-8");
            System.out.println(" [ConsumerConfirmRecv] Received '" + message + "'");
            // 消息正向确认
            channel.basicAck(envelope.getDeliveryTag(),true);
             // 消息否定确认: 如果设置multiple=false，requeue值启作用，如果设置multiple=true，则requeue无论设置什么值，后台统一处理成true
            // channel.basicNack(envelope.getDeliveryTag(),false, false);
        }
    };
    // 接收消息：设置非自动确认
    channel.basicConsume(QUEUE_NAME, false, consumer);

## 消费者端抓包分析 ##

执行以上的代码，并进行抓包分析其中关键包。和一般的消息包不同点时，一般的消费者端的包分析见[中间件系列九 RabbmtiMQ 通过wireshark抓包学习AMQP协议][RabbmtiMQ _wireshark_AMQP]

### 正向确认-Basic.ack ###

和一般的消息包不同点时，一般的消费者端的包分析见本文，只是多了Basic.ack包  
![这里写图片描述][SouthEast]  
131-132帧 RabbitMQ向消费者推送消息，消费者端进行否定确认  
131帧 RabbitMQ向消费者推送消息

Frame 131: 269 bytes on wire (2152 bits), 269 bytes captured (2152 bits) on interface 0
    Ethernet II, Src: PcsCompu_48:72:ad (08:00:27:48:72:ad), Dst: Giga-Byt_bf:ae:ce (40:8d:5c:bf:ae:ce)
    Internet Protocol Version 4, Src: 10.240.80.147, Dst: 10.240.80.99
    Transmission Control Protocol, Src Port: 5672, Dst Port: 49877, Seq: 600, Ack: 642, Len: 215
    # 返回Consume的调用结果
    Advanced Message Queueing Protocol
        Type: Method (1) Channel: 1
        Length: 36
        Class: Basic (60)
        Method: Consume-Ok (21)
        Arguments
        # 此消息的消费者的编号
            Consumer-Tag: amq.ctag-AhhpMhEMC5VuWcny42rObg
    # 要投递的消息参数
    Advanced Message Queueing Protocol
        Type: Method (1) Channel: 1
        Length: 87
        Class: Basic (60)
        Method: Deliver (60)
        Arguments
        # 要投递的消息者标志
            Consumer-Tag: amq.ctag-AhhpMhEMC5VuWcny42rObg
        # 消息的编号，从1开始
            Delivery-Tag: 1        
        # 是否是重新投递的消息，如果消息是第一次投递，则此值是false，如果此消息是重新投递的，则此值为true
            .... ...0 = Redelivered: False
        # 来源的交换机
            Exchange: consumerconfirm-exchange
        # 路由键
            Routing-Key: consumer-confirm
    # 要投递的消息的头
    Advanced Message Queueing Protocol
        Type: Content header (2)
        Channel: 1
        Length: 27
        Class ID: Basic (60)
        Weight: 0
        Body size: 33
        Property flags: 0x9800
        # 消息的属性 
        Properties
            Content-Type: text/plain
            Delivery-Mode: 2
            Priority: 0
    # 要投递的消息的内容
    Advanced Message Queueing Protocol
        Type: Content body (3)
        Channel: 1
        Length: 33
        Payload: 436f6e73756d6572436f6e6669726d53656e642131353137...

132帧 消费者端对消息的确认  
![这里写图片描述][SouthEast 1]

### 否定确认-Basic.Nack ###

和一般的消息包不同点时，一般的消费者端的包分析见本文，只是多了Basic.Nack包  
![这里写图片描述][SouthEast 2]

**275-276帧** RabbitMQ向消费者推送消息，消费者端进行否定确认  
**275帧** RabbitMQ向消费者推送消息，和上节内容相似，这里略  
**276帧** 消费者端进行否定确认，比basic.ack多了Requeue属性  
![这里写图片描述][SouthEast 3]

## 其他消费者端的为了保证正确处理数据的机制 ##

在消费者端为了正确处理数据，还可以设置Qos和投递失败

### 通道预取设置（Channel Prefetch Setting (QoS)） ###

只能在消息手动确认模式中启作用。  
为了避免消费者端一次同时处理过多的消息，可以通过basic.qos设置最大的预取值。该值定义了通道上允许的最大未确认消息，一旦未确认消息的数量达到配置值，RabbitMQ将停止在通道上传送更多消息，直到至少有一个未被确认的消息被确认。  
**备注**：通道预取设置在basic.get (“pull API”)中是不启作用，即使在消息手动确认模式中

### 消费者确认模式，预取和吞吐量 ###

在RabbitMQ中影响吞吐量最大的参数是：消息确认模式和Qos预取值  
自动消息确认模式或设置Qos预取值为无限虽然可以最大的提高消息的投递速度，但是在消费者端未及时处理的消息的数量也将增加，从而增加消费者RAM消耗，使用消费者端奔溃。所以以上两种情况需要谨慎使用。

RabbitMQ官方推荐Qos预取值设置在 100到300范围内的值通常提供最佳的吞吐量，并且不会有使消费者奔溃的问题

### 消费者失败或失去连接时：自动重新排队 ###

在消息手动确认模式中，如果发生以下情况投递消息所有的通道或连接被突然关闭（包括消费者端丢失TCP连接、消费者应用程序（进程）挂掉、通道级别的协议异常）任何已经投递的消息但是没有被消费者端确认的消息会自动重新排队。  
请注意，连接检测不可用客户端需要一段时间才会发现，所以会有一段时间内的所有消息会重新投递  
因为消息的可能重新投递，所有必须保证消费者端的接口的幂等。

### 多次确认和对未知delivery\_tag进行确认 ###

如果消费者对同一delivery\_tag进行多次确认，则抛出通道异常PRECONDITION\_FAILED。如果对  
未知的delivery\_tag进行确认，也会抛出通道异常。

## 代码 ##

所有的详细代码见github代码，请尽量使用[tag v0.15，不要使用master，因为master一直在变，不能保证文章中代码和github上的代码一直相同][tag v0.15_master_master_github]

[RabbmtiMQ _wireshark_AMQP]: http://blog.csdn.net/hry2015/article/details/79392587
[SouthEast]: /images/20220530/44dcdd8c2e4144b894bcba8797f576cd.png
[SouthEast 1]: /images/20220530/2752e155fd314a1dab9b95d733252d1a.png
[SouthEast 2]: /images/20220530/7c795e87db5845e5877698823803ed37.png
[SouthEast 3]: /images/20220530/790668343b804a75932e3835469253c5.png
[tag v0.15_master_master_github]: https://github.com/hryou0922/spring_boot/tree/v0.15/rabbitmq/src/main/java/com/hry/spring/rabbitmq/advanced/consumerconfirm