RabbitMQ 常见工作模式及发布确认实现

1.2 消费者

我们创建两个消费者模拟。消费者步骤：

1. 创建连接
2. 创建 Channel
3. 声明一个队列 Queue
4. 消费消息
5. 释放资源

package org.example.rabbitmq.workqueues;

import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class Consumer1 {
    public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
        // 建立连接
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("localhost");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("guest");
        connectionFactory.setPassword("guest");
        connectionFactory.setVirtualHost("/");
        Connection connection = connectionFactory.newConnection();
        
        // 创建信道
        Channel channel = connection.createChannel();
        
        // 声明队列
        channel.queueDeclare("workQueues", false, false, true, null);
        
        // 消费消息
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("Consumer1 接收到消息：" + new String(body));
            }
        };
        channel.basicConsume("workQueues", true, consumer);
        Thread.sleep(100);
        
        // 释放资源
        // channel.close();
        // connection.close();
    }
}

二、Publish/Subscribe（发布/订阅）

在发布/订阅模型中，多了一个 Exchange 角色。

Exchange 常见有三种类型，分别代表不同的路由规则，也就分别对应不同的工作模式：

1. Fanout：广播，将消息交给所有绑定到交换机的队列（Publish/Subscribe 模式）
2. Direct：定向，把消息交给符合指定 routing key 的队列（Routing 模式）
3. Topic：通配符，把消息交给符合 routing pattern（路由模式）的队列（Topics 模式）

2.1 生产者

生产者步骤：

1. 创建连接
2. 创建信道
3. 声明交换机
4. 声明两个队列
5. 交换机与队列进行绑定
6. 生产消息
7. 释放资源

声明交换机的方法是 Channel 类下的 exchangeDeclare 方法。

• exchange – the name of the exchange，交换机名称
• type – the exchange type，交换机类型
• durable – true if we are declaring a durable exchange，是否可持久化
• autoDelete – true if the server should delete the exchange when it is no longer in use，是否自动删除
• internal – true if the exchange is internal，是否内部使用
• arguments – other properties，参数

package org.example.rabbitmq.fanout;

import com.rabbitmq.client.BuiltinExchangeType;
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 Producer {
    public static void main(String[] args) throws IOException, TimeoutException {
        // 1. 创建连接
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("localhost");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("guest");
        connectionFactory.setPassword("guest");
        connectionFactory.setVirtualHost("/");
        Connection connection = connectionFactory.newConnection();
        
        // 2. 创建信道
        Channel channel = connection.createChannel();
        
        // 3. 声明交换机
        channel.exchangeDeclare("fanout.exchange", BuiltinExchangeType.FANOUT, false);
        
        // 4. 声明队列
        channel.queueDeclare("fanout.queue1", true, false, false, null);
        channel.queueDeclare("fanout.queue2", true, false, false, null);
        
        // 5. 交换机与队列绑定
        channel.queueBind("fanout.queue1", "fanout.exchange", "");
        channel.queueBind("fanout.queue2", "fanout.exchange", "");
        
        // 6. 生产消息
        for (int i = 0; i < 10; i++) {
            String msg = "hello fanout" + i;
            channel.basicPublish("fanout.exchange", "", null, msg.getBytes());
        }
        System.out.println("发送消息成功");
        
        // 7. 释放资源
        channel.close();
        connection.close();
    }
}

2.2 消费者

两个消费者分别消费两个队列的消息。

消费者步骤：

1. 创建连接
2. 创建 Channel
3. 声明一个队列 Queue
4. 消费消息
5. 释放资源

package org.example.rabbitmq.fanout;

import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class Consumer1 {
    public static void main(String[] args) throws IOException, TimeoutException {
        // 创建连接
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("localhost");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("guest");
        connectionFactory.setPassword("guest");
        connectionFactory.setVirtualHost("/");
        Connection connection = connectionFactory.newConnection();
        
        // 创建信道
        Channel channel = connection.createChannel();
        
        // 声明队列
        channel.queueDeclare("fanout.queue1", true, false, false, null);
        
        // 消费消息
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("Consumer1 接收到消息：" + new String(body));
            }
        };
        channel.basicConsume("fanout.queue1", consumer);
        /* //释放资源
        channel.close();
        connection.close();*/
    }
}

三、Routing（路由模式）

Routing（路由模式）：队列和交换机的绑定，不能是任意的绑定了，而是要指定一个 BindingKey（RoutingKey 的一种），消息的发送方在向 Exchange 发送消息时，也需要指定消息的 RoutingKey，Exchange 也不再把消息交给每一个绑定的 key，而是根据消息的 RoutingKey 进行判断，只有队列绑定时的 BindingKey 和发送消息的 RoutingKey 完全一致，才会接收到消息。

