Какой лучший способ отправить сообщение в rabbitmq с помощью spring-amqp из обратного вызова PublisherReturn?

Я использую spring-amqp:2.1.6.RELEASE

У меня есть RabbitTemplate с обратным вызовом PublisherReturn.

• Если я отправлю сообщение на routingKey, у которого нет очередей, привязанных к это, то обратный вызов обратного вызова вызывается правильно. Когда это происходит, я хотите отправить сообщение на альтернативный routingKey. Однако, если Я использую RabbitTemplate в ReturnCallback, он просто зависает. я не вижу ничего о том, что сообщение может/не может быть отправлено, RabbitTemplate не возвращает управление моему ReturnCallback, и я также не вижу никакого PublisherConfirm.
• Если я создам новый RabbitTemplate (с тем же CachingConnectionFactory) потом ведет себя так же. Мой звонок просто зависает.
• Если я отправлю сообщение на routingKey, к которому привязана очередь, то сообщение корректно поступает в очередь. ReturnCallback не вызывается в этом сценарии.

После некоторого расследования я пришел к выводу, что rabbitTemplate и/или соединение заблокированы до тех пор, пока исходное сообщение не будет полностью обработано.

Если я создам второй CachingConnectionFactory и RabbitTemplate и использую их в обратном вызове PublisherReturn, то, похоже, все будет работать нормально.

Итак, вот вопрос: как лучше всего отправить сообщение в обратном вызове PublisherReturn с помощью spring-amqp?

Я искал, но не могу найти ничего, что объясняет, как вы должны это сделать.

Вот упрощенные детали того, что у меня есть:

@Configuration
public class MyConfig {

    @Bean
    public ConnectionFactory connectionFactory() {
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory("localhost");
        connectionFactory.setPublisherReturns(true);
        // ... other settings left out for brevity
        return connectionFactory;
    }

    @Bean
    @Qualifier("rabbitTemplate")
    public RabbitTemplate rabbitTemplate(ReturnCallbackForAlternative returnCallbackForAlternative) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory());
        rabbitTemplate.setMandatory(true);
        rabbitTemplate.setReturnCallback(returnCallbackForAlternative);
        // ... other settings left out for brevity
        return rabbitTemplate;
    }

    @Bean
    @Qualifier("connectionFactoryForUndeliverable")
    public ConnectionFactory connectionFactoryForUndeliverable() {
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory("localhost");
        // ... other settings left out for brevity
        return connectionFactory;
    }

    @Bean
    @Qualifier("rabbitTemplateForUndeliverable")
    public RabbitTemplate rabbitTemplateForUndeliverable() {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactoryForUndeliverable());
        // ... other settings left out for brevity
        return rabbitTemplate;
    }

}

Затем, чтобы отправить сообщение, которое я использую

    @Autowired
    @Qualifier("rabbitTemplate")
    private RabbitTemplate rabbitTemplate;

    public void send(Message message) {
        rabbitTemplate.convertAndSend(
            "exchange-name",
            "primary-key",
            message);
    }

И код в ReturnCallback

@Component
public class ReturnCallbackForAlternative implements RabbitTemplate.ReturnCallback {

    @Autowired
    @Qualifier("rabbitTemplateForUndeliverable")
    private RabbitTemplate rabbitTemplate;

    @Override
    public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
        rabbitTemplate.convertAndSend(
            "exchange-name",
            "alternative-key",
            message);
    }

}

РЕДАКТИРОВАТЬ

Упрощенный пример для воспроизведения проблемы. Чтобы запустить его:

1. Запустите RabbitMq
2. Связать обмен с именем foo с очередью с именем foo
3. Запустить как весеннее загрузочное приложение

Вы увидите следующий вывод:

in returnCallback before message send

но ты не увидишь:

in returnCallback after message send

Если вы закомментируете connectionFactory.setPublisherConfirms(true);, он работает нормально.

@SpringBootApplication
public class HangingApplication {

    public static void main(String[] args) {
      SpringApplication.run(HangingApplication.class, args);
    }

    @Bean
    public ConnectionFactory connectionFactory() {
      CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
      connectionFactory.setPublisherReturns(true);
      connectionFactory.setPublisherConfirms(true);
      return connectionFactory;
    }

    @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
      RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
      rabbitTemplate.setExchange("foo");
      rabbitTemplate.setMandatory(true);

      rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
        System.out.println("Confirm callback for main template. Ack = " + ack);
      });

      rabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> {
        System.out.println("in returnCallback before message send");
        rabbitTemplate.send("foo", message);
        System.out.println("in returnCallback after message send");
      });

      return rabbitTemplate;
    }

    @Bean
    public ApplicationRunner runner(@Qualifier("rabbitTemplate") RabbitTemplate template) {

      return args -> {
        template.convertAndSend("BADKEY", "foo payload");
      };
    }

    @RabbitListener(queues = "foo")
    public void listen(String in) {
      System.out.println("Message received on undeliverable queue : " + in);
    }

}

Вот build.gradle, который я использовал:

plugins {
    id 'org.springframework.boot' version '2.1.5.RELEASE'
    id 'java'
}

apply plugin: 'io.spring.dependency-management'

group 'pcoates'
version '1.0-SNAPSHOT'

sourceCompatibility = 1.11

repositories {
    mavenCentral()
}

dependencies {
    compile 'org.springframework.boot:spring-boot-starter-amqp'
}