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

ゝ一世哀愁。 2022-05-30 06:23 174阅读 0赞

## 1. 概述 ##

在消息代理（如RabbitMQ），由于发送消息的AMQP协议方法不能保证消息一定到达对方或被成功处理，所以发布者和消费者都需要一个交付和处理确认的机制。 在上一篇文章[中间件系列十 RabbitMQ之消费者端的消息确认机制][RabbitMQ]，我们介绍了消费端的消息确认机制。本篇我们介绍发送端的消息确认机制

RabbitMQ在收到消息后，还需要有一段时间才能将消息存入磁盘之中。RabbitMQ并不是为每条消息都做fsync的处理，可能仅仅保存到cache中而不是物理磁盘上，在这段时间内RabbitMQ broker发生crash, 消息保存到cache但是还没来得及落盘，那么这些消息将会丢失。为了解决这个问题RabbitMQ引入发送端消息确认机制，主要通过事务和publisher Confirm机制。

本篇的主要内容如下：  
1. 通过AMQP事务和publisher Confirm机制保证发送端的消息不丢失  
2. 演示RabbitMQ中的事务用法，并通过抓包分析协议，最后说明事务的事务的优点和缺点  
3. 演示RabbitMQ中Publisher Confirm模式的用法，并对以下3种方式通过抓包分析协议并说明其缺点和优点

-  同步方式的发送端的单个Publisher Confirm模式       
    -  同步方式的发送端的批量Publisher Confirm机制 
    -  异步方式的发送端的Publisher Confirm机制

4. 其他发送者端的注意事项: 否定确认、消息确认的时机、持久化消息的确认延迟、发送顺序

## 2. 事务机制保证消息不丢失 ##

RabbitMQ支持事务(transaction)，通过调用tx.select方法开启事务模式。当开启了事务模式后，只有当一个消息被所有的镜像队列保存完毕后，RabbitMQ才会调用tx.commit-ok返回给客户端。

### 2.1. 代码 ###

工程名称：[rabbitmq][]  
发送端的关键代码[TransactionalSend][]：通过channel.txSelect()开启事务，发送消息，最后执行channel.txCommit()提交事务。如果发送失败，则使用channel.txRollback()回滚事务

// 开启事务
    channel.txSelect();
    // 发送消息
    while(num-- > 0) {
        // 发送一个持久化消息到特定的交换机
        channel.basicPublish(EXCHANGE_NAME, routingKey, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes("UTF-8"));
        System.out.println(" [TransactionalSend] Sent + [" + num + "] '" + message + "'");
    }
    
    // 不注解下面语句，可以进入channel.txRollback()逻辑
    // if(true){ 
    // throw new IOException("consumer channel.txRollback() ");
    // }
    // 提交事务
    channel.txCommit();
    }catch(IOException e){
    e.printStackTrace();
    // 回滚事务
    channel.txRollback();
    }

测试代码：使用[PublisherConfirmTest][]的方法进行测试

### 2.2. 使用wireshark抓包分析正常的事务提交流程 ###

使用wireshark截获以上测试时产生的包：  
所有的包：  
![这里写图片描述][SouthEast]

全部包如下：  
**496-497帧**：开启事务  
**496帧**：Tx.Select 客户端向RabbitMQ要求开启事务  
![这里写图片描述][SouthEast 1]  
**497帧**： Tx.Select-Ok 服务端处理事务请求并返回成功结果  
![这里写图片描述][SouthEast 2]  
**498帧**：发送消息，和普通发送消息同，这里略

**499-502帧**：提交事务  
**499帧**：Tx.Commit 客户端向RabbitMQ要求开启事务  
![这里写图片描述][SouthEast 3]  
**502帧**：Tx.Commit-Ok 服务端处理事务提交并返回成功结果  
![这里写图片描述][SouthEast 4]

### 2.3. 使用wireshark抓包分析执行事务失败，自动回滚 ###

使用wireshark截获以上测试时产生的包：  
如果事务执行失败，则会执行tx.rollback执行回滚操作  
由于包信息比较简单，这里略，只上截图  
![这里写图片描述][SouthEast 5]

### 2.4. 事务的优点和缺点 ###

事务的实现简单，能够保证消息正确到达RabbitMQ，但是它的效率低，只有一般发送消息的效率的1/250

## 3. publisher confirms机制保证消息不丢失 ##

在AMPQ-0-9-1中，有定义从消费者到RabbitMQ的处理确认机制。但是没有定义消息代理到生产者的确认机制，在RabbitMQ中对此进行扩展，叫做publisher confirms机制