3.1 生产者

生产者步骤：

1. 创建连接
2. 创建信道
3. 声明交换机，类型为 direct
4. 声明队列
5. 队列与交换机绑定，绑定的时候加上 BindingKey 参数
6. 生产消息，消息发送的时候指定 routingKey
7. 释放资源

package org.example.rabbitmq.direct;

import com.rabbitmq.client.BuiltinExchangeType;
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 Producer {
    public static void main(String[] args) throws IOException, TimeoutException {
        // 1. 创建连接
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("localhost");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("guest");
        connectionFactory.setPassword("guest");
        connectionFactory.setVirtualHost("/");
        Connection connection = connectionFactory.newConnection();
        
        // 2. 创建信道
        Channel channel = connection.createChannel();
        
        // 3. 声明交换机
        channel.exchangeDeclare("direct.exchange", BuiltinExchangeType.DIRECT, true);
        
        // 4. 声明队列
        channel.queueDeclare("direct.queue1", true, false, false, null);
        channel.queueDeclare("direct.queue2", true, false, false, null);
        
        // 5. 队列与交换机绑定
        channel.queueBind("direct.queue1", "direct.exchange", "a");
        channel.queueBind("direct.queue2", "direct.exchange", "a");
        channel.queueBind("direct.queue2", "direct.exchange", "b");
        channel.queueBind("direct.queue2", "direct.exchange", "c");
        
        // 6. 生产消息
        String msgA = "hello direct routingKey is a";
        channel.basicPublish("direct.exchange", "a", null, msgA.getBytes());
        String msgB = "hello direct routingKey is b";
        channel.basicPublish("direct.exchange", "b", null, msgB.getBytes());
        String msgC = "hello direct routingKey is c";
        channel.basicPublish("direct.exchange", "c", null, msgC.getBytes());
        System.out.println("发送消息成功");
        
        // 7. 释放资源
        channel.close();
        connection.close();
    }
}

3.2 消费者

两个消费者分别消费两个队列的消息。

消费者步骤：

1. 创建连接
2. 创建 Channel
3. 声明一个队列 Queue
4. 消费消息
5. 释放资源

package org.example.rabbitmq.direct;

import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class Consumer1 {
    public static void main(String[] args) throws IOException, TimeoutException {
        // 创建连接
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("localhost");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("guest");
        connectionFactory.setPassword("guest");
        connectionFactory.setVirtualHost("/");
        Connection connection = connectionFactory.newConnection();
        
        // 创建信道
        Channel channel = connection.createChannel();
        
        // 声明队列
        channel.queueDeclare("direct.queue1", true, false, false, null);
        
        // 消费消息
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("Consumer1 接收到消息：" + new String(body));
            }
        };
        channel.basicConsume("direct.queue1", consumer);
        /* //释放资源
        channel.close();
        connection.close();*/
    }
}

四、Topics（通配符模式）

Topics 和 Routing 模式的区别是：

1. topics 模式使用的交换机类型为 topic（Routing 模式使用的交换机类型为 direct）
2. topic 类型的交换机在匹配规则上进行了扩展，Binding Key 支持通配符匹配（direct 类型的交换机路由规则是 BindingKey 和 RoutingKey 完全匹配）。

在 topic 类型的交换机在匹配规则上，有些要求：

1. RoutingKey 是一系列由点 ( . ) 分隔的单词，比如 "stock.usd.nyse"，"nyse.vmw"，"quick.orange.rabbit"
2. BindingKey 和 RoutingKey 一样，也是点 ( . ) 分割的字符串。
3. Binding Key 中可以存在两种特殊字符串，用于模糊匹配
   • * 表示一个单词
   • # 表示多个单词 (0-N 个)

4.1 生产者

生产者步骤：

1. 创建连接
2. 创建信道
3. 声明交换机，类型为 topic
4. 声明队列
5. 队列与交换机绑定，绑定的时候加上 BindingKey 参数
6. 生产消息，消息发送的时候指定 routingKey
7. 释放资源

package org.example.rabbitmq.topics;

