【RabbitMQ】发布确认机制的具体实现

椰椰椰耶

发布于 2025-05-15 01:45:44

8400

代码可运行

文章被收录于专栏：学习学习

运行总次数：0

代码可运行

模式介绍

作为消息中间件，都会面临消息丢失的问题，消息丢失大概分为三种情况：

生产者问题：因为应用程序故障，网络抖动等各种原因，生产者没有成功向 broker 发送消息
消息中间件自身问题：生产者成功发送给了 Broker，但是 Broker 没有把消息保存好，导致消息丢失
消费者问题：Broker 发送消息到消费者，消费者在消费消息时，因为没有处理好，导致 broker 将消费失败的消息从列表中删除了

RabbitMQ 也对上述问题给出了相应的解决方案。
- 问题二可以通过持久化机制
- 问题三可以采用消息应答机制
- 问题一可以采用发布确认机制

发布确认属于 RabbitMQ 的七大工作模式之一

生产者将信道设置成 confirm（确认）模式

一旦信道进入 confirm 模式，所有在该信道上面发布的消息都会被指派一个唯一的 ID（从 1 开始）
- 同一个 channel 下，序号不可能重复
一旦消息被投递到所有匹配的对类之后，brocker 就会发送一个确认(ACK)给生产者（包含消息的唯一 ID）
- 这就使得生产者知道消息已经正确到达目的的队列了
如果消息和对类是可持久化的，那么确认消息会在将消息写入磁盘之后发出
- broker 回传给生产者的确认消息中 deliveryTag 包含了确认消息的序号
- 此外 broker 也可以设置 channel.basicAck 方法中的 multiple 参数，表示到这个序号之前的所有消息都已经得到了处理
image.png

发送确认机制最大的好处在于它是异步的，生产者可以同时发布消息和等待信道返回确认消息

当消息最终得到确认之后，生产者可以通过回调方法来处理该确认消息
如果 RabbitMQ 因为自身内部错误导致消息丢失，就会发送一条 nack(Basic.Nack) 命令，生产者同样可以在回调方法中处理该 nack 命令

使用发布确认机制，必须要信道设置成 confirm (确认) 模式

发布确认是 AMQP 0.9.1 协议的扩展，默认情况下它不会被启用
生产者通过 channel.confirmSelect() 将信道设置为 confirm 模式

Channel channel = connection.createChannel();
channel.confirmSelect();

发布确认有三种策略，接下来我们来介绍这三种策略

Producer、Brocker、Consumer 都有可能丢失消息

发布确认是来解决生产者 Producer 消息丢失的问题
生产者可以在发送消息的同时，等待返回确认消息

建立连接

因为每一个策略都需要重复建立链接这一步骤，所以我们将其提出来，单独作为一个方法，需要的时候直接调用即可

之后就不用重复写这一部分代码了
在类里面，main 方法外面，使用一个静态方法

public class PublisherConfirms {  
  
    static Connection createConnection() throws Exception {  
        ConnectionFactory connectionFactory = new ConnectionFactory();  
        connectionFactory.setHost(Constants.HOST);  
        connectionFactory.setPort(Constants.PORT);  
        connectionFactory.setUsername(Constants.USER_NAME);  
        connectionFactory.setPassword(Constants.PASSWORD);  
        connectionFactory.setVirtualHost(Constants.VIRTUAL_HOST);  
        return connectionFactory.newConnection();  
    }  
  
    public static void main(String[] args) {  
        
    }  
  
}

单独确认

Publishing Messages Individually