在标准的AMQP 0-9-1，保证消息不会丢失的唯一方法是使用事务：在通道上开启事务，发布消息，提交事务。但是事务是非常重量级的，它使得RabbitMQ的吞吐量降低250倍。为了解决这个问题，RabbitMQ引入的Publisher Confirms机制，它是模仿AMQP协议中消费者消息确认机制

生产者端可以通过confirm.select来启用方法Publisher Confirms机制，RabbitMQ服务端根据是否设置no-wait的值，返回confirm.select-ok。一旦在通道上使用confirm.select方法，就认为它处于Publisher Confirms模式。事务通道不能进入Publisher Confirms模式**，一旦通道处于Publisher Confirms模式，不能开启事务。即事务和Publisher Confirms模式只能二选一**。

Publisher Confirm模式有以下几种使用方式：

1.  同步方式的发送端的单个Publisher Confirm模式
2.  同步方式的发送端的批量Publisher Confirm机制
3.  异步方式的发送端的Publisher Confirm机制

## 4. 同步方式的发送端的Publisher Confirm模式 ##

### 4.1. 测试代码 ###

测试工程名称：[rabbitmq][]

关键代码:　[SimpleConfirmSend][]  
这个代码实现发送者端发送一个持久化消息到特定的交换机，然后等待服务端返回Basic.Ack后，才执行发送消息

while(num-- > 0) {
        // 发送一个持久化消息到特定的交换机
        channel.basicPublish(EXCHANGE_NAME, routingKey, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes("UTF-8"));
        System.out.println(" [SimpleConfirmSend] Sent '" + message + "'");
        // 等待服务端返回Basic.Ack后，才执行下一个循环
        if(!channel.waitForConfirms()){
            System.out.println("message haven't arrived broker");
              // 在这里可以对发送失败的记录进行处理：如重发
        }
    }

### 4.2. 使用wireshark抓包单个确认消息确认 ###

这里我们发送端只发送一个消息，并进行抓包，分析抓到的包，详细如下：  
![这里写图片描述][SouthEast 6]

和普通发送最大的不同是，在执行发送消息前执行Confirm.Select，RabbitMQ在消息已经收到并处理完毕(如果消息需要，则持久化消息后，才返回Basic.Ok; 如果对应消息的镜像队列，则队列完全同步后，才返回Basic.Ok。总之，必须保证消息不会因为RabbitMQ异常丢失)后返回Basic.Ok给客户端

**110 -111 帧**：  
110帧 Confirm.Select: 客户端请求开启Confirm模式  
![这里写图片描述][SouthEast 7]  
111帧 Confirm.Select-Ok：服务端执行完毕，并返回成功结果  
![这里写图片描述][SouthEast 8]

**112 帧**：发送消息  
这里我们发现，这个包里没有Delivery-Tag值，但是后面RabbitMQ回送的消息包里有Delivery-Tag。如果我们一次发送大量的消息，即使RabbitMQ对收到前N条消息进行确认，发送者端也不知道RabbitMQ是不是收到发送者端最先发送的N条记录，因为消息到达RabbbitMQ是无序的。只有RabbitMQ对本次发送的所有记录进行确认，我们才知道消息全部发送成功。如果不小心丢失一条消息，我们是不知道那条记录丢失的，唯一的办法是本次所有记录重发  
![这里写图片描述][SouthEast 9]  
**113帧**：Basic.Ack  
113帧 RabbitMQ服务端在处理完毕消息后，回送Basic.Ack。这里表示服务端已经收到delivery-Tag=1的数据  
![这里写图片描述][SouthEast 10]

### 4.3. 使用wireshark抓包发送多个消息抓包分析 ###

这里我们发送端发送N个消息，并进行抓包，分析抓到的包，详细如下：  
![这里写图片描述][SouthEast 11]

这里和发送单个消息类似，都是Basic.Publish和Basic.Ack对。最大的不同是Basic.Ack消息中Delivery-Tag值随着消息数量递增  
![这里写图片描述][SouthEast 12]

## 5. 同步方式的发送端的批量Publisher Confirm机制 ##

之前我们是发送一个消息确认一个消息。本节我们发送N条记录后，再进行确认  
测试工程名称：[rabbitmq][]

关键代码:　  
[TransactionalSend][]：

