RabbitMQ之消息确认机制（事务、Confirm）

浅浅的花香味﹌ 2023-03-13 02:14 6阅读 0赞

## 概述 ##

在使用RabbitMQ的时候，我们可以通过消息持久化操作来解决因为服务器的异常奔溃导致的消息丢失，除此之外我们还会遇到一个问题，当消息的发布者在将消息发送出去之后，消息到底有没有正确到达broker代理服务器呢？如果不进行特殊配置的话，默认情况下发布操作是不会返回任何信息给生产者的，也就是默认情况下我们的生产者是不知道消息有没有正确到达broker的，如果在消息到达broker之前已经丢失的话，持久化操作也解决不了这个问题，因为消息根本就没到达代理服务器，你怎么进行持久化，那么这个问题该怎么解决呢？

RabbitMQ为我们提供了两种方式：

*  通过AMQP事务机制实现，这也是AMQP协议层面提供的解决方案；
 *  通过将channel设置成confirm模式来实现，即通过发送方确认（publisher confirm）机制实现

## 事务机制 ##

这里首先探讨下RabbitMQ事务机制。

RabbitMQ中与事务机制有关的方法有三个：txSelect(), txCommit()以及txRollback(), txSelect用于将当前channel设置成transaction模式，txCommit用于提交事务，txRollback用于回滚事务，在通过txSelect开启事务之后，我们便可以发布消息给broker代理服务器了，如果txCommit提交成功了，则消息一定到达了broker了，如果在txCommit执行之前broker异常崩溃或者由于其他原因抛出异常，这个时候我们便可以捕获异常通过txRollback回滚事务了。

关键代码：  
![在这里插入图片描述][20200509161221810.png]  
通过wirkshark抓包（ip.addr==xxx.xxx.xxx.xxx && amqp），可以看到：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0lUXzEw_size_16_color_FFFFFF_t_70]  
（注意这里的Tx.Commit与Tx.Commit-Ok之间的时间间隔294ms，由此可见事务还是很耗时的。）

我们先来看看没有事务的通信过程是什么样的：  
![在这里插入图片描述][2020050916130195.png]  
下面我们来看下事务回滚是什么样子的。关键代码如下：  
![在这里插入图片描述][20200509161401140.png]  
同样通过wireshark抓包可以看到：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0lUXzEw_size_16_color_FFFFFF_t_70 1]  
代码中先是发送了消息至broker中但是这时候发生了异常，之后在捕获异常的过程中进行事务回滚。

事务确实能够解决producer与broker之间消息确认的问题，只有消息成功被broker接受，事务提交才能成功，否则我们便可以在捕获异常进行事务回滚操作同时进行消息重发，**但是使用事务机制的话会降低RabbitMQ的性能**，那么有没有更好的方法既能保障producer知道消息已经正确送到，又能基本上不带来性能上的损失呢？从AMQP协议的层面看是没有更好的方法，但是RabbitMQ提供了一个更好的方案，即将channel信道设置成confirm模式。

## 发送方确认机制（Confirm模式） ##

上面我们介绍了RabbitMQ可能会遇到的一个问题，即生成者不知道消息是否真正到达broker，随后通过AMQP协议层面为我们提供了事务机制解决了这个问题，但是采用事务机制实现会降低RabbitMQ的消息吞吐量，那么有没有更加高效的解决方式呢？答案是采用Confirm模式。

#### producer端confirm模式的实现原理 ####

生产者将信道设置成confirm模式，一旦信道进入confirm模式，所有在该信道上面发布的消息都会被指派一个唯一的ID(从1开始)，一旦消息被投递到所有匹配的队列之后，broker就会发送一个确认给生产者（包含消息的唯一ID）,这就使得生产者知道消息已经正确到达目的队列了，如果消息和队列是可持久化的，那么确认消息会将消息写入磁盘之后发出，broker回传给生产者的确认消息中deliver-tag域包含了确认消息的序列号，此外broker也可以设置basic.ack的multiple域，表示到这个序列号之前的所有消息都已经得到了处理。

confirm模式最大的好处在于他是异步的，一旦发布一条消息，生产者应用程序就可以在等信道返回确认的同时继续发送下一条消息，当消息最终得到确认之后，生产者应用便可以通过回调方法来处理该确认消息，如果RabbitMQ因为自身内部错误导致消息丢失，就会发送一条nack消息，生产者应用程序同样可以在回调方法中处理该nack消息。

