RabbitMQ发布确认详解

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

public class PublishingMessagesIndividually {

    public static void main(String[] args) {
        try(Channel channel = RabbitMQConfig.getChannel()) {
            //随机生成队列
            String queryName = UUID.randomUUID().toString();
            //生成队列 不持久化 不共享 不删除 参数为空
            channel.queueDeclare(queryName, false, false, false, null);
            //开启发布确认
            channel.confirmSelect();
            long startTime = System.nanoTime();
            //消息数量为10
            for (int i = 0; i < 10; i++) {
                String message = i + "";
                channel.basicPublish("", queryName, null, message.getBytes());
                if (channel.waitForConfirms()) {
                    System.out.println("消息发送成功");

                }
            }
            long endTime = System.nanoTime();
            System.out.println("发布10" + "个单独确认消息,耗时" + (endTime - startTime) +
                    "ms");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

消息发送成功
消息发送成功
消息发送成功
消息发送成功
消息发送成功
消息发送成功
消息发送成功
消息发送成功
消息发送成功
消息发送成功
发布10个单独确认消息,耗时18288200ms

public class BatchesPublishingMessages {

    public static void main(String[] args) {
        try(Channel channel = RabbitMQConfig.getChannel()) {
            String queueName = UUID.randomUUID().toString();
            channel.queueDeclare(queueName, false, false, false, null);
            //开启发布确认
            channel.confirmSelect();
            //批量确认消息大小
            int msgSize = 100;
            //初始化未确认消息个数
            int outstandingMessageCount = 0;
            long start = System.currentTimeMillis();
            for (int i = 0; i < 10; i++) {
                String message = i + "";
                channel.basicPublish("", queueName, null, message.getBytes());
                outstandingMessageCount++;
                if (outstandingMessageCount == msgSize) {
                    channel.waitForConfirms();
                    outstandingMessageCount = 0;
                }
                //为了确保还有剩余没有确认消息 再次确认
                if (outstandingMessageCount > 0) {
                    channel.waitForConfirms();
                }

            }
            long end = System.currentTimeMillis();
            System.out.println("发布10"  + "个批量确认消息,耗时" + (end - start) +
                    "ms");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

发布10个批量确认消息,耗时6ms

public class AsyncHandlingPublisherConfirms {

    public static void main(String[] args) throws Exception {
        asyncPublishMessage();
    }

    public static void asyncPublishMessage() throws Exception{
        try(Channel channel = RabbitMQConfig.getChannel()) {
            String queueName = UUID.randomUUID().toString();
            channel.queueDeclare(queueName, false, false, false, null);
            //开启发布确认
            channel.confirmSelect();
            /**
             * 线程安全有序的一个哈希表，适用于高并发的情况
             * 1.轻松的将序号与消息进行关联
             * 2.轻松批量删除条目 只要给到序列号
             * 3.支持并发访问
             */
            ConcurrentSkipListMap<Long, String> outstandingConfirms = new ConcurrentSkipListMap<>();
            ConfirmCallback callback = (sequenceNumber, multiple) ->{
                if (multiple){
                    //返回的是小于等于当前序列号的未确认消息 是一个 map
                    ConcurrentNavigableMap<Long, String> confirmed = outstandingConfirms.headMap(
                            sequenceNumber, true
                    );
                    //清除该部分未确认消息
                    confirmed.clear();
                }else {
                    //只清除当前序列号的消息
                    outstandingConfirms.remove(sequenceNumber);
                }
            };
            ConfirmCallback nackCallback = (sequenceNumber, multiple) -> {
                String message = outstandingConfirms.get(sequenceNumber);
                System.out.println("发布的消息"+message+"未被确认，序列号"+sequenceNumber);
            };
            //接收消息回调 未接收消息回调
            channel.addConfirmListener(callback, null);
            long begin = System.currentTimeMillis();
            for (int i = 0; i < 10; i++) {
                String message = "消息" + i;
                /**
                 * channel.getNextPublishSeqNo()获取下一个消息的序列号
                 * 通过序列号与消息体进行一个关联
                 * 全部都是未确认的消息体
                 */
                outstandingConfirms.put(channel.getNextPublishSeqNo(), message);
                channel.basicPublish("", queueName, null, message.getBytes());
            }
            long end = System.currentTimeMillis();
            System.out.println("发布10" + "个异步确认消息,耗时" + (end - begin) +
                    "ms");
        }
    }
}

发布10个异步确认消息,耗时4ms

1000个消息的执行时间对比
发布1000个单独确认消息,耗时561ms
发布1000个批量确认消息,耗时558ms
发布1000个异步确认消息,耗时42ms