代码实现逻辑

    /**  
     * 单独确认  
     */  
    private static void publishingMessagesIndividually() throws Exception {  
        // 1. 创建连接  
        // 我们将连接的建立写在 try 里面，这样就不用再去关闭了  
        try(Connection connection = createConnection()) {  
            // 2. 开启信道  
            Channel channel = connection.createChannel();  
  
            // 3. 设置信道为 confirms 模式  
            channel.confirmSelect();  
  
            // 4. 声明队列(交换机就使用内置的，就不再声明了)  
            // 队列对象、是否持久化、是否独占、是否自动删除、传递参数  
            channel.queueDeclare(Constants.PUBLISHER_CONFIRMS_QUEUE1, true, false, false,null);  
  
            // 5. 发送消息，并等待确认  
            // 这里我们需要可以再创建一个 MESSAGE_COUNT 全局变量，来指定消息的数量  
            long start = System.currentTimeMillis();  
            for (int i = 0; i < MESSAGE_COUNT; i++) {  
                String msg = "hello publisher confirms" + i;  
                // 信道的发送  
                // 交换机的名称(我们使用的是内置交换机，也就是空的)、routingKey、  
                channel.basicPublish("", Constants.PUBLISHER_CONFIRMS_QUEUE1, null, msg.getBytes());  
  
                // 等待确认(等待确认消息，只要消息被确认，这个方法就会被返回)  
                // 有 waitForConfirms() 和 waitForConfirmsOrDie() 随便用哪个  
                // 如果超时过期，则抛出 TimeoutException。如果任何消息被 nack(丢失)，waitForConfirmsOrDie 则抛出 Exception  
                channel.waitForConfirmsOrDie(5000);  
            }  
            long end = System.currentTimeMillis();  
            // 这里注意是 printf            System.out.printf("单独确认==>消息条数: %d, 耗时: %d ms \n", MESSAGE_COUNT, end-start);  
        }  
    }

运行结果

单独确认==>消息条数: 200, 耗时: 5793 ms

可以发现，耗时较长

观察上面代码，会发现这种策略是每发送一条消息后就调用 channel.waitForConfirmsOrDie() 方法, 之后等待服务器的确认

这其实是一种串行同步等待的方式
尤其对于持久化的消息来说，需要等待消息确认存储在硬盘之后才会返回 (调用 Linux 内核中的 fsync 方法)

但是消息确认机制是支持异步的，可以一边发送消息，一边等待消息确认。由此进行了改进，我们看另外两种策略

Publishing Messages in Batches(批量确认)：每发送一批消息之后，调用 channel.waitForConfirms 方法，等待服务器的确认返回
Handling Publisher Confirms Asynchronously(异步确认)：提供一个回调方法，服务端确认了一条或者多条消息后，客户端会对这个方法进行处理

批量确认

Publishing Messages in Batches

代码实现逻辑

/**  
 * 批量确认  
 */  
private static void publishingMessagesInBatches() throws Exception {  
    // 1. 建立连接  
    try(Connection connection = createConnection()){  
        // 2. 开启信道  
        Channel channel = connection.createChannel();  
  
        // 3. 设置信道为 confirm 模式  
        channel.confirmSelect();  
  
        // 4. 声明队列  
        channel.queueDeclare(Constants.PUBLISHER_CONFIRMS_QUEUE2, true, false, false, null);  
  
        // 5. 发送消息，并进行确认  
        // 设置批量处理的大小和计数器  
        long start = System.currentTimeMillis();  
        int batchSize = 100;  
        int outstandingMessageCount = 0;  
        for (int i = 0; i < MESSAGE_COUNT; i++) {  
            String msg = "hello publisher confirms" + i;  
            channel.basicPublish("", Constants.PUBLISHER_CONFIRMS_QUEUE2, null, msg.getBytes());  
            outstandingMessageCount++;  
            // 当计数器的大小达到了设置的批量处理大小，就进行确认  
            if(outstandingMessageCount == batchSize) {  
                channel.waitForConfirmsOrDie(5000);  
                // 消息确认后，计数器要清零  
                outstandingMessageCount = 0;  
            }  
  
            // 当计数器大小 < 100 的时候，由于没有达到批量发送的标准，所以单独再进行发送  
            if (outstandingMessageCount > 0) {  
                channel.waitForConfirmsOrDie(5000);  
            }  
        }  
        long end = System.currentTimeMillis();  
  
        System.out.printf("批量确认==>消息条数: %d, 耗时: %d ms", MESSAGE_COUNT, end-start);  
    }  
}

运行结果

