如何配置RabbitMQ（在Spring Boot 2.x中），以便手动确认工作

Question

所以我阅读了所有的例子，我有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");
}

Answer 1

从头开始，结果证明了

@Transactional(rollbackOn = Exception.class)

造成了这个问题。如果我删除它，它正在工作