问题现象: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: 9000spring:application:name: mall-orderdatasource:username: rootpassword: 123456url: jdbc:mysql://192.168.217.129:3306/mall_oms?useUnicode=true&characterEncoding=UTF-8&serverTimezone=Asia/Shanghaidriver-class-name: com.mysql.jdbc.Drivercloud:nacos:discovery:server-addr: localhost:8848rabbitmq:host: 192.168.217.129#开启发送端确认publisher-confirms: truepublisher-returns: truetemplate:mandatory: true #只要消息抵达队列,以异步方式优先回调returnsConfirmlistener:direct:acknowledge-mode: manual # 消费端手动ack消息 thymeleaf:cache: falsesession: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 /8/23*/@Configuration@Slf4jpublic class RabbitMQConfig {@Autowiredprivate RabbitTemplate rabbitTemplate;@Beanpublic SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(ConnectionFactory connectionFactory, MessageConverter messageConverter) {SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();factory.setConnectionFactory(connectionFactory);factory.setMessageConverter(messageConverter);factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);return factory;}/*** 消息对象序列化器*/@Beanpublic MessageConverter messageConverter() {return new Jackson2JsonMessageConverter();}/*** 定制RabbitTemplate,确保消息不丢失* 生产端消ConfirmCallback,ReturnCallback* 消费端ACK机制*/@PostConstructpublic void initRabbitTemplate() {// ConfirmCallback消息抵达交换机的回调rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {/*** 交换机(Exchange)收到消息就会回调* CorrelationData 当前消息的唯一关联数据* ack 消息是否成功送达* cause 失败的原因*/@Overridepublic 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*/@Overridepublic void returnedMessage(Message message, int replayCode, String replayText, String exchange, String routingKey) {System.out.println(message);log.error("ID:[{}]的消息失败投递到队列",message.getMessageProperties().getMessageId());}});}}