// ======== 批量确认模式 end ======
    // 发送消息
    while(num-- > 0) {
        // 发送一个持久化消息到特定的交换机
        channel.basicPublish(EXCHANGE_NAME, routingKey, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes("UTF-8"));
        System.out.println(" [SimpleConfirmSend] Sent '" + message + "'");
    }
    // 批量等待确认: 返回true: 如果所有的消息都收到有确认应答，没有消息被拒绝
    if(!channel.waitForConfirms()){
        System.out.println("Not all message have arrived broker");
        // 实际应用中，需要在这是添加发送消息失败的处理逻辑：如重发等等
        // 在这种的模式中，如果发送N条消息，如果有一条失败，则所有的消息都需要重新推送
    }

### 5.1. 使用wireshark抓包分析批量确认机制 ###

这里我们发送端连续发送10个消息，然后再basic.Ack进行确认，再进行抓包，分析抓到的包，详细如下  
![这里写图片描述][SouthEast 13]

**79 - 89帧** ：连续发送10个请求  
**91-92帧**：对消息进行应答，虽然有10个消息，但是为了效率RabbitMQ不是对每个包发送一个确认包，而是进行批量确认  
91帧：  
Basic.ACK对收到的前5条已经处理完毕的消息进行确认  
![这里写图片描述][SouthEast 14]  
Basic.Ack对收到的前10条已经处理完毕的消息进行确认  
![这里写图片描述][SouthEast 15]

**备注：**  
这里我们发现，Basic.Ack的Delivery-Tag的值是RabbitMQ服务端生成的，客户端在发送消息的没有这个值，这里不能保证Delivery-Tag值为1，是对客户端发送的第一条记录进行确认。如91帧是RabbitMQ对最新处理完毕的前5条记录进行应答，但是不能保证100%是对客户端发送的前5条记录进行应答。RabbitMQ可能先处理完1，2，3，4，6条记录，再发送这条确认信息。  
所以假如我们第二条的Delivery-Tag值为8且以后不在收到新的Basic.ACK，那么表示2条消息丢失，但是我们无法确认100%确认丢失的消息是第9，10条，此时我们需要重发这里的全部10条记录

## 6. 异步Publisher Confirm机制 ##

测试工程名称：[rabbitmq][]

关键代码[AsynConfirmSend][]：通过回调方法进行处理

// 添加回调对象，处理返回值
    channel.addConfirmListener(new ConfirmListener(){
        @Override
        public void handleAck(long deliveryTag, boolean multiple) throws IOException {
            System.out.println("[AsynConfirmSend] handleAck : deliveryTag = " + deliveryTag + " multiple = " + multiple);
        }
        @Override
        public void handleNack(long deliveryTag, boolean multiple) throws IOException {
            System.out.println("[AsynConfirmSend] handleNack : deliveryTag = " + deliveryTag + " multiple = " + multiple);
        }
    });
    
    // 开启confirm模式：
    channel.confirmSelect();
    // 发送消息
    while(num-- > 0) {
        // 发送一个持久化消息到特定的交换机
        channel.basicPublish(EXCHANGE_NAME, routingKey, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes("UTF-8"));
        System.out.println(" [TransactionalSend] Sent '" + message + "'");
    }

这里和上面的同步批量confirm的最大不同时，发送端在发送消息时，另一个线程同步进行消息的确认。使用此接口连续发送10个消息，包的信令和同步批量confirm是几乎相同的，在信令上两者没有本质区别。  
![这里写图片描述][SouthEast 16]

## 7. 使用Publisher Confirms的其它注意事项 ##

### 7.1. 否定确认和重新入队 ###

当RabbitMQ无法成功的处理消息时，它会返回生产者端basic.nack，而不是basic.ack。在这种情况下， basic.nack的字段与basic.ack相对应的字段意义相同，且requeue 字段是没有意义的。是否重发消息由发送者端自己决定。  
当进入通道进入Publisher Confirms模式，所有的消息只能被confirmed确认或者nack一次，每个消息只能同时被confirmed和nack。另外没有机制保证消息需要多久被confirmed。  
basic.nack只有Erlange进程在处理队列时发生内部错误时才会被回送。  
当消息需要重新入队重发时，如果可能的话，它将被放置在其原来的队列中的原来位置。如果不能（由于当多个消费者共享一个队列时，其他消费者的也会并发递送和确认消息），则该消息将被重新排队到靠近排队头的位置。

### 7.2. 消息确认的时机 ###

对于不可路由的消息(即RabbitMQ发现收到的消息不能被路由到任务队列)，有两种情况：