在channel 被设置成 confirm 模式之后，所有被 publish 的后续消息都将被 confirm（即 ack） 或者被nack一次。但是没有对消息被 confirm 的快慢做任何保证，并且同一条消息不会既被 confirm又被nack 。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0lUXzEw_size_16_color_FFFFFF_t_70 2]  
已经在transaction事务模式的channel是不能再设置成confirm模式的，即这两种模式是不能共存的。

#### 编程模式 ####

对于固定消息体大小和线程数，如果消息持久化，生产者confirm(或者采用事务机制)，消费者ack那么对性能有很大的影响.

消息持久化的优化没有太好方法，用更好的物理存储（SAS, SSD, RAID卡）总会带来改善。生产者confirm这一环节的优化则主要在于客户端程序的优化之上。归纳起来，客户端实现生产者confirm有三种编程方式：

*  **普通confirm模式**：每发送一条消息后，调用waitForConfirms()方法，等待服务器端confirm。实际上是一种串行confirm了。
 *  **批量confirm模式**：每发送一批消息后，调用waitForConfirms()方法，等待服务器端confirm。
 *  **异步confirm模式**：提供一个回调方法，服务端confirm了一条或者多条消息后\* Client端会回调这个方法。

**从编程实现的复杂度上来看：**  
**第1种：普通confirm模式**  
普通confirm模式最简单，publish一条消息后，等待服务器端confirm,如果服务端返回false或者超时时间内未返回，客户端进行消息重传。  
代码如下：

import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.ConnectionFactory;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    public class NormalConfirmProducer { 
        private final static String EXCHANGE_NAME = "normal-confirm-exchange";
    
        public static void main(String[] args) throws IOException, TimeoutException { 
            // 创建连接
            ConnectionFactory factory = new ConnectionFactory();
            // 设置 RabbitMQ 的主机名
            factory.setHost("localhost");
            // 创建一个连接
            Connection connection = factory.newConnection();
            // 创建一个通道
            Channel channel = connection.createChannel();
            // 创建一个Exchange
            channel.exchangeDeclare(EXCHANGE_NAME, "direct");
    
            try { 
            	// 将信道设置成confirm模式
                channel.confirmSelect();
                // 发送消息
                String message = "normal confirm test";
                channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes());
                // 等待发送消息的确认消息，如果发送成功，则返回ture
                // 如果发送失败，则返回false
                // 注意：如果队列不存在，也认为发送消息成功
                if (channel.waitForConfirms()) { 
                    System.out.println("send message success");
                } else { 
                    System.out.println("send message failed");
                    // do something else...
                }
            } catch (InterruptedException e) { 
                e.printStackTrace();
            }
    
            // 关闭频道和连接
            channel.close();
            connection.close();
        }
    }

wirkShark抓包可以看到如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0lUXzEw_size_16_color_FFFFFF_t_70 3]  
普通confirm模式是每发送一条消息后就调用channel.waitForConfirms()方法，之后等待服务端的确认，这实际上是一种串行同步等待的方式。因此相比于事务机制，性能提升的并不多。

> **事务机制和publisher confirm机制确保的是消息能够正确地发送至RabbitMQ，这里的“发送至RabbitMQ”的含义是指消息被正确地发往至RabbitMQ的交换器，如果此交换器没有匹配的队列，那么消息也会丢失。从业务角度看，这个消息是发送失败的，但是从confirm机制上看，这个消息是发送成功的。所以在使用这两种机制的时候要确保所涉及的交换器能够有匹配的队列。**

**第二种：批量confirm**  
批量confirm模式是每发送一批消息后，调用channel.waitForConfirms()方法，等待服务器的确认返回，因此相比于普通confirm模式，性能更好。

但是不好的地方在于，如果出现返回Basic.Nack或者超时情况，生产者客户端需要将这一批次的消息全部重发，这样会带来明显的重复消息数量，如果消息经常丢失，批量confirm模式的性能应该是不升反降的。

