RabbitMQ消息确认机制之publisher confirm

分手后的思念是犯贱 2021-07-24 11:52 493阅读 0赞

**一 Confirm模式**

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

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

在channel 被设置成 confirm 模式之后，所有被 publish 的后续消息都将被 confirm（即 ack） 或者被nack一次。但是没有对消息被 confirm 的快慢做任何保证，并且同一条消息不会既被 confirm又被nack 。

**二 编程模式**

客户端实现生产者confirm有三种编程方式：

1 普通confirm模式：每发送一条消息后，调用waitForConfirms()方法，等待服务器端confirm。实际上是一种串行confirm了。

2 批量confirm模式：每发送一批消息后，调用waitForConfirms()方法，等待服务器端confirm。

3 异步confirm模式：提供一个回调方法，服务端confirm了一条或者多条消息后Client端会回调这个方法。

**三 生产端代码**

1 普通confirm模式

/**
    * Copyright (C), 2020-2020, 软件公司
    * FileName: ConfirmSend
    * Author:   cakin
    * Date:     2020/4/25
    * Description: 生产者：普通confirm模式
    */
    package com.rabbitmq.productconfirm;
    
    
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.util.ConnectionUtils;
    import lombok.extern.slf4j.Slf4j;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    /**
    * @ClassName: ConfirmSend
    * @Description: 生产者：普通confirm模式
    * @Date: 2020/4/25
    * @Author: cakin
    */
    @Slf4j
    public class ConfirmSend {
        /**
         * 队列名称
         */
        private static final String QUEUE_NAME = "test_queue_confirm1";
    
        public static void main( String[] args ) throws IOException, TimeoutException, InterruptedException {
            // 获取连接
            Connection connection = ConnectionUtils.getConnection();
            // 从连接开一个通道
            Channel channel = connection.createChannel();
            // 声明一个队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
            channel.confirmSelect(); // 开始confirm模式
    
            // 发送消息
            String message = "hello, confirm message";
            channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
    
            if (channel.waitForConfirms()) {
                log.info(" [x] Sent message : '" + message + "' ok ");
            } else {
                log.info(" [x] Sent message : '" + message + "' fail ");
            }
    
            channel.close();
            connection.close();
        }
    }

2 批量confirm模式

/**
    * Copyright (C), 2020-2020, 软件公司
    * FileName: ConfirmSend2
    * Author:   cakin
    * Date:     2020/4/25
    * Description: 生产者：批量confirm模式
    */
    package com.rabbitmq.productconfirm;
    
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.util.ConnectionUtils;
    import lombok.extern.slf4j.Slf4j;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    /**
    * @ClassName: ConfirmSend2
    * @Description: 生产者：批量confirm模式
    * @Date: 2020/4/25
    * @Author: cakin
    */
    @Slf4j
    public class ConfirmSend2 {
        /**
         * 队列名称
         */
        private static final String QUEUE_NAME = "test_queue_confirm1";
        /**
         * 常量20
         */
        private static final int NUM20 = 20;
    
        public static void main( String[] args ) throws IOException, TimeoutException, InterruptedException {
            // 获取连接
            Connection connection = ConnectionUtils.getConnection();
            // 从连接开一个通道
            Channel channel = connection.createChannel();
            // 声明一个队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    
            channel.confirmSelect(); // 开始confirm模式
    
            for (int i = 0; i < NUM20; i++) {
                // 发送消息
                String message = "hello, confirm message " + i;
                channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
            }
    
            if (channel.waitForConfirms()) {
                log.info(" [x] Sent message ok ");
            } else {
                log.info(" [x] Sent message fail ");
            }
    
            channel.close();
            connection.close();
        }
    }

3 异步confirm模式

/**
    * Copyright (C), 2020-2020, 软件公司
    * FileName: ConfirmSend3
    * Author:   cakin
    * Date:     2020/4/25
    * Description: 生产者：异步confirm模式
    */
    package com.rabbitmq.productconfirm;
    
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.ConfirmListener;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.util.ConnectionUtils;
    import lombok.extern.slf4j.Slf4j;
    
    import java.io.IOException;
    import java.util.Collections;
    import java.util.SortedSet;
    import java.util.TreeSet;
    import java.util.concurrent.TimeoutException;
    
    /**
    * @ClassName: ConfirmSend3
    * @Description: 生产者：异步confirm模式
    * @Date: 2020/4/25
    * @Author: cakin
    */
    @Slf4j
    public class ConfirmSend3 {
        /**
         * 队列名
         */
        private static final String QUEUE_NAME = "test_queue_confirm1";
        /**
         * 常量
         */
        private static final int NUM20 = 20;
    
        public static void main( String[] args ) throws IOException, TimeoutException, InterruptedException {
            // 获取连接
            Connection connection = ConnectionUtils.getConnection();
            // 从连接开一个通道
            Channel channel = connection.createChannel();
            // 声明一个队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    
            channel.confirmSelect(); // 开始confirm模式
    
            // 未确认的消息列表
            SortedSet<Long> confirmSet = Collections.synchronizedSortedSet(new TreeSet<Long>());
    
            channel.addConfirmListener(new ConfirmListener() {
                // 有效的应答
                public void handleAck( long deliveryTag, boolean multiple ) throws IOException {
                    if (multiple) {
                        confirmSet.headSet(deliveryTag + 1).clear();
                    } else {
                        confirmSet.remove(deliveryTag);
                    }
                }
    
                // 有问题的应答
                public void handleNack( long deliveryTag, boolean multiple ) throws IOException {
                    log.info("Nack, SeqNo: " + deliveryTag + ", multiple: " + multiple);
                    // TODO 处理未确认的应答
                    if (multiple) {
                        confirmSet.headSet(deliveryTag + 1).clear();
                    } else {
                        confirmSet.remove(deliveryTag);
                    }
                }
            });
    
            for (int i = 0; i < NUM20; i++) {
                long nextSeqNo = channel.getNextPublishSeqNo();
                String message = "hello, confirm message " + i;
                channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
                confirmSet.add(nextSeqNo);
            }
        }
    }

