2、Rabbitmq消费者确认机制

旧城等待， 2022-01-13 07:27 226阅读 0赞

[2019独角兽企业重金招聘Python工程师标准>>> ][2019_Python_] ![hot3.png][]

从上一篇文章的基本消息模型案例中，我们发现消息一旦被消费者接收，队列中的消息就会被删除。

那么问题来了：RabbitMQ怎么知道消息被接收了呢？

如果消费者领取消息后，还没执行操作就挂掉了呢？或者抛出了异常？消息消费失败，但是RabbitMQ无从得知，这样消息就丢失了！

因此，RabbitMQ有一个ACK机制。当消费者获取消息后，会向RabbitMQ发送回执ACK，告知消息已经被接收。不过这种回执ACK分两种情况：

*  自动ACK：消息一旦被接收，消费者自动发送ACK
 *  手动ACK：消息接收后，不会发送ACK，需要手动调用

在基本消息模型案例的代码中有一段

// 监听队列，第二个参数：是否自动进行消息确认。
            channel.basicConsume(QUEUE_NAME, true, consumer);

![e4416dd903a89cd396223bafccfbcb070a4.jpg][]

当autoAck=true时，一旦消费者接收到了消息，就视为自动确认了消息，rabbitmq就删除了该条消息

自动ack存在的问题：

生产者不做任何修改，直接运行，可以看到一条消息存在队列中：

![9ff664da69f1ec1cd0bf0c6595e032d4195.jpg][]

消费者代码，模拟了一个异常：

/**
     * 消费者
     */
    @Slf4j
    public class ConsumerError {
        private final static String QUEUE_NAME = "hello";
    
        public static void main(String[] argv) throws Exception {
            // 获取到连接
            Connection connection = ConnectionUtil.getConnection();
            // 创建一个信道，意味着每个线程单独一个信道
            Channel channel = connection.createChannel();
            // 声明队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
            // 定义队列的消费者
            DefaultConsumer consumer = new DefaultConsumer(channel) {
                // 获取消息，并且处理，这个方法类似事件监听，如果有消息的时候，会被自动调用
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties,
                                           byte[] body) throws IOException {
                    // body 即消息体
                    String msg = new String(body);
                    //异常模拟
                    int i = 1/0;
                    log.debug("消费消息:{}",msg);
                }
            };
            // 监听队列，第二个参数：是否自动进行消息确认。
            channel.basicConsume(QUEUE_NAME, true, consumer);
        }
    }

运行消费者

![f1d14488595784a3395948558b69d74a680.jpg][]

实际上并没有消费成功消息，队列中的消息却删除了。

![c7382dd122a3307fe25232dd36b0e7adf28.jpg][]

手动确认消息

/**
     * 消费者
     */
    @Slf4j
    public class Consumer {
        private final static String QUEUE_NAME = "hello";
    
        public static void main(String[] argv) throws Exception {
            // 获取到连接
            Connection connection = ConnectionUtil.getConnection();
            // 创建一个信道，意味着每个线程单独一个信道
            Channel channel = connection.createChannel();
            // 声明队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
            // 定义队列的消费者
            DefaultConsumer consumer = new DefaultConsumer(channel) {
                // 获取消息，并且处理，这个方法类似事件监听，如果有消息的时候，会被自动调用
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties,
                                           byte[] body) throws IOException {
                    // body 即消息体
                    String msg = new String(body);
                    int i = 1/0;
                    log.debug("消费消息:{}",msg);
                    // 手动进行ACK
    
                    channel.basicAck(envelope.getDeliveryTag(), false);
                }
            };
            // 监听队列，第二个参数：是否自动进行消息确认。
            channel.basicConsume(QUEUE_NAME, false, consumer);
        }
    }

运行生产者，生产一条消息。

![ccd953876022ac7d1a0bdf51ba346faa67d.jpg][]

运行消费者，出现异常

![6fc8e70bf134383f6bba083e4fb7688c638.jpg][]

查看管理界面，发现该条消息还存在。

![9fbccaf2ac71014de990d8b03fa44b8e4cf.jpg][]

将异常去掉，运行。

![6f7efca9a2389e9e00e10a1557fdf0ffbca.jpg][]

控制台打印出日志

![d44bd8ca55abdf21752366c4340cc9c520c.jpg][]

查看管理界面，发现消息已经被消费。

![93f8bf3dfcb395437b1e3d933cd42b0b90d.jpg][]

详细源码地址

[https://github.com/suzhe2018/rabbitmq-item][https_github.com_suzhe2018_rabbitmq-item]

![8d6d4cfaf36b9127309d40b706da14b6935.jpg][]

转载于:https://my.oschina.net/suzheworld/blog/3002275

[2019_Python_]: https://my.oschina.net/u/2663968/blog/3061697
[hot3.png]: /images/20220113/8dcdf524d81844978bae1a248e603053.png
[e4416dd903a89cd396223bafccfbcb070a4.jpg]: /images/20220113/1edcd100d3a04215bd22ff7a466f1410.png
[9ff664da69f1ec1cd0bf0c6595e032d4195.jpg]: /images/20220113/f1a2659e6b5340ceb4ea5cf1859d0fb3.png
[f1d14488595784a3395948558b69d74a680.jpg]: /images/20220113/fee474ba44034ee1ab344938a23f9db5.png
[c7382dd122a3307fe25232dd36b0e7adf28.jpg]: /images/20220113/9d85975a249548d1a6f2e4e0bac67c89.png
[ccd953876022ac7d1a0bdf51ba346faa67d.jpg]: /images/20220113/fc9a45126b63422ca76c876ac019aef5.png
[6fc8e70bf134383f6bba083e4fb7688c638.jpg]: /images/20220113/45c54496df5240038f535aae75d5c426.png
[9fbccaf2ac71014de990d8b03fa44b8e4cf.jpg]: /images/20220113/e047d380465647a58dcf30b802d6090a.png
[6f7efca9a2389e9e00e10a1557fdf0ffbca.jpg]: /images/20220113/6f61e34035684d5dae0477c8e5e23fb9.png
[d44bd8ca55abdf21752366c4340cc9c520c.jpg]: /images/20220113/3300655c06f249d68c4de40430c90dda.png
[93f8bf3dfcb395437b1e3d933cd42b0b90d.jpg]: /images/20220113/33137d1d60454ba19c7ab51d42e19e6c.png
[https_github.com_suzhe2018_rabbitmq-item]: https://github.com/suzhe2018/rabbitmq-item
[8d6d4cfaf36b9127309d40b706da14b6935.jpg]: /images/20220113/9fd45d592849407c817df34088e232d9.png