SpringBoot RabbitMQ ACK 机制

女爷i 2023-10-11 10:04 8阅读 0赞

### 文章目录 ###

*  SpringBoot RabbitMQ ACK 机制
 *   *  Rabbit ACK 机制流程图
     *  实现步骤
     *   *  application.yml 配置
         *  消息生产端发送确认配置
         *  消息消费关消费消息确认

# SpringBoot RabbitMQ ACK 机制 #

## Rabbit ACK 机制流程图 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L215TmFtZUlzc2xz_size_16_color_FFFFFF_t_70]

## 实现步骤 ##

### application.yml 配置 ###

spring:
      rabbitmq:
        host: localhost
        port: 5672
        username: guest
        password: guest
        # publisher-confirms和publisher-returns是对于消息生产端的配置
        publisher-confirms: true # 开启发送消息确认 对应RabbitTemplate.ConfirmCallback接口
        publisher-returns: true  # 开启发送消息失败返回 对应RabbitTemplate.ReturnCallback接口
    
        # 这个配置是针对消息消费端的配置
        listener:
          direct:
            acknowledge-mode: manual # 开启 ack 手动确认
          simple:
            acknowledge-mode: manual # 开启 ack 手动确认

### 消息生产端发送确认配置 ###

import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.amqp.core.Message;
    import org.springframework.amqp.rabbit.connection.CorrelationData;
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.context.annotation.Configuration;
    
    import javax.annotation.PostConstruct;
    
    @Configuration
    public class RabbitTemplateConfig implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnCallback{
    
        Logger logger = LoggerFactory.getLogger(this.getClass());
    
        @Autowired
        private RabbitTemplate rabbitTemplate;
    
        @PostConstruct
        public void init(){
            rabbitTemplate.setConfirmCallback(this);            // 指定 ConfirmCallback
            rabbitTemplate.setReturnCallback(this);             // 指定 ReturnCallback
        }
    
    
        /**
         * 如果消息到达 exchange, 则 confirm 回调, ack = true
         * 如果消息不到达 exchange, 则 confirm 回调, ack = false
         * 需要设置spring.rabbitmq.publisher-confirms=true
         * @param correlationData
         * @param ack
         * @param cause
         */
        @Override
        public void confirm(CorrelationData correlationData, boolean ack, String cause) {
    
            logger.info("消息是否到达Exchange:{}", ack == true ? "消息成功到达Exchange" : "消息到达Exchange失败");
            if (!ack) {
                logger.info("消息到达Exchange失败原因:{}", cause);
    
                // 根据业务逻辑实现消息补偿机制
    
            }
    
        }
    
        /**
         * exchange 到达 queue, 则 returnedMessage 不回调
         * exchange 到达 queue 失败, 则 returnedMessage 回调
         * 需要设置spring.rabbitmq.publisher-returns=true
         * @param message
         * @param replyCode
         * @param replyText
         * @param exchange
         * @param routingKey
         */
        @Override
        public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
            logger.info("消息报文:{}", new String(message.getBody()));
            logger.info("消息编号:{}", replyCode);
            logger.info("描述:{}", replyText);
            logger.info("交换机名称:{}", exchange);
            logger.info("路由名称:{}", routingKey);
    
            // 根据业务逻辑实现消息补偿机制
    
        }
    }

### 消息消费关消费消息确认 ###

@RabbitListener(queues = {Constant.QUEUE_NAME_TEST})
    public void queueTestConsumer(RabbitObject rabbitObject, Channel channel, Message message){
    
        String msg = new String(message.getBody());
    
        try {
    
            // 当前线程休眠5秒钟，模拟消息消费过程
            Thread.sleep(5000);
            // 正常消费消息
            logger.info("-------- 消息-id:{}", rabbitObject.toString());
    
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            logger.error("消息[{}]确认处理成功", msg);
    
    
        } catch (Exception e) {
    
            // redelivered = true, 表明该消息是重复处理消息
            Boolean redelivered = message.getMessageProperties().getRedelivered();
    
            /**
             * 这里对消息重入队列做设置，例如将消息序列化缓存至 Redis, 并记录重入队列次数
             * 如果该消息重入队列次数达到一次次数，比如3次，将不再重入队列，直接拒绝
             * 这时候需要对消息做补偿机制处理
             *
             * channel.basicNack与channel.basicReject要结合越来使用
             *
             */
            try {
    
                if (redelivered) {
    
                    /**
                     * 1. 对于重复处理的队列消息做补偿机制处理
                     * 2. 从队列中移除该消息，防止队列阻塞
                     */
                    // 消息已重复处理失败, 扔掉消息
                    channel.basicReject(message.getMessageProperties().getDeliveryTag(), false); // 拒绝消息
                    logger.error("消息[{}]重新处理失败，扔掉消息", msg);
                }
    
                // redelivered != true,表明该消息是第一次消费
                if (!redelivered) {
    
                    // 消息重新放回队列
                    channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
                    logger.error("消息[{}]处理失败，重新放回队列", msg);
                }
    
            } catch (IOException e1) {
                e1.printStackTrace();
            }
    
    
            e.printStackTrace();
        }
    }

*  channel.basicAck: 表示消息正常消费
 *  channel.basicNack: 消息消费异常，将消息重新往往队列
 *  channel.basicReject: 消息消费异常，拒绝该消息入队列

博客地址：  
[SpringBoot 企业级应用实战][SpringBoot]  
[基于 SpringBoot 集成 RabbitMQ][SpringBoot _ RabbitMQ]  
源码地址：  
[基于 SpringBoot 集成 RabbitMQ][SpringBoot _ RabbitMQ 1]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L215TmFtZUlzc2xz_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/2019092216543761.jpg?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L215TmFtZUlzc2xz,size_16,color_FFFFFF,t_70
[SpringBoot]: https://blog.csdn.net/myNameIssls/article/details/80628768
[SpringBoot _ RabbitMQ]: https://blog.csdn.net/myNameIssls/article/details/100185729
[SpringBoot _ RabbitMQ 1]: https://github.com/myNameIssls/springboot-study/tree/master/springboot-rabbitmq