import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.ConnectionFactory;
    
    import java.io.IOException;
    import java.util.concurrent.ArrayBlockingQueue;
    import java.util.concurrent.BlockingQueue;
    import java.util.concurrent.TimeoutException;
    
    public class BatchConfirmProducer { 
        private final static String EXCHANGE_NAME = "batch-confirm-exchange";
    
        public static void main(String[] args) throws IOException, TimeoutException { 
            // 创建连接
            ConnectionFactory factory = new ConnectionFactory();
            // 设置 RabbitMQ 的主机名
            factory.setHost("localhost");
            // 创建一个连接
            Connection connection = factory.newConnection();
            // 创建一个通道
            Channel channel = connection.createChannel();
            // 创建一个Exchange
            channel.exchangeDeclare(EXCHANGE_NAME, "direct");
    
            int batchCount = 100;
            int msgCount = 0;
            BlockingQueue blockingQueue = new ArrayBlockingQueue(100);
            try { 
                channel.confirmSelect();
                while (msgCount <= batchCount) { 
                    String message = "batch confirm test";
                    channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes());
                    // 将发送出去的消息存入缓存中，缓存可以是一个ArrayList或者BlockingQueue之类的
                    blockingQueue.add(message);
                    if (++msgCount >= batchCount) { 
                        try { 
                            if (channel.waitForConfirms()) { 
                                // 将缓存中的消息清空
                                blockingQueue.clear();
                            } else { 
                                // 将缓存中的消息重新发送
                            }
                        } catch (InterruptedException e) { 
                            e.printStackTrace();
                            // 将缓存中的消息重新发送
                        }
                    }
                }
            } catch (IOException e) { 
                e.printStackTrace();
            }
    
            // 关闭频道和连接
            channel.close();
            connection.close();
        }
    }

**第三种：异步confirm模式**  
异步confirm模式是在生产者客户端添加ConfirmListener回调接口，重写接口的handAck()和handNack()方法，分别用来处理RabblitMQ回传的Basic.Ack和Basic.Nack。

这两个方法都有两个参数，第1个参数deliveryTag用来标记消息的唯一序列号，第2个参数multiple表示的是是否为多条确认,值为true代表是多个确认，值为false代表是单个确认。

import com.rabbitmq.client.*;
    
    import java.io.IOException;
    import java.util.SortedSet;
    import java.util.TreeSet;
    import java.util.concurrent.TimeoutException;
    
    public class AsyncConfirmProducer { 
    
        private final static String EXCHANGE_NAME = "async-confirm-exchange";
    
        public static void main(String[] args) throws IOException, TimeoutException { 
            // 创建连接
            ConnectionFactory factory = new ConnectionFactory();
            // 设置 RabbitMQ 的主机名
            factory.setHost("localhost");
            // 创建一个连接
            Connection connection = factory.newConnection();
            // 创建一个通道
            Channel channel = connection.createChannel();
            // 创建一个Exchange
            channel.exchangeDeclare(EXCHANGE_NAME, "direct");
    
            int batchCount = 100;
            long msgCount = 1;
            SortedSet<Long> confirmSet = new TreeSet<Long>();
            channel.confirmSelect();
            channel.addConfirmListener(new ConfirmListener() { 
                @Override
                public void handleAck(long deliveryTag, boolean multiple) throws IOException { 
                    System.out.println("Ack,SeqNo：" + deliveryTag + ",multiple：" + multiple);
                    if (multiple) { 
                        confirmSet.headSet(deliveryTag - 1).clear();
                    } else { 
                        confirmSet.remove(deliveryTag);
                    }
                }
    
                @Override
                public void handleNack(long deliveryTag, boolean multiple) throws IOException { 
                    System.out.println("Nack,SeqNo：" + deliveryTag + ",multiple：" + multiple);
                    if (multiple) { 
                        confirmSet.headSet(deliveryTag - 1).clear();
                    } else { 
                        confirmSet.remove(deliveryTag);
                    }
                    // 注意这里需要添加处理消息重发的场景
                }
            });
            // 演示发送100个消息
            while (msgCount <= batchCount) { 
                long nextSeqNo = channel.getNextPublishSeqNo();
                channel.basicPublish(EXCHANGE_NAME, "", null, "async confirm test".getBytes());
                confirmSet.add(nextSeqNo);
                msgCount = nextSeqNo;
            }
    
            // 关闭频道和连接
            channel.close();
            connection.close();
        }
    }

#### 性能测试 ####

Client端机器和RabbitMQ机器配置：CPU:24核，2600MHZ, 64G内存，1TB硬盘。  
Client端发送消息体大小10B，线程数为1即单线程，消息都持久化处理（deliveryMode:2）。  
分别采用事务模式、普通confirm模式，批量confirm模式和异步confirm模式进行producer实验，比对各个模式下的发送性能。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0lUXzEw_size_16_color_FFFFFF_t_70 4]  
发送平均速率：

