SpringBoot使用RabbitMQ的ACK机制

秒速五厘米 2023-10-01 19:49 23阅读 0赞

**什么是ACK消息确认？**

在使用MQ消息队列过程中，生产者发送消息到mq队列中，消费者监听这个队列，并拿取消息进行消费使用。但如果消费者因为在处理业务逻辑过程中，发生了非业务逻辑的异常(如断电、服务崩了等)这些，那么这条消息就没有完成消费，最后导致数据丢失。所以为了确保数据不会丢失，MQ支持自定义ACK消息确实机制。

**ACK消息确实机制**

ACK确实机制是消费者从队列中收到消息，进行业务逻辑的处理，在这一过程中如果消费者出现服务器异常、网络不稳定等，都不会有ACK应答，这时候RabbitMQ就认为这条消息没有正常消费成功，就会将消息重新放回队列中。直到消费者发送ACK应答，这条消息才会从队列中消费掉。

注意：如果Consumer忘记ACK应答，那么当Consumer退出的时候，消息还是会一直重新分发，最后导致RabbitMQ占用会越来越多，可能会因此导致内存泄漏风险，所以应结合自己项目是否需要使用，做好一些应对机制。

### 1、引入pom.xml依赖 ###

<dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-amqp</artifactId>
        <version>2.2.10.RELEASE</version>
    </dependency>

### 2、配置文件 ###

spring
      # rabbitMQ消息中间件
      rabbitmq:
        host: MQ服务IP地址
        port: 5672
        username: 用户名
        password: 密码
        virtual-host: /
        listener:
          simple:
            retry:
              # 是否开启重试
              enabled: true
              # 重试时间间隔 ms
              initial-interval: 3000
              # 重试次数
              max-attempts: 3
              # 重试最大时间间隔 ms
              max-interval: 15000
              # 倍数
              multiplier: 2 
            # 注意：开启手动Ack
            acknowledge-mode: manual

### 3、MQ常量类 ###

public class RabbitMQConstant {
        
    
        public static final String EXCHANGE_NAME = "delay_exchange";
    
        public static final String QUEUE_NAME = "one_queue";
    
    }

### 4、配置类 ###

import me.zhengjie.rabbitmq.common.RabbitMQConstant;
    import me.zhengjie.rabbitmq.common.TopicEnum;
    import org.springframework.amqp.core.*;
    import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
    import org.springframework.amqp.support.converter.MessageConverter;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    import java.util.HashMap;
    import java.util.Map;
    
    @Configuration
    public class RabbitMQConfig {
        
    
        /**
         * 定义一个交换机
         */
        @Bean
        public CustomExchange customExchange() {
        
            return new DirectExchange(RabbitMQConstant.EXCHANGE_NAME, true, false);
        }
    
        /**
         * 创建队列
         */
        @Bean()
        public Queue createOneQueue() {
        
            return QueueBuilder.durable(RabbitMQConstant.QUEUE_NAME).build();
        }
    
        /**
         * 绑定路由与队列
         */
        @Bean
        public Binding exchangeBindOneQueue() {
        
            return BindingBuilder.bind(createSendQueue()).to(TestDirectExchange()).with(RabbitMQConstant.ROUTER_NAME);
        }
        
        /**
         * 序列化配置
         */
        @Bean
        public MessageConverter messageConverter() {
        
            return new Jackson2JsonMessageConverter();
        }
    }

### 5、生产消息类 ###

import com.zhf.rabbitmq.constants.RabbitMQConstant;
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.stereotype.Component;
    
    @Component
    public class RabbitProducer {
        
    
        private final RabbitTemplate rabbitTemplate;
    
        public RabbitProducer(RabbitTemplate rabbitTemplate) {
        
            this.rabbitTemplate = rabbitTemplate;
        }
    
        /**
         * 发送消息
         *
         * @param obj 消息对象
         * @param routingKey 消息队列一些参数
         */
        public void sendMsg(Object obj, String routingKey) {
        
            rabbitTemplate.convertAndSend(RabbitMQConstant.EXCHANGE_NAME, routingKey, obj, (message) -> {
        
                message.getMessageProperties();
                return message;
            });
        }
    
    }

### 6、消费者 ###

import com.zhf.rabbitmq.constants.RabbitMQConstant;
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.amqp.support.AmqpHeaders;
    import com.rabbitmq.client.Channel;
    import org.springframework.stereotype.Service;
    
    @Service
    public class Consumer {
        
    
        /**
         * 消费者
         * 
         * @param message 接收到的消息内容
         * @param channel 信道
         * @param tag 标记当前消息的tag，用于ACK
         */
        @RabbitListener(queues = RabbitMQConstant.QUEUE_NAME)
        public void myAckListener(String message, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long tag) throws IOException {
        
            try {
        
                // 业务逻辑处理成功，告诉RabbitMQ，已经接收到消息并做了处理了。这样消息队列这条消息才算真正消费成功
                channel.basicAck(tag, false);
            } catch (Exception e) {
        
                // 处理过程中，发生了不是业务逻辑的错误异常，则不答复ACK，这样MQ会认为这条消息未成功消费，所以会重新把该条消息放回队列中，直到ACK正常答复
                channel.basicNack(tag, false, true);
            }
        }
    }