批量确认==>消息条数: 200, 耗时: 128 ms

异步确认

Handling Publisher Confirms Asynchronously

实现逻辑介绍

异步 confirm 方法的编程实现最为复杂

Channel 接口提供了一个方法 addConfirmListener()
这个方法可以添加 ConfirmListener 回调接口

ConfirmListener 接口中包含两个方法：

handleAck(long deliveryTag, boolean multiple)，处理 RabbitMQ 发送给生产者的 ack
- deliveryTag 表示发送消息的序号
- multiple 表示是否批量确认
handleNack(long deliveryTag, boolean multiple)，处理 RabbitMQ 发送给生产者的 nack

我们需要为每一个 Channel 维护一个已发送消息的序号集合

当收到 RabbitMQ 的 confirm 回调时，从集合中删除对应的消息
当 Channel 开启 confirm 模式后，channel 上发送消息都会附带一个从 1 开始递增的 deliveryTag 序号
我们可以使用 SortedSet 的有序性来维护这个已发消息的集合
1. 当收到 ack 时，从序列中删除该消息的序号。如果为批量确认消息，表示小于当前序号 deliveryTag 的消息都收到了，则清楚对应集合
2. 当收到 nack 时，处理逻辑类似，不过需要结合具体业务情况，进行消息重发等操作

代码实现逻辑

/**  
 * 异步确认  
 */  
private static void handlingPublisherConfirmsAsynchronously() throws Exception {  
    // 1. 建立连接  
    try(Connection connection = createConnection()) {  
        // 2. 开启信道  
        Channel channel = connection.createChannel();  
  
        // 3. 设置信道为 confirm 模式  
        channel.confirmSelect();  
  
        // 4. 声明队列  
        channel.queueDeclare(Constants.PUBLISHER_CONFIRMS_QUEUE3, true, false, false, null);  
  
        // 5. 监听 confirm        long start = System.currentTimeMillis();  
        // 创建一个集合，用来存放未确认的消息(的id)  
  
        SortedSet<Long> confirmSeqNo = Collections.synchronizedSortedSet(new TreeSet<>());  
  
        channel.addConfirmListener(new ConfirmListener() {  
            @Override  
            public void handleAck(long deliveryTag, boolean multiple) throws IOException {  
                // 如果是批量确认，就要将集合中 <= deliveryTag 的 id 都给清除掉  
                if(multiple) {  
                    // headSet(n)方法返回当前集合中小于 n 的集合  
                    // 先获取到这部分 id，然后一起 clear 清除掉即可  
                    confirmSeqNo.headSet(deliveryTag + 1).clear();  
                }else {  
                    // 单独确认，只需要移除当前这个 id 即可  
                    confirmSeqNo.remove(deliveryTag);  
                }  
            }  
  
            @Override  
            public void handleNack(long deliveryTag, boolean multiple) throws IOException {  
                // 和 ack 处理模式基本是相似的，只是多了一步重发处理  
                if(multiple) {  
                    confirmSeqNo.headSet(deliveryTag + 1).clear();  
                }else {  
                    confirmSeqNo.remove(deliveryTag);  
                }  
  
                // 业务需要根据实际场景进行处理，比如重发，此处代码省略  
            }  
        });  
  
        // 6. 发送消息  
        for (int i = 0; i < MESSAGE_COUNT; i++) {  
            String msg = "hello publisher confirms" + i;  
            long seqNo = channel.getNextPublishSeqNo(); // 拿到消息的序号  
            channel.basicPublish("", Constants.PUBLISHER_CONFIRMS_QUEUE3, null, msg.getBytes());  
            confirmSeqNo.add(seqNo); // 将消息的序号加入集合中  
        }  
  
        // 确认消息已处理完  
        while (!confirmSeqNo.isEmpty()) {  
            // 没有处理完，就休眠一段时间后再确认一下，看是否处理完  
            Thread.sleep(10);  
        }  
        long end = System.currentTimeMillis();  
        System.out.printf("异步确认==>消息条数: %d, 耗时: %d ms", MESSAGE_COUNT, end-start);  
    }  
}

