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

文章目录

- 模式介绍
- 建立连接
- 单独确认
- - 代码实现逻辑
  - 运行结果
- 批量确认
- - 代码实现逻辑
  - 运行结果
- 异步确认
- - 实现逻辑介绍
  - 代码实现逻辑
  - 运行结果
- 三种策略对比以及完整代码

模式介绍

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

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

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

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

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

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

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

当消息最终得到确认之后，生产者可以通过回调方法来处理该确认消息
如果 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);  
        }  
    }  
  
}