Как сделать так, чтобы несколько потоков брали из очереди RabbitMQ с помощью Spring Boot?

Question

Наше приложение использует данные из нескольких очередей, предоставленных RabbitMQ.Чтобы увеличить пропускную способность, мы запускаем несколько потоков в очереди, которые выполняют блокировку из этих очередей.

Для нового сервиса мы хотим использовать Spring Boot и снова иметь несколько потоков в очереди, которые получают данные из этих очередей.Вот канонический код Spring Boot для обработки данных, поступивших из некоторой очереди:

@StreamListener(target = Processor.INPUT)
@SendTo(Processor.OUTPUT)
public Message<SomeData> process(Message<SomeData> message) {
    SomeData result = service.process(message.getPayload());
    return MessageBuilder
            .withPayload(result)
            .copyHeaders(message.getHeaders())
            .build();
}

Теперь вопрос состоит в том, как заставить Spring Boot порождать несколько потоков для обслуживания одной очереди вместо одного потока.Пропускная способность очень важна для нашего приложения, следовательно, это необходимо.

Answer 1

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

@Bean
public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory
        (MessageConverter contentTypeConverter,
         SimpleRabbitListenerContainerFactoryConfigurer configurer) {
    SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();

    // the number of consumers is set as 5
    factory.setConcurrentConsumers(5);

    configurer.configure(factory, connectionFactory);
    factory.setMessageConverter(contentTypeConverter);
    return factory;
}

Answer 2

Проверьте доступные свойства , найдите rabbitmq.

spring.rabbitmq.listener.simple.concurrency = # Минимальное количество потоков вызывающего слушателя

Это выглядит многообещающе