spring rabbitmq получает все ответы на сообщение с фанатами

Question

Следующий класс входит в несколько потребительских приложений:

@Component
@Configuration
public class HealthListener {

    public static final String HEALTH_CHECK_QUEUE_NAME = "healthCheckQueue";
    public static final String HEALTH_CHECK_FANOUT_EXCHANGE_NAME = "health-check-fanout";


    @Bean
    public Binding healthListenerBinding(
            @Qualifier("healthCheckQueue") Queue queue,
            @Qualifier("instanceFanoutExchange") FanoutExchange exchange) {
        return BindingBuilder.bind(queue).to(exchange);
    }

    @Bean
    public FanoutExchange instanceFanoutExchange() {
        return new FanoutExchange(HEALTH_CHECK_FANOUT_EXCHANGE_NAME, true, false);
    }

    @Bean
    public Queue healthCheckQueue() {
        return new Queue(HEALTH_CHECK_QUEUE_NAME);
    }

    @RabbitListener(queues = HEALTH_CHECK_QUEUE_NAME)
    public String healthCheck() {
        return "some result";
    }

}

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

Я могу отправить сообщение и получить первый ответ, например:

@Autowired
RabbitTemplate template;

// ...
String firstReply = template.convertSendAndReceiveAsType("health-check-fanout", "", "", ParameterizedTypeReference.forType(String.class));

Однако мне нужно получить все ответы на это сообщение, а не только первое. Мне нужно настроить прослушиватель ответов, но я не знаю, как.

Answer 1

Методы (convertS|s)endAndReceive.*() не предназначены для обработки нескольких ответов; это строго один метод запроса / одного ответа.

Вам потребуется использовать метод (convertAndS|s)end() для отправки запроса и реализовать собственный механизм ответа, возможно, с использованием контейнера слушателя для ответов, вместе с некоторым компонентом для агрегирования ответов.

Для этого вы можете использовать что-то вроде Spring Integration Aggregator, но вам потребуется какой-то механизм (ReleaseStrategy), который бы знал, когда будут получены все ожидаемые ответы.

Или вы можете просто получать отдельные ответы и обрабатывать их индивидуально.

EDIT

@SpringBootApplication
public class So54207780Application {

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

    @Bean
    public ApplicationRunner runner(RabbitTemplate template) {
        return args -> template.convertAndSend("fanout", "", "foo", m -> {
            m.getMessageProperties().setReplyTo("replies");
            return m;
        });
    }

    @RabbitListener(queues = "queue1")
    public String listen1(String in) {
        return in.toUpperCase();
    }

    @RabbitListener(queues = "queue2")
    public String listen2(String in) {
        return in + in;
    }

    @RabbitListener(queues = "replies")
    public void replyHandler(String reply) {
        System.out.println(reply);
    }

    @Bean
    public FanoutExchange fanout() {
        return new FanoutExchange("fanout");
    }

    @Bean
    public Queue queue1() {
        return new Queue("queue1");
    }

    @Bean
    public Binding binding1() {
        return BindingBuilder.bind(queue1()).to(fanout());
    }

    @Bean
    public Queue queue2() {
        return new Queue("queue2");
    }

    @Bean
    public Binding binding2() {
        return BindingBuilder.bind(queue2()).to(fanout());
    }

    @Bean
    public Queue replies() {
        return new Queue("replies");
    }

}

и

FOO
foofoo