**四 消费端代码**

/**
    * Copyright (C), 2020-2020, 软件公司
    * FileName: ConfirmRecv
    * Author:   cakin
    * Date:     2020/4/25
    * Description: 消费者
    */
    package com.rabbitmq.productconfirm;
    
    import com.rabbitmq.client.AMQP;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.DefaultConsumer;
    import com.rabbitmq.util.ConnectionUtils;
    import com.rabbitmq.client.Envelope;
    import lombok.extern.slf4j.Slf4j;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    /**
    * @ClassName: ConfirmRecv
    * @Description: 消费者
    * @Date: 2020/4/25
    * @Author: cakin
    */
    @Slf4j
    public class ConfirmRecv {
        /**
         * 消费者
         */
        private static final String QUEUE_NAME = "test_queue_confirm1";
    
        public static void main( String[] args ) throws IOException, TimeoutException {
            // 获取连接
            Connection connection = ConnectionUtils.getConnection();
    
            // 打开通道
            Channel channel = connection.createChannel();
    
            // 申明要消费的队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    
            // 消费消息
            channel.basicConsume(QUEUE_NAME, true, new DefaultConsumer(channel) {
                @Override
                public void handleDelivery( String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body ) throws IOException {
                    // 接收到的消息
                    String message = new String(body);
                    log.info(" [x] Received '" + message + "'");
                }
            });
        }
    }

**五 测试结果**

1 普通confirm模式

生产端

16:16:46.032 [main] INFO com.rabbitmq.productconfirm.ConfirmSend -  [x] Sent message : 'hello, confirm message' ok

消费端

16:16:46.040 [pool-1-thread-4] INFO com.rabbitmq.productconfirm.ConfirmRecv -  [x] Received 'hello, confirm message'

2 批量confirm模式

生产端

16:23:28.176 [main] INFO com.rabbitmq.productconfirm.ConfirmSend2 -  [x] Sent message ok

消费端

16:23:28.169 [pool-1-thread-7] INFO com.rabbitmq.productconfirm.ConfirmRecv -  [x] Received 'hello, confirm message 0'
    16:23:28.172 [pool-1-thread-8] INFO com.rabbitmq.productconfirm.ConfirmRecv -  [x] Received 'hello, confirm message 1'
    16:23:28.172 [pool-1-thread-8] INFO com.rabbitmq.productconfirm.ConfirmRecv -  [x] Received 'hello, confirm message 2'
    16:23:28.173 [pool-1-thread-8] INFO com.rabbitmq.productconfirm.ConfirmRecv -  [x] Received 'hello, confirm message 3'
    16:23:28.173 [pool-1-thread-8] INFO com.rabbitmq.productconfirm.ConfirmRecv -  [x] Received 'hello, confirm message 4'
    16:23:28.173 [pool-1-thread-8] INFO com.rabbitmq.productconfirm.ConfirmRecv -  [x] Received 'hello, confirm message 5'
    16:23:28.173 [pool-1-thread-8] INFO com.rabbitmq.productconfirm.ConfirmRecv -  [x] Received 'hello, confirm message 6'
    16:23:28.173 [pool-1-thread-8] INFO com.rabbitmq.productconfirm.ConfirmRecv -  [x] Received 'hello, confirm message 7'
    16:23:28.173 [pool-1-thread-8] INFO com.rabbitmq.productconfirm.ConfirmRecv -  [x] Received 'hello, confirm message 8'
    16:23:28.173 [pool-1-thread-8] INFO com.rabbitmq.productconfirm.ConfirmRecv -  [x] Received 'hello, confirm message 9'
    16:23:28.174 [pool-1-thread-9] INFO com.rabbitmq.productconfirm.ConfirmRecv -  [x] Received 'hello, confirm message 10'
    16:23:28.174 [pool-1-thread-9] INFO com.rabbitmq.productconfirm.ConfirmRecv -  [x] Received 'hello, confirm message 11'
    16:23:28.174 [pool-1-thread-9] INFO com.rabbitmq.productconfirm.ConfirmRecv -  [x] Received 'hello, confirm message 12'
    16:23:28.174 [pool-1-thread-9] INFO com.rabbitmq.productconfirm.ConfirmRecv -  [x] Received 'hello, confirm message 13'
    16:23:28.174 [pool-1-thread-9] INFO com.rabbitmq.productconfirm.ConfirmRecv -  [x] Received 'hello, confirm message 14'
    16:23:28.174 [pool-1-thread-9] INFO com.rabbitmq.productconfirm.ConfirmRecv -  [x] Received 'hello, confirm message 15'
    16:23:28.174 [pool-1-thread-9] INFO com.rabbitmq.productconfirm.ConfirmRecv -  [x] Received 'hello, confirm message 16'
    16:23:28.174 [pool-1-thread-9] INFO com.rabbitmq.productconfirm.ConfirmRecv -  [x] Received 'hello, confirm message 17'
    16:23:28.174 [pool-1-thread-9] INFO com.rabbitmq.productconfirm.ConfirmRecv -  [x] Received 'hello, confirm message 18'
    16:23:28.175 [pool-1-thread-10] INFO com.rabbitmq.productconfirm.ConfirmRecv -  [x] Received 'hello, confirm message 19'