运行结果

单独确认==>消息条数: 200, 耗时: 93 ms

三种策略对比以及完整代码

package rabbitmq.publisher.confirms;  
  
import com.rabbitmq.client.Channel;  
import com.rabbitmq.client.ConfirmListener;  
import com.rabbitmq.client.Connection;  
import com.rabbitmq.client.ConnectionFactory;  
import rabbitmq.constant.Constants;  
  
import java.io.IOException;  
import java.util.Collections;  
import java.util.SortedSet;  
import java.util.TreeSet;  
  
public class PublisherConfirms {  
  
    private static final Integer MESSAGE_COUNT = 200;  
  
    static Connection createConnection() throws Exception {  
        ConnectionFactory connectionFactory = new ConnectionFactory();  
        connectionFactory.setHost(Constants.HOST);  
        connectionFactory.setPort(Constants.PORT);  
        connectionFactory.setUsername(Constants.USER_NAME);  
        connectionFactory.setPassword(Constants.PASSWORD);  
        connectionFactory.setVirtualHost(Constants.VIRTUAL_HOST);  
        return connectionFactory.newConnection();  
    }  
  
    public static void main(String[] args) throws Exception {  
        // Strategy #1: Publishing Messages Individually  
        // 单独确认  
        publishingMessagesIndividually();  
  
        // Strategy #2: Publishing Messages in Batches  
        // 批量确认  
        publishingMessagesInBatches();  
  
        // Strategy #3: Handling Publisher Confirms Asynchronously  
        // 异步确认  
        handlingPublisherConfirmsAsynchronously();  
    }  
  
