rabbitmq reply-text=PRECONDITION_FAILED - unknown delivery tag 1

超、凢脫俗 2022-12-08 15:35 114阅读 0赞

**问题现象:RabbitMQ double ack 报错**

16:50:10.134 ERROR 17788 ---  o.s.a.r.c.CachingConnectionFactory       : 
    Channel shutdown: channel error; protocol method: #method<channel.close>(reply-code=406, reply-text=PRECONDITION_FAILED - unknown delivery tag 1, class-id=60, method-id=80)

**使用rabbitmq的时候总是报错信道关闭,而且这个错居然不影响消息队列运行。  
原因是因为进行了两次消息确认double ack.**

**yml中配置手动签收模式失效，被注解注入的SimpleRabbitListenerContainerFactory覆盖，而它默认使用了自动签收。但是消费消息的时候又手动进行channel.basicAck(deliveryTag, false)，于是导致了两次ack,所以报错。  
解决方法是在rabbitmq的factory中指定ack模式factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);**

**yml配置**

server:
      port: 9000
    
    spring:
      application:
        name: mall-order
      datasource:
        username: root
        password: 123456
        url: jdbc:mysql://192.168.217.129:3306/mall_oms?useUnicode=true&characterEncoding=UTF-8&serverTimezone=Asia/Shanghai
        driver-class-name: com.mysql.jdbc.Driver
      cloud:
        nacos:
          discovery:
            server-addr: localhost:8848
      rabbitmq:
        host: 192.168.217.129
      #开启发送端确认
        publisher-confirms: true
        publisher-returns: true
        template:
          mandatory: true  #只要消息抵达队列,以异步方式优先回调returnsConfirm
        listener:
          direct:
            acknowledge-mode: manual # 消费端手动ack消息    
      thymeleaf:
        cache: false
      session:
        store-type: redis    
      redis:
        host: 192.168.217.129

**config配置**

import lombok.extern.slf4j.Slf4j;
    import org.springframework.amqp.core.AcknowledgeMode;
    import org.springframework.amqp.core.Message;
    import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
    import org.springframework.amqp.rabbit.connection.ConnectionFactory;
    import org.springframework.amqp.rabbit.connection.CorrelationData;
    import org.springframework.amqp.rabbit.core.RabbitAdmin;
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
    import org.springframework.amqp.support.converter.MessageConverter;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    import javax.annotation.PostConstruct;
    
    /**
     * @author jl
     * Created on 2020/8/23
     */
    @Configuration
    @Slf4j
    public class RabbitMQConfig {
        @Autowired
        private RabbitTemplate rabbitTemplate;
    
        @Bean
        public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(ConnectionFactory connectionFactory, MessageConverter messageConverter) {
            SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
            factory.setConnectionFactory(connectionFactory);
            factory.setMessageConverter(messageConverter);
            factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
            return factory;
        }
    
    
        /**
         * 消息对象序列化器
         */
        @Bean
        public MessageConverter messageConverter() {
            return new Jackson2JsonMessageConverter();
        }
    
        /**
         * 定制RabbitTemplate,确保消息不丢失
         * 生产端消ConfirmCallback,ReturnCallback
         * 消费端ACK机制
         */
        @PostConstruct
        public void initRabbitTemplate() {
            // ConfirmCallback消息抵达交换机的回调
            rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
                /**
                 * 交换机(Exchange)收到消息就会回调
                 * CorrelationData 当前消息的唯一关联数据
                 * ack 消息是否成功送达
                 * cause 失败的原因
                 */
                @Override
                public void confirm(CorrelationData correlationData, boolean ack, String cause) {
                    log.info("ID:[{}]的消息成功投递到交换机",correlationData.getId());
                }
            });
            // ReturnCallback消息抵达队列的回调
            rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
                /**
                 * 只要消息没有投递给指定的队列就会触发该回调
                 * @param message 投递失败的消息
                 * @param replayCode 回复的状态码
                 * @param replayText 回复的文本内容
                 * @param exchange 交换机
                 * @param routingKey 路由key
                 */
                @Override
                public void returnedMessage(Message message, int replayCode, String replayText, String exchange, String routingKey) {
                    System.out.println(message);
                    log.error("ID:[{}]的消息失败投递到队列",message.getMessageProperties().getMessageId());
                }
            });
        }
    }