如何配置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中),以便手动确认工作的主要内容,如果未能解决你的问题,请参考以下文章

Spring Boot项目配置RabbitMQ集群

Spring Boot 配置多源的 RabbitMQ

Spring Boot 整合 RabbitMQ

Spring Boot整合RabbitMQ

Spring Boot整合RabbitMQ

Spring Boot整合RabbitMQ