Итак, я прочитал все примеры, у меня есть com.rabbitmq.client.Channel и @Header (AmqpHeaders.DELIVERY_TAG), но когда я пытаюсь вызвать Channel.basicNack (long deliveryTag, логическое множество, логическое требование)В результате получается «java.lang.IllegalStateException: канал закрыт;Я вижу, что CachingConnectionFactory не поддерживает ни один из методов подтверждения.Поэтому у меня вопрос: какую фабрику соединений я должен использовать и как ее настроить, чтобы 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");
}