    /**  
     * 单独确认  
     */  
    private static void publishingMessagesIndividually() throws Exception {  
        // 1. 创建连接  
        // 我们将连接的建立写在 try 里面，这样就不用再去关闭了  
        try(Connection connection = createConnection()) {  
            // 2. 开启信道  
            Channel channel = connection.createChannel();  
  
            // 3. 设置信道为 confirms 模式  
            channel.confirmSelect();  
  
            // 4. 声明队列(交换机就使用内置的，就不再声明了)  
            // 队列对象、是否持久化、是否独占、是否自动删除、传递参数  
            channel.queueDeclare(Constants.PUBLISHER_CONFIRMS_QUEUE1, true, false, false,null);  
  
            // 5. 发送消息，并等待确认  
            // 这里我们需要可以再创建一个 MESSAGE_COUNT 全局变量，来指定消息的数量  
            long start = System.currentTimeMillis();  
            for (int i = 0; i < MESSAGE_COUNT; i++) {  
                String msg = "hello publisher confirms" + i;  
                // 信道的发送  
                // 交换机的名称(我们使用的是内置交换机，也就是空的)、routingKey、  
                channel.basicPublish("", Constants.PUBLISHER_CONFIRMS_QUEUE1, null, msg.getBytes());  
  
                // 等待确认(等待确认消息，只要消息被确认，这个方法就会被返回)  
                // 有 waitForConfirms() 和 waitForConfirmsOrDie() 随便用哪个  
                // 如果超时过期，则抛出 TimeoutException。如果任何消息被 nack(丢失)，waitForConfirmsOrDie 则抛出 Exception                channel.waitForConfirmsOrDie(5000);  
            }  
            long end = System.currentTimeMillis();  
            // 这里注意是 printf            System.out.printf("单独确认==>消息条数: %d, 耗时: %d ms \n", MESSAGE_COUNT, end-start);  
        }  
    }  
  
  
    /**  
     * 批量确认  
     */  
    private static void publishingMessagesInBatches() throws Exception {  
        // 1. 建立连接  
        try(Connection connection = createConnection()){  
            // 2. 开启信道  
            Channel channel = connection.createChannel();  
  
            // 3. 设置信道为 confirm 模式  
            channel.confirmSelect();  
  
            // 4. 声明队列  
            channel.queueDeclare(Constants.PUBLISHER_CONFIRMS_QUEUE2, true, false, false, null);  
  
            // 5. 发送消息，并进行确认  
            // 设置批量处理的大小和计数器  
            long start = System.currentTimeMillis();  
            int batchSize = 100;  
            int outstandingMessageCount = 0;  
            for (int i = 0; i < MESSAGE_COUNT; i++) {  
                String msg = "hello publisher confirms" + i;  
                channel.basicPublish("", Constants.PUBLISHER_CONFIRMS_QUEUE2, null, msg.getBytes());  
                outstandingMessageCount++;  
                // 当计数器的大小达到了设置的批量处理大小，就进行确认  
                if(outstandingMessageCount == batchSize) {  
                    channel.waitForConfirmsOrDie(5000);  
                    // 消息确认后，计数器要清零  
                    outstandingMessageCount = 0;  
                }  
  
                // 当计数器大小 < 100 的时候，由于没有达到批量发送的标准，所以单独再进行发送  
                if (outstandingMessageCount > 0) {  
                    channel.waitForConfirmsOrDie(5000);  
                }  
            }  
            long end = System.currentTimeMillis();  
  
            System.out.printf("批量确认==>消息条数: %d, 耗时: %d ms \n", MESSAGE_COUNT, end-start);  
        }  
    }  
  
  
    /**  
     * 异步确认  
     */  
    private static void handlingPublisherConfirmsAsynchronously() throws Exception {  
        // 1. 建立连接  
        try(Connection connection = createConnection()) {  
            // 2. 开启信道  
            Channel channel = connection.createChannel();  
  
            // 3. 设置信道为 confirm 模式  
            channel.confirmSelect();  
  
            // 4. 声明队列  
            channel.queueDeclare(Constants.PUBLISHER_CONFIRMS_QUEUE3, true, false, false, null);  
  
            // 5. 监听 confirm            long start = System.currentTimeMillis();  
            // 创建一个集合，用来存放未确认的消息(的id)  
  
            SortedSet<Long> confirmSeqNo = Collections.synchronizedSortedSet(new TreeSet<>());  
  
            channel.addConfirmListener(new ConfirmListener() {  
                @Override  
                public void handleAck(long deliveryTag, boolean multiple) throws IOException {  
                    // 如果是批量确认，就要将集合中 <= deliveryTag 的 id 都给清除掉  
                    if(multiple) {  
                        // headSet(n)方法返回当前集合中小于 n 的集合  
                        // 先获取到这部分 id，然后一起 clear 清除掉即可  
                        confirmSeqNo.headSet(deliveryTag + 1).clear();  
                    }else {  
                        // 单独确认，只需要移除当前这个 id 即可  
                        confirmSeqNo.remove(deliveryTag);  
                    }  
                }  
  
                @Override  
                public void handleNack(long deliveryTag, boolean multiple) throws IOException {  
                    // 和 ack 处理模式基本是相似的，只是多了一步重发处理  
                    if(multiple) {  
                        confirmSeqNo.headSet(deliveryTag + 1).clear();  
                    }else {  
                        confirmSeqNo.remove(deliveryTag);  
                    }  
  
                    // 业务需要根据实际场景进行处理，比如重发，此处代码省略  
                }  
            });  
  
            // 6. 发送消息  
            for (int i = 0; i < MESSAGE_COUNT; i++) {  
                String msg = "hello publisher confirms" + i;  
                long seqNo = channel.getNextPublishSeqNo(); // 拿到消息的序号  
                channel.basicPublish("", Constants.PUBLISHER_CONFIRMS_QUEUE3, null, msg.getBytes());  
                confirmSeqNo.add(seqNo); // 将消息的序号加入集合中  
            }  
  
            // 确认消息已处理完  
            while (!confirmSeqNo.isEmpty()) {  
                // 没有处理完，就休眠一段时间后再确认一下，看是否处理完  
                Thread.sleep(10);  
            }  
            long end = System.currentTimeMillis();  
            System.out.printf("异步确认==>消息条数: %d, 耗时: %d ms", MESSAGE_COUNT, end-start);  
        }  
    }  
  
}