import com.rabbitmq.client.BuiltinExchangeType;
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 Producer {
    public static void main(String[] args) throws IOException, TimeoutException {
        // 1. 创建连接
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("localhost");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("guest");
        connectionFactory.setPassword("guest");
        connectionFactory.setVirtualHost("/");
        Connection connection = connectionFactory.newConnection();
        
        // 2. 创建信道
        Channel channel = connection.createChannel();
        
        // 3. 声明交换机
        channel.exchangeDeclare("topic.exchange", BuiltinExchangeType.TOPIC, true);
        
        // 4. 声明队列
        channel.queueDeclare("topic.queue1", true, false, false, null);
        channel.queueDeclare("topic.queue2", true, false, false, null);
        
        // 5. 队列与交换机绑定
        channel.queueBind("topic.queue1", "topic.exchange", "*.a.*");
        channel.queueBind("topic.queue2", "topic.exchange", "*.*.b");
        channel.queueBind("topic.queue2", "topic.exchange", "c.#");
        
        // 6. 生产消息
        String msgA = "hello topic routingKey is word.a.word";
        channel.basicPublish("topic.exchange", "word.a.word", null, msgA.getBytes());
        String msgB = "hello topic routingKey is word.word.b";
        channel.basicPublish("topic.exchange", "word.word.b", null, msgB.getBytes());
        String msgC = "hello topic routingKey is c.word.word.word.word.b";
        channel.basicPublish("topic.exchange", "c.word.word.word.word.b", null, msgC.getBytes());
        String msgD = "hello topic routingKey is c.a.b";
        channel.basicPublish("topic.exchange", "c.a.b", null, msgD.getBytes());
        System.out.println("发送消息成功");
        
        // 7. 释放资源
        channel.close();
        connection.close();
    }
}

4.2 消费者

两个消费者分别消费两个队列的消息。

消费者步骤：

1. 创建连接
2. 创建 Channel
3. 声明一个队列 Queue
4. 消费消息
5. 释放资源

package org.example.rabbitmq.topics;

import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class Consumer1 {
    public static void main(String[] args) throws IOException, TimeoutException {
        // 创建连接
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("localhost");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("guest");
        connectionFactory.setPassword("guest");
        connectionFactory.setVirtualHost("/");
        Connection connection = connectionFactory.newConnection();
        
        // 创建信道
        Channel channel = connection.createChannel();
        
        // 声明队列
        channel.queueDeclare("topic.queue1", true, false, false, null);
        
        // 消费消息
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("Consumer1 接收到消息：" + new String(body));
            }
        };
        channel.basicConsume("topic.queue1", consumer);
        // 释放资源
        // channel.close();
        // connection.close();
    }
}

五、RPC 通信

RPC（Remote Procedure Call），即远程过程调用。它是一种通过网络从远程计算机上请求服务，而不需要了解底层网络的技术。类似于 Http 远程调用。

RabbitMQ 实现 RPC 通信的过程，大概是通过两个队列实现一个可回调的过程。

大概流程如下：

1. 客户端发送消息到一个指定的队列，并在消息属性中设置 replyTo 字段，这个字段指定了一个回调队列，服务端处理后，会把响应结果发送到这个队列。
2. 服务端接收到请求后，处理请求并发送响应消息到 replyTo 指定的回调队列
3. 客户端在回调队列上等待响应消息，一旦收到响应，客户端会检查消息的 correlationId 属性，以确保它是所期望的响应。

5.1 客户端

客户端步骤：

1. 声明两个队列，包含回调队列 replyQueueName，声明本次请求的唯一标志 corrId
2. 将 replyQueueName 和 corrId 配置到要发送的消息队列中
3. 使用阻塞队列来阻塞当前进程，监听回调队列中的消息，把请求放到阻塞队列中
4. 阻塞队列有消息后，主线程被唤醒，打印返回内容

package org.example.rabbitmq.rpc;

import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.UUID;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeoutException;

public class Client {
    public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
        // 1. 创建连接
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("localhost");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("guest");
        connectionFactory.setPassword("guest");
        connectionFactory.setVirtualHost("/");
        Connection connection = connectionFactory.newConnection();
        
        // 2. 创建信道
        Channel channel = connection.createChannel();
        
        // 3. 声明队列
        channel.queueDeclare("rpc.request.queue", true, false, false, null);
        channel.queueDeclare("rpc.response.queue", true, false, false, null);
        
        // 4. 发送消息
        // 设置请求标识
        String correlationId = UUID.randomUUID().toString();
        AMQP.BasicProperties properties = new AMQP.BasicProperties().builder()
                .correlationId(correlationId)
                .replyTo("rpc.response.queue")
                .build();
        String msg = "hello rpc";
        channel.basicPublish("", "rpc.request.queue", properties, msg.getBytes());
        
