如何配置RabbitMQ(在Spring Boot 2.x中),以便手动确认工作
Posted
tags:
篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了如何配置RabbitMQ(在Spring Boot 2.x中),以便手动确认工作相关的知识,希望对你有一定的参考价值。
所以我阅读了所有的例子,我有com.rabbitmq.client.Channel和@Header(AmqpHeaders.DELIVERY_TAG),但是当我尝试调用Channel.basicNack(long deliveryTag,boolean multiple,boolean requeue)时,结果是“ java.lang.IllegalStateException:通道关闭;不能ack / nack“。我可以看到CachingConnectionFactory不支持任何确认方法。所以我的问题是我必须使用ConnectionFactory以及如何配置它,以便basicAck / basicNack工作?
Spring Boot Version 2.1.0.RELEASE application.yaml:
spring:
rabbitmq:
host: ${RABBITMQ_HOST:localhost}
port: ${RABBITMQ_PORT:5672}
username: ${RABBITMQ_USERNAME:guest}
password: ${RABBITMQ_PASSWORD:guest}
listener:
type: simple
simple:
acknowledge-mode: manual
配置类:
@EnableRabbit
@Configuration
public class RabbitMqConfig implements RabbitListenerConfigurer {
@Value("${app.rabbitmq.incoming-queue}")
private String incomingQueue;
private AmqpAdmin amqpAdmin;
@Autowired
public RabbitMqConfig(AmqpAdmin amqpAdmin) {
this.amqpAdmin = amqpAdmin;
}
@Bean
public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
rabbitTemplate.setChannelTransacted(true);
rabbitTemplate.setMessageConverter(jsonMessageConverter());
return rabbitTemplate;
}
@Bean
MessageConverter jsonMessageConverter() {
return new Jackson2JsonMessageConverter();
}
@Override
public void configureRabbitListeners(RabbitListenerEndpointRegistrar registar) {
registar.setMessageHandlerMethodFactory(createDefaultMessageHandlerMethodFactory());
}
@Bean
public DefaultMessageHandlerMethodFactory createDefaultMessageHandlerMethodFactory() {
DefaultMessageHandlerMethodFactory factory = new DefaultMessageHandlerMethodFactory();
factory.setMessageConverter(new MappingJackson2MessageConverter());
return factory;
}
@Bean
public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(
ConnectionFactory connectionFactory, CommonAmqpErrorHandler commonAmqpErrorHandler) {
SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
factory.setConnectionFactory(connectionFactory);
factory.setMessageConverter(jsonMessageConverter());
factory.setErrorHandler(commonAmqpErrorHandler);
return factory;
}
@PostConstruct
public void afterInit() {
amqpAdmin.declareQueue(new Queue(getDeadLetterQueueName(incomingQueue), true));
amqpAdmin.declareQueue(
QueueBuilder.durable(incomingQueue).withArgument("x-dead-letter-exchange", "")
.withArgument("x-dead-letter-routing-key",
getDeadLetterQueueName(incomingQueue)).build());
}
private String getDeadLetterQueueName(String queueName) {
return queueName + ".dead-letter.queue";
}
}
听众代码:
@Transactional(rollbackOn = Exception.class)
@RabbitListener(queues = "${app.rabbitmq.incoming-queue}", errorHandler = "notificationListenerErrorHandler")
public void onMessage(@Valid @Payload NotificationDto notification, Message message,
Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long tag) throws IOException {
System.out.println("00000 > " + tag);
System.out.println("11111");
channel.basicNack(tag, false, true);
System.out.println("222222");
}
答案
从头开始,结果证明了
@Transactional(rollbackOn = Exception.class)
造成了这个问题。如果我删除它,它正在工作
以上是关于如何配置RabbitMQ(在Spring Boot 2.x中),以便手动确认工作的主要内容,如果未能解决你的问题,请参考以下文章