3 异步confirm模式

生产端

没打印，说明消息都发成功了。

消费端

16:30:26.277 [pool-1-thread-4] INFO com.rabbitmq.productconfirm.ConfirmRecv -  [x] Received 'hello, confirm message 0'
    16:30:26.277 [pool-1-thread-4] INFO com.rabbitmq.productconfirm.ConfirmRecv -  [x] Received 'hello, confirm message 1'
    16:30:26.277 [pool-1-thread-4] INFO com.rabbitmq.productconfirm.ConfirmRecv -  [x] Received 'hello, confirm message 2'
    16:30:26.277 [pool-1-thread-4] INFO com.rabbitmq.productconfirm.ConfirmRecv -  [x] Received 'hello, confirm message 3'
    16:30:26.277 [pool-1-thread-4] INFO com.rabbitmq.productconfirm.ConfirmRecv -  [x] Received 'hello, confirm message 4'
    16:30:26.277 [pool-1-thread-4] INFO com.rabbitmq.productconfirm.ConfirmRecv -  [x] Received 'hello, confirm message 5'
    16:30:26.277 [pool-1-thread-4] INFO com.rabbitmq.productconfirm.ConfirmRecv -  [x] Received 'hello, confirm message 6'
    16:30:26.278 [pool-1-thread-4] INFO com.rabbitmq.productconfirm.ConfirmRecv -  [x] Received 'hello, confirm message 7'
    16:30:26.278 [pool-1-thread-5] INFO com.rabbitmq.productconfirm.ConfirmRecv -  [x] Received 'hello, confirm message 8'
    16:30:26.278 [pool-1-thread-5] INFO com.rabbitmq.productconfirm.ConfirmRecv -  [x] Received 'hello, confirm message 9'
    16:30:26.278 [pool-1-thread-5] INFO com.rabbitmq.productconfirm.ConfirmRecv -  [x] Received 'hello, confirm message 10'
    16:30:26.278 [pool-1-thread-5] INFO com.rabbitmq.productconfirm.ConfirmRecv -  [x] Received 'hello, confirm message 11'
    16:30:26.278 [pool-1-thread-5] INFO com.rabbitmq.productconfirm.ConfirmRecv -  [x] Received 'hello, confirm message 12'
    16:30:26.278 [pool-1-thread-5] INFO com.rabbitmq.productconfirm.ConfirmRecv -  [x] Received 'hello, confirm message 13'
    16:30:26.279 [pool-1-thread-3] INFO com.rabbitmq.productconfirm.ConfirmRecv -  [x] Received 'hello, confirm message 14'
    16:30:26.279 [pool-1-thread-3] INFO com.rabbitmq.productconfirm.ConfirmRecv -  [x] Received 'hello, confirm message 15'
    16:30:26.279 [pool-1-thread-3] INFO com.rabbitmq.productconfirm.ConfirmRecv -  [x] Received 'hello, confirm message 16'
    16:30:26.279 [pool-1-thread-3] INFO com.rabbitmq.productconfirm.ConfirmRecv -  [x] Received 'hello, confirm message 17'
    16:30:26.281 [pool-1-thread-9] INFO com.rabbitmq.productconfirm.ConfirmRecv -  [x] Received 'hello, confirm message 18'
    16:30:26.281 [pool-1-thread-9] INFO com.rabbitmq.productconfirm.ConfirmRecv -  [x] Received 'hello, confirm message 19'