spring-boot - RabbitMQ Spring Boot AMQP - 使用并发线程

Question

我希望我的应用能够同时处理从 RabbitMQ 接收到的多条消息。我已经尝试了可能所有的 google-page-1 解决方案，但它不起作用。这是我的设置:

POM.xml

<parent>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-parent</artifactId>
    <version>1.5.10.RELEASE</version>
</parent>
.
.
.
<dependency>
    <groupId>org.springframework.amqp</groupId>
    <artifactId>spring-rabbit</artifactId>
</dependency>

应用程序属性:

#############################
#         RabbitMQ          #
#############################
#AMQP RabbitMQ configuration 
spring.rabbitmq.host=zzzzzzzz
spring.rabbitmq.port=5672
spring.rabbitmq.username=zzzzzzz
spring.rabbitmq.password=zzzzzzz
#Rabbit component names
com.cp.neworder.queue.name = new-order-queue-stg
com.cp.neworder.queue.exchange = new-order-exchange-stg
com.cp.completedorder.queue.name = completed-order-queue
com.cp.completedorder.queue.exchange = completed-order-exchange
#Rabbit MQ concurrect consumers config
spring.rabbitmq.listener.simple.concurrency=3
spring.rabbitmq.listener.simple.retry.initial-interval=3000

配置文件:

@Configuration
public class RabbitMQConfig {

    @Value("${com.cp.neworder.queue.name}")
    private String newOrderQueueName;
    @Value("${com.cp.neworder.queue.exchange}")
    private String newOrderExchangeName;

    @Bean
    Queue queue() {
        return new Queue(newOrderQueueName, true);
    }

    @Bean
    TopicExchange exchange() {
        return new TopicExchange(newOrderExchangeName);
    }

    @Bean
    Binding binding(Queue queue, TopicExchange exchange) {
        return BindingBuilder.bind(queue).to(exchange).with(newOrderQueueName);
    }

    @Bean
    SimpleMessageListenerContainer container(ConnectionFactory connectionFactory, MessageListenerAdapter listenerAdapter) {
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        container.setQueueNames(newOrderQueueName);
        container.setMessageListener(listenerAdapter);
        return container;
    }

    @Bean
    MessageListenerAdapter listenerAdapter(OrderMessageListener receiver) {
        return new MessageListenerAdapter(receiver, "receiveOrder");
    }

}

我的消费者类按预期工作，它一次只处理一个请求。我怎么知道？

1. 我将异步请求的处理过程保存在数据库中，因此我可以查询当前有多少处理，它始终只有 1。
2. 我可以查看 RabbitMQ 管理平台，我看到它正在一个接一个地出队。

我的设置有哪些错误？如何让它发挥作用？

谢谢。

Answer 1

SimpleMessageListenerContainer 有一种设置并发消费者的方法。它有 setConcurrentConsumers method您可以在其中设置消费者数量。

    @Bean
    SimpleMessageListenerContainer container(ConnectionFactory connectionFactory, MessageListenerAdapter listenerAdapter) {
      SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
      container.setConnectionFactory(connectionFactory);
      container.setQueueNames(newOrderQueueName);
      container.setMessageListener(listenerAdapter);
      container. setConcurrentConsumers(10);
      return container;
  }

有了这个配置，当你启动应用程序时，你将能够在 RabbitMQ admin 中看到多个消费者