        // 5. 接收响应
        // 使用阻塞队列存储响应
        final BlockingQueue<String> response = new ArrayBlockingQueue<>(1);
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("Client 接收到消息：" + new String(body));
                // 判断唯一标识正确，放到阻塞队列中
                if (correlationId.equals(properties.getCorrelationId())) {
                    response.offer(new String(body));
                }
            }
        };
        channel.basicConsume("rpc.response.queue", true, consumer);
        
        // 获取响应的结果
        String result = response.take();
        System.out.println(" [RPC_Client] Result:" + result);
    }
}

5.2 服务器

服务器步骤：

1. 接收消息
2. 根据消息内容进行响应处理，把应答结果返回到回调队列中

package org.example.rabbitmq.rpc;

import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class Server {
    public static void main(String[] args) throws IOException, TimeoutException {
        // 1. 创建连接
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("localhost");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("guest");
        connectionFactory.setPassword("guest");
        connectionFactory.setVirtualHost("/");
        Connection connection = connectionFactory.newConnection();
        
        // 2. 创建信道
        Channel channel = connection.createChannel();
        
        // 接收请求
        // 每次接受一条
        channel.basicQos(1);
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("Server 接收到请求：" + new String(body));
                
                // 响应请求
                String response = "响应";
                AMQP.BasicProperties props = new AMQP.BasicProperties().builder()
                        .correlationId(properties.getCorrelationId())
                        .replyTo("rpc.request.queue")
                        .build();
                channel.basicPublish("", "rpc.response.queue", props, response.getBytes());
                
                // 手动确认
                channel.basicAck(envelope.getDeliveryTag(), false);
            }
        };
        channel.basicConsume("rpc.request.queue", false, consumer);
    }
}

六、Publisher Confirms（发布确认）

消息中间件都会有消息丢失的问题发生，大概分为以下三种丢失情况：

1. 生产者问题：因为应用故障，网络等问题，生产者没有成功向消息中间件发送消息；
2. 消息中间件问题：生产者成功发送了消息，消息中间件自己原因导致消息丢失；
3. 消费者问题：消费者消费消息时，处理出现问题，导致消费者消费失败的消息从消息队列中删除了。

RabbitMQ 针对上面三种情况给出的解决方案：

1. 针对生产者问题：采取 Publisher Confirms（发布确认）机制解决；
2. 针对消息中间件问题：通过持久化机制解决；
3. 针对消费者问题：通过消息应答机制解决；

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

6.1 Publishing Messages Individually（单独确认）

跟生产者发送消息，只有调用 Channel 类 confirmSelect() 设置信道为 confirm 模式，和 Channel 类 waitForConfirmsOrDie() 方法等待手动确认。

private static void publishingMessagesIndividually() throws IOException, TimeoutException, InterruptedException {
    try (Channel channel = createChannel()) {
        // 设置信道为 confirm 模式
        channel.confirmSelect();
        
        // 声明队列
        channel.queueDeclare("publish.confirm.queue1", true, false, false, null);
        
        long start = System.currentTimeMillis();
        
        // 发送消息
        for (int i = 0; i < 200; i++) {
            String msg = "Publishing Messages Individually " + i;
            channel.basicPublish("", "publish.confirm.queue1", null, msg.getBytes());
            
            // 等待确认
            channel.waitForConfirmsOrDie(5000);
        }
        
        long end = System.currentTimeMillis();
        System.out.println("Publishing Messages Individually(单独确认) 发送 200 条消息耗时 " + (end - start));
    }
}

可以发现，发送 200 条消息，耗时很长。

观察上面代码，会发现这种策略是每发送一条消息后就调用 channel.waitForConfirmsOrDie() 方法，之后等待服务端的确认，这实际上是一种串行同步等待的方式。 尤其对于持久化的消息来说，需要等待消息确认存储在磁盘之后才会返回（调用 Linux 内核的 fsync 方法）。 但是发布确认机制是支持异步的。可以一边发送消息，一边等待消息确认。

6.2 Publishing Messages in Batches（批量确认）

每发送一批消息后，调用 channel.waitForConfirms 方法，等待服务器的确认返回。

跟单独确认区别就是，发送到一定消息再进行等待确认。