*  事务模式（tx）：1637.484
 *  普通confirm模式(common)：1936.032
 *  批量confirm模式(batch)：10432.45
 *  异步confirm模式(async)：10542.06

可以看到事务模式性能是最差的，普通confirm模式性能比事务模式稍微好点，但是和批量confirm模式还有异步confirm模式相比，还是小巫见大巫。批量confirm模式的问题在于confirm之后返回false之后进行重发这样会使性能降低，异步confirm模式(async)编程模型较为复杂，至于采用哪种方式，那是仁者见仁智者见智了。

## 消息确认（Consumer端） ##

为了保证消息从队列可靠地到达消费者，RabbitMQ提供消息确认机制(message acknowledgment)。消费者在声明队列时，可以指定noAck参数，当noAck=false时，RabbitMQ会等待消费者显式发回ack信号后才从内存(和磁盘，如果是持久化消息的话)中移去消息。否则，RabbitMQ会在队列中消息被消费后立即删除它。

采用消息确认机制后，只要令noAck=false，消费者就有足够的时间处理消息(任务)，不用担心处理消息过程中消费者进程挂掉后消息丢失的问题，因为RabbitMQ会一直持有消息直到消费者显式调用basicAck为止。

当noAck=false时，对于RabbitMQ服务器端而言，队列中的消息分成了两部分：一部分是等待投递给消费者的消息；一部分是已经投递给消费者，但是还没有收到消费者ack信号的消息。如果服务器端一直没有收到消费者的ack信号，并且消费此消息的消费者已经断开连接，则服务器端会安排该消息重新进入队列，等待投递给下一个消费者（也可能还是原来的那个消费者）。

RabbitMQ不会为未ack的消息设置超时时间，它判断此消息是否需要重新投递给消费者的唯一依据是消费该消息的消费者连接是否已经断开。这么设计的原因是RabbitMQ允许消费者消费一条消息的时间可以很久很久。

RabbitMQ管理平台界面上可以看到当前队列中Ready状态和Unacknowledged状态的消息数，分别对应上文中的等待投递给消费者的消息数和已经投递给消费者但是未收到ack信号的消息数。也可以通过命令行来查看上述信息：  
![在这里插入图片描述][20200509161804456.png]  
代码示例（关闭自动消息确认，进行手动ack）：  
![在这里插入图片描述][20200509161815314.png]  
basicRecover：是路由不成功的消息可以使用recovery重新发送到队列中。  
basicReject：是接收端告诉服务器这个消息我拒绝接收,不处理,可以设置是否放回到队列中还是丢掉，而且只能一次拒绝一个消息,官网中有明确说明不能批量拒绝消息，为解决批量拒绝消息才有了basicNack。  
basicNack：可以一次拒绝N条消息，客户端可以设置basicNack方法的multiple参数为true，服务器会拒绝指定了delivery\_tag的所有未确认的消息(tag是一个64位的long值，最大值是9223372036854775807)。

[20200509161221810.png]: /images/20230312/a0acb34a89d248c58d5a78531d95dbab.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0lUXzEw_size_16_color_FFFFFF_t_70]: /images/20230312/ef8f39a6decb41098a50bf73da250ba1.png
[2020050916130195.png]: /images/20230312/879d4914c613475f809eca744c1b0873.png
[20200509161401140.png]: /images/20230312/0fcd66b6c337450490149471b8be1c26.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0lUXzEw_size_16_color_FFFFFF_t_70 1]: /images/20230312/b965bad0709f4d1a992a1623a29c7395.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0lUXzEw_size_16_color_FFFFFF_t_70 2]: /images/20230312/00602f4699004dc3b263f57f1d3c4072.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0lUXzEw_size_16_color_FFFFFF_t_70 3]: /images/20230312/39550771741a4aa0a41d361bd09304d7.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0lUXzEw_size_16_color_FFFFFF_t_70 4]: /images/20230312/cc063e6903584942a64a547536ce85e0.png
[20200509161804456.png]: /images/20230312/a7901d78cef645b897c766d6b63d8e81.png
[20200509161815314.png]: /images/20230312/eb7c38727ac548f0b979fbba43becf79.png