*  a. 消息的mandatory=false，RabbitMQ发现消息不可路由后，马上确认消息，即发送basic.ack或baisc.nack给生产者
 *  b. 消息的mandatory=true，RabbitMQ发现消息可路由后，先发送basic.return，再确认消息，即发送basic.ack或baisc.nack给生产者

对于可路由的消息，需要同时满足如下所有的条件才可以回送确认消息

*  1 消息被路由到所有的队列中
 *  2 对于路由到持久队列的持久消息，需要持久化消息到磁盘
 *  3 如果队列是镜像队列，则需要将消息同步到所有的队列中

### 7.3. 持久化消息的确认延迟 ###

持久化的消息路由到持久化队列时，RabbitMQ会将消息存储到磁盘空间。为了保证持久化效率，RabbitMQ不是来一条存一条，而是定时批量地持久化消息到磁盘，这个时间间隔通常是几百毫秒，或者队列空闲执行消息持久化。如果队列支持镜像队列，则延迟时间更大。如果生产者每发送一条消息，等待basic.ack来了再发送一条消息，则等待时间可以达到几百毫秒。  
为了提高吞吐量，RabbitMQ强烈建议应用程序异步处理确认或批量发送消息后再等待未完成的确认

### 7.4. Publisher Confirms的顺序 ###

在大多数情况下，RabbitMQ将以发布的顺序向发布者确认消息（这适用于发布在单个通道上的消息）。 但是，发布者确认是异步发送的，并且可以确认一条消息或一组消息。 发出确认的确切时间取决于消息的传递模式（持久性与非持久性）以及消息被路由到的队列的属性。所以RabbitMQ可能不以发布的顺序向发布者发送确认消息。生产者端尽量不要依赖消息确认的顺序做服务

## 8. 代码 ##

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

[RabbitMQ]: http://blog.csdn.net/hry2015/article/details/79416540
[rabbitmq]: https://github.com/hryou0922/spring_boot/tree/v0.16/rabbitmq/src/main/java/com/hry/spring/rabbitmq/advanced/publisherconfirm
[TransactionalSend]: https://github.com/hryou0922/spring_boot/blob/v0.16/rabbitmq/src/main/java/com/hry/spring/rabbitmq/advanced/publisherconfirm/TransactionalSend.java
[PublisherConfirmTest]: https://github.com/hryou0922/spring_boot/blob/v0.16/rabbitmq/src/test/java/com/hry/spring/rabbitmq/advanced/PublisherConfirmTest.java
[SouthEast]: /images/20220530/3abb53f3506440ef826d43c5cd5e06d8.png
[SouthEast 1]: /images/20220530/ef284c884d2847d4addc6f8638779498.png
[SouthEast 2]: /images/20220530/72a289069807432995b9688c6798dc76.png
[SouthEast 3]: /images/20220530/51dbdf66b25649a9bd2aaba4a29d2f76.png
[SouthEast 4]: /images/20220530/0c1075db4c02432a86b0d978ffe31d69.png
[SouthEast 5]: /images/20220530/1ec78662b3d143ccb394d08caf8a8d11.png
[SimpleConfirmSend]: https://github.com/hryou0922/spring_boot/blob/v0.16/rabbitmq/src/main/java/com/hry/spring/rabbitmq/advanced/publisherconfirm/SimpleConfirmSend.java
[SouthEast 6]: /images/20220530/9758cfbd82804916bde1f98085031745.png
[SouthEast 7]: /images/20220530/089163cc612b434c9bc70907a22701df.png
[SouthEast 8]: /images/20220530/bc0b0833297d42dd89bce09d56ab74d5.png
[SouthEast 9]: /images/20220530/5964b0f4fdca427c92b4c88b9ec7254e.png
[SouthEast 10]: /images/20220530/4054eaa14a5b4bd381fcc50f205b42f7.png
[SouthEast 11]: /images/20220530/4a658afe67a34d3daa695b53e52191ce.png
[SouthEast 12]: /images/20220530/e2bf456d509446228b5bd535d7aacac5.png
[SouthEast 13]: /images/20220530/8f8621b06e524a1fa173bba6d2aba32d.png
[SouthEast 14]: /images/20220530/47990c0f2686461990647ada5b8f5321.png
[SouthEast 15]: /images/20220530/de0da48d75ab4c3c9254c5d0f6e89ae9.png
[AsynConfirmSend]: https://github.com/hryou0922/spring_boot/blob/v0.16/rabbitmq/src/main/java/com/hry/spring/rabbitmq/advanced/publisherconfirm/AsynConfirmSend.java
[SouthEast 16]: /images/20220530/f16e10e9994a413a900a3ef9b0933d22.png