private static void publishingMessagesInBatches() throws IOException, TimeoutException, InterruptedException {
    try (Channel channel = createChannel()) {
        // 设置信道为 confirm 模式
        channel.confirmSelect();
        
        // 声明队列
        channel.queueDeclare("publish.confirm.queue2", true, false, false, null);
        
        long start = System.currentTimeMillis();
        
        // 发送消息
        int batchSize = 100;
        int flag = 0;
        for (int i = 0; i < 200; i++) {
            String msg = "Publishing Messages in Batches " + i;
            channel.basicPublish("", "publish.confirm.queue2", null, msg.getBytes());
            
            // 批量 等待确认
            if (flag == batchSize) {
                channel.waitForConfirmsOrDie(5000);
                flag = 0;
            }
            flag++;
        }
        if (flag > 0) {
            channel.waitForConfirmsOrDie(5000);
        }
        
        long end = System.currentTimeMillis();
        System.out.println("Publishing Messages in Batches(批量确认) 发送 200 条消息耗时 " + (end - start));
    }
}

相比于单独确认策略，批量确认极大地提升了 confirm 的效率， 缺点是出现 Basic.Nack 或者超时时，我们不清楚具体哪条消息出了问题。客户端需要将这一批次的消息全部重发，这会带来明显的重复消息数量。 当消息经常丢失时，批量确认的性能应该是不升反降的。

6.3 Handling Publisher Confirms Asynchronously（异步确认）

提供一个回调方法，服务端确认了一条或者多条消息后客户端会回这个方法进行处理。

异步 confirm 方法的编程实现最为复杂。Channel 接口提供了一个方法 addConfirmListener，这个方法可以添加 ConfirmListener 回调接口。

ConfirmListener 接口中包含两个方法：handleAck(long deliveryTag, boolean multiple) 和 handleNack(long deliveryTag, boolean multiple)，分别对应处理 RabbitMQ 发送给生产者的 ack 和 nack。

deliveryTag 表示发送消息的序号，multiple 表示是否批量确认。我们需要为每一个 Channel 维护一个已发送消息的序号集合。 当收到 RabbitMQ 的 confirm 回调时，从集合中删除对应的消息。当 Channel 开启 confirm 模式后，channel 上发送消息都会附带一个从 1 开始递增的 deliveryTag 序号。

我们可以使用 SortedSet 的有序性来维护这个已发消息的集合。

1. 当收到 ack 时，从序列中删除该消息的序号。如果为批量确认消息，表示小于等于当前序号 deliveryTag 的消息都收到了，则清除对应集合
2. 当收到 nack 时，处理逻辑类似，不过需要结合具体的业务情况，进行消息重发等操作。

private static void handlingPublisherConfirmsAsynchronously() throws IOException, TimeoutException, InterruptedException {
    try (Channel channel = createChannel()) {
        // 设置信道为 confirm 模式
        channel.confirmSelect();
        
        // 声明队列
        channel.queueDeclare("publish.confirm.queue3", true, false, false, null);
        
        // 有序集合，元素按照自然顺序进行排序，存储未 confirm 消息序号
        SortedSet<Long> confirmSet = Collections.synchronizedSortedSet(new TreeSet<>());
        
        long start = System.currentTimeMillis();
        
        // 监听
        channel.addConfirmListener(new ConfirmListener() {
            @Override
            public void handleAck(long deliveryTag, boolean multiple) throws IOException {
                // 批量确认：将集合中小于等于当前序号 deliveryTag 元素的集合清除，表示这批序号的消息都已经被 ack 了
                if (multiple) {
                    confirmSet.headSet(deliveryTag + 1).clear();
                } else {
                    // 单独确认
                    confirmSet.remove(deliveryTag);
                }
            }

            @Override
            public void handleNack(long deliveryTag, boolean multiple) throws IOException {
                // 批量确认：将集合中小于等于当前序号 deliveryTag 元素的集合清除，表示这批序号的消息都已经被 ack 了
                if (multiple) {
                    confirmSet.headSet(deliveryTag + 1).clear();
                } else {
                    // 单独确认
                    confirmSet.remove(deliveryTag);
                }
                // 根据业务处理
            }
        });
        
        // 发送消息
        int flag = 0;
        for (int i = 0; i < 200; i++) {
            String msg = "Handling Publisher Confirms Asynchronously " + i;
            // 得到下次发送消息的序号，从 1 开始
            long nextPublishSeqNo = channel.getNextPublishSeqNo();
            channel.basicPublish("", "publish.confirm.queue3", null, msg.getBytes());
            // 存入集合
            confirmSet.add(nextPublishSeqNo);
        }
        while (confirmSet.isEmpty()) {
            Thread.sleep(20);
        }
        
        long end = System.currentTimeMillis();
        System.out.println("Handling Publisher Confirms Asynchronously(异步确认) 发送 200 条消息耗时 " + (end - start));
    }
}