RabbitMQ中的单个确认、批量确认、异步确认模式

准备代码 生产Channel的工具类

    public static Channel getChannel() {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("127.0.0.1");
        factory.setUsername("guest");
        factory.setPassword("guest");
        try {
            connection = factory.newConnection();
            channel = connection.createChannel();
        } catch (IOException e) {
            e.printStackTrace();
        } catch (TimeoutException e) {
            e.printStackTrace();
        }
        return channel;
    }

//消息确认
boolean flag = channel.waitForConfirms();

Channel channel = RabbitUtils.getChannel();
channel.queueDeclare(PUBLISH_QUEUE, true, false, false, null);
//开启发布确认
channel.confirmSelect();
long begin = System.currentTimeMillis();
for (int i = 0; i < MESSAGE_NUM; i++) {
    String message = i+"";
    channel.basicPublish("", PUBLISH_QUEUE, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes("UTF-8"));
    //单个消息马上进行确认
    boolean flag = channel.waitForConfirms();
    if (flag) {
       System.out.println("消息被确认 = " + message);
    }
}
long end = System.currentTimeMillis();
System.out.println("end-begin = " + (end - begin));

boolean flag = channel.waitForConfirms();

public static void publishMessageBatch() throws Exception {
        Channel channel = RabbitUtils.getChannel();
        channel.queueDeclare(PUBLISH_QUEUE, true, false, false, null);
        //开启发布确认
        channel.confirmSelect();
        long begin = System.currentTimeMillis();
        for (int i = 0; i < MESSAGE_NUM; i++) {
            String message = i+"";
            channel.basicPublish("", PUBLISH_QUEUE, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes("UTF-8"));

            if (i % 100 == 0) {
                boolean flag = channel.waitForConfirms();
                System.out.println("消息被确认 = " + message);
            }
        }
        long end = System.currentTimeMillis();
        System.out.println("批量确认时间 ：end-begin = " + (end - begin));
}

//监听确认回调和未确认回调
channel.addConfirmListener(/confirm/iCallback, nackCallback);

public static void publishMessageAsync() throws Exception {
        Channel channel = RabbitUtils.getChannel();
        channel.queueDeclare(PUBLISH_QUEUE, true, false, false, null);
        //开启发布确认
        channel.confirmSelect();

        //消息确认成功回调
        ConfirmCallback confirmCallback = (deliveryTag, multiple) ->{
            System.out.println("已确认的消息" + deliveryTag);
        };
        //消息确认失败回调
        //deliveryTag :  消息的标记
        //multiple : 是否批量确认
        ConfirmCallback nackCallback = (deliveryTag, multiple)->{
            System.out.println("未确认的消息" + deliveryTag);
        };
        //开启消息监听，监听哪些消息成功，哪些消息失败
        channel.addConfirmListener(/confirm/iCallback, nackCallback);

        long begin = System.currentTimeMillis();
        for (int i = 0; i < MESSAGE_NUM; i++) {
            String message = i+"";
            channel.basicPublish("", PUBLISH_QUEUE, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes("UTF-8"));
        }
        long end = System.currentTimeMillis();
        System.out.println("异步确认时间 ：end-begin = " + (end - begin));
}

 重点代码

1、创建一个ConcurrentSkipListMap存放消息的序号和消息

ConcurrentSkipListMap outstandingConfirms= new ConcurrentSkipListMap<>();

public static void publishMessageAsync() throws Exception {
        Channel channel = RabbitUtils.getChannel();
        channel.queueDeclare(PUBLISH_QUEUE, true, false, false, null);
        //开启发布确认
        channel.confirmSelect();
        
        ConcurrentSkipListMap outstandingConfirms= new ConcurrentSkipListMap<>();

        //消息确认成功回调
        ConfirmCallback confirmCallback = (deliveryTag, multiple) ->{
            if (multiple) {
                ConcurrentNavigableMap confirmed = outstanding/confirm/is.headMap(deliveryTag);
                /confirm/ied.clear();
            } else {
                outstanding/confirm/is.remove(deliveryTag);
            }
            System.out.println("已确认的消息" + deliveryTag);
        };
        //消息确认失败回调
        //deliveryTag :  消息的标记
        //multiple : 是否批量确认
        ConfirmCallback nackCallback = (deliveryTag, multiple)->{
            System.out.println("未确认的消息" + deliveryTag);
            System.out.println("outstandingConfirms = " + outstanding/confirm/is);
        };
        //开启消息监听，监听哪些消息成功，哪些消息失败
        channel.addConfirmListener(/confirm/iCallback, nackCallback);

        long begin = System.currentTimeMillis();
        for (int i = 0; i < MESSAGE_NUM; i++) {
            String message = i+"";
            channel.basicPublish("", PUBLISH_QUEUE, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes("UTF-8"));
            //outstanding/confirm/is中存放序号和消息
            outstanding/confirm/is.put(channel.getNextPublishSeqNo(), message);
        }
        long end = System.currentTimeMillis();
        System.out.println("异步确认时间 ：end-begin = " + (end - begin));
}