四、发布确认

news2025/3/10 5:24:20

1、发布确认原理

生产者将信道设置成 confirm 模式，一旦信道进入 confirm 模式，所有在该信道上面发布的消息都将会被指派一个唯一的 ID(从 1 开始)，一旦消息被投递到所有匹配的队列之后，broker就会发送一个确认给生产者(包含消息的唯一 ID)，这就使得生产者知道消息已经正确到达目的队列了;

如果消息和队列是可持久化的，那么确认消息会在将消息写入磁盘之后发出，broker 回传给生产者的确认消息中 delivery-tag 域包含了确认消息的序列号，此外 broker 也可以设置basic.ack 的 multiple 域，表示到这个序列号之前的所有消息都已经得到了处理kao

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

2、发布确认策略

2.1 开启发布确认的方法

发布确认模式时没有开启的，如果需要开启，则需要在channel上调用 confirmSelect() 方法

//开启发布确认
channel.confirmSelect();

2.2 单个发布确认

这是一种简单的确认方式，它是一种同步确认发布的方式，也就是发布一个消息之后只有它被确认发布，后续的消息才能继续发布。

waitForConfirmsOrDie(long)这个方法只有在消息被确认的时候才返回，如果在指定时间范围内这个消息没有被确认那么它将抛出异常。
（也就是说调用 channel.waitForConfirms() 后，信道会一直等待消息进行确认后才会返回true，否则一直阻塞，直到超时发生异常）

这种确认方式有一个最大的缺点就是:发布速度特别的慢，因为如果没有确认发布的消息就会阻塞所有后续消息的发布，这种方式最多提供每秒不超过数百条发布消息的吞吐量。当然对于某些应用程序来说这可能已经足够了。

//1、单个确认模式
    public static void pulishMessageSingleConfirm() throws Exception {
        try (Channel channel = RabbitMqUtils.getChannel();) {
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
            //开启发布确认模式
            channel.confirmSelect();
            long begin = System.currentTimeMillis();
            for (int i = 0; i < MESSAGE_COUNT; i++) {
                String message = i + "";
                channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
                boolean flag = channel.waitForConfirms();
                if (flag) {
                    System.out.println("消息发送成功");
                }
            }
            long end = System.currentTimeMillis();
            System.out.println("发布" + MESSAGE_COUNT + "个单独确认消息，耗时" + (end - begin) + "ms");
        }
    }

2.3 批量发布确认

上面那种方式非常慢，与单个等待确认消息相比，先发布一批消息然后一起确认可以极大地提高吞吐量，当然这种方式的缺点就是:当发生故障导致发布出现问题时，不知道是哪个消息出现问题了，我们必须将整个批处理保存在内存中，以记录重要的信息而后重新发布消息。当然这种方案仍然是同步的，也一样阻塞消息的发布

思路：攒100个消息一起进行发布，当发布完第一百个消息时，监听消息是否被确认

 //2、批量发布确认模式
    public static void pulishMessageBatchConfirm() throws Exception {
        try (Channel channel = RabbitMqUtils.getChannel();) {
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
            //开启发布确认模式
            channel.confirmSelect();
            long begin = System.currentTimeMillis();
            //生产者每次发布100个消息，确认一次
            int batchConfirmSize = 0;
            for (int i = 0; i < MESSAGE_COUNT; i++) {
                String message = i + "";
                channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
                batchConfirmSize++;
                if (batchConfirmSize % 100 == 0) {
                    boolean flag = channel.waitForConfirms();
                    if (flag) {
                        System.out.println("消息发送成功");
                    }
                    batchConfirmSize = 0;
                }
            }
            long end = System.currentTimeMillis();
            System.out.println("发布" + MESSAGE_COUNT + "个批量确认消息，耗时" + (end - begin) + "ms");
        }
    }

2.4 异步发布确认

维护一个所有已发布消息的map，通过回调函数传递回来当前确认的消息，然后从map中移除掉已经确认的消息，剩下的就是已经发布但是没有确认的消息
在这里插入图片描述

//3、异步发布确认模式
    public static void pulishMessageAsyncConfirm() throws Exception {
        try (Channel channel = RabbitMqUtils.getChannel();) {
            //创建一个队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
            //开启发布确认
            channel.confirmSelect();
            /**
             * 线程安全有序的一个哈希表，适用于高并发的情况
             * 1、轻松的将序号和消息进行关联
             * 2、轻松的批量删除条目  只需要给到序列号
             * 3、支持并发访问
             * */
            ConcurrentSkipListMap<Long, String> outstandingConfirms = new ConcurrentSkipListMap<>();
            /**
             * 确认收到消息的一个回调
             * 1、参数1：当前收到的消息的序列号
             * 2、参数2：是否批量确认
             */
            ConfirmCallback ackCallBack = (sequenceNumber, mutiple) -> {
                if (mutiple) {
                    System.out.println("生产者发布的消息" + outstandingConfirms.get(sequenceNumber) + "被确认，序列号" + sequenceNumber);
                    //返回的是小于等于当前序列号的未确认消息，是一个map
                    ConcurrentNavigableMap<Long, String> confirmed = outstandingConfirms.headMap(sequenceNumber, true);
                    //消除该部分未确认消息
                    confirmed.clear();
                } else {
                    System.out.println("生产者发布的消息" + outstandingConfirms.get(sequenceNumber) + "被确认，序列号" + sequenceNumber);
                    //只消除当前序号的消息
                    outstandingConfirms.remove(sequenceNumber);
                }

            };
            //消息未确认的回调
            ConfirmCallback nackCallBack = (sequenceNumber, mutiple) -> {
                String message = outstandingConfirms.get(sequenceNumber);
                System.out.println("生产者发布的消息" + message + "未被确认，序列号" + sequenceNumber);
            };
            /*
              添加一个异步确认的监听器
                1、确认收到消息的回调
                2、未收到消息的回调
             */
            channel.addConfirmListener(ackCallBack, nackCallBack);
            long begin = System.currentTimeMillis();
            for (int i = 0; i < MESSAGE_COUNT; i++) {
                String message = i + "";
                outstandingConfirms.put(channel.getNextPublishSeqNo(), message);
                channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
            }
            long end = System.currentTimeMillis();
            System.out.println("发布" + MESSAGE_COUNT + "个异步确认消息，耗时" + (end - begin) + "ms");
        }
    }