java - Springboot RabbitMQ 与并发消费者

Question

我有一个简单的架构，其中三个进程之一(实际上是一个 spring-boot dockerized 容器)向交换/队列生成数千条消息，并且所有三个进程都消耗这些消息。

我希望消费者端是多线程的以实现最大吞吐量，但以我当前的配置，我没有实现我想要的。

这是设置:

听众:

  @RabbitListener(
    bindings = @QueueBinding(
      value = @Queue(
                value = RabbitMQConfiguration.BORD_QUEUE, 
                durable = "true", 
                arguments = @Argument(name = "x-queue-mode", value = "lazy")),
      exchange = @Exchange(value = RabbitMQConfiguration.QUEUE_EXCHANGE),
      key = "B"), 
    concurrency = "8-8")

  public void listenBordero(Long bId, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long tag) {
    log.info("Received message: {{}}", bId);
    processor.process(bId);
  }

这是 - 可能相关 - springboot application.properties 配置:

spring.rabbitmq.listener.type=simple
spring.rabbitmq.listener.simple.prefetch=1
spring.rabbitmq.listener.simple.max-concurrency=8
spring.rabbitmq.listener.simple.concurrency=8

尽管我相信由于 @RabbitListener 基于注释的配置，并非所有这些都是必需的。

在rabbitmq-managment中我看到 奥 git _awith状态主要是空闲。

从我的 springboot-logs 中，我看到线程几乎没有发生(日志中的线程名称仅在数十行后发生变化，而我希望切换得更频繁)。

队列显示在管理控制台中:

我的期望是在 channel 页面上看到消费者的统计数据更多地处于“运行”状态。

谁能告诉我一下吗？

提前致谢!

编辑:激活 spring-amqp 的调试后，我看到正在接收多条消息，但随后名为“..Container#0-n”的线程仅被整体处理。我预计多个正在运行的容器线程的输出会混合在一起。这可能是线程池问题吗？

这是(降噪的)日志摘录:

2020-03-05 08:15:21.418  INFO 32960 --- [ntContainer#0-7] d.i.clearing.rabbit.RabbitMQListener     : Received message: {436}
2020-03-05 08:15:21.418  INFO 32960 --- [ntContainer#0-5] d.i.clearing.rabbit.RabbitMQListener     : Received message: {450}
2020-03-05 08:15:21.418  INFO 32960 --- [tContainer#0-12] d.i.clearing.rabbit.RabbitMQListener     : Received message: {414}
2020-03-05 08:15:21.418  INFO 32960 --- [tContainer#0-11] d.i.clearing.rabbit.RabbitMQListener     : Received message: {418}
2020-03-05 08:15:21.418  INFO 32960 --- [ntContainer#0-8] d.i.clearing.rabbit.RabbitMQListener     : Received message: {432}
2020-03-05 08:15:21.418  INFO 32960 --- [ntContainer#0-2] d.i.clearing.rabbit.RabbitMQListener     : Received message: {456}
2020-03-05 08:15:21.418  INFO 32960 --- [tContainer#0-15] d.i.clearing.rabbit.RabbitMQListener     : Received message: {469}
2020-03-05 08:15:21.419  INFO 32960 --- [tContainer#0-16] d.i.clearing.rabbit.RabbitMQListener     : Received message: {394}
2020-03-05 08:15:21.418  INFO 32960 --- [tContainer#0-13] d.i.clearing.rabbit.RabbitMQListener     : Received message: {409}
2020-03-05 08:15:21.419  INFO 32960 --- [ntContainer#0-4] d.i.clearing.rabbit.RabbitMQListener     : Received message: {452}
2020-03-05 08:15:21.418  INFO 32960 --- [ntContainer#0-9] d.i.clearing.rabbit.RabbitMQListener     : Received message: {431}
2020-03-05 08:15:21.419  INFO 32960 --- [tContainer#0-10] d.i.clearing.rabbit.RabbitMQListener     : Received message: {420}
2020-03-05 08:15:21.419  INFO 32960 --- [tContainer#0-14] d.i.clearing.rabbit.RabbitMQListener     : Received message: {350}
2020-03-05 08:15:21.419  INFO 32960 --- [ntContainer#0-1] d.i.clearing.rabbit.RabbitMQListener     : Received message: {400}
2020-03-05 08:15:21.419  INFO 32960 --- [ntContainer#0-3] d.i.clearing.rabbit.RabbitMQListener     : Received message: {453}
2020-03-05 08:15:21.419  INFO 32960 --- [ntContainer#0-6] d.i.clearing.rabbit.RabbitMQListener     : Received message: {449}
... Logging Noise removed
2020-03-05 08:15:26.409  INFO 32960 --- [ntContainer#0-3] de.idslogistik.clearing.api.p2.Clearing  : Calculating {7183042} - {8330}/{5500}  
2020-03-05 08:15:26.411  INFO 32960 --- [ntContainer#0-3] de.idslogistik.clearing.api.p2.Clearing  : ... Logging Noise removed
2020-03-05 08:15:26.412 DEBUG 32960 --- [ntContainer#0-3] de.idslogistik.clearing.api.p2.Clearing  : ... Logging Noise removed
2020-03-05 08:15:26.413  INFO 32960 --- [ntContainer#0-3] de.idslogistik.clearing.api.p2.Clearing  : ... Logging Noise removed
2020-03-05 08:15:26.507  INFO 32960 --- [ntContainer#0-3] de.idslogistik.clearing.api.p2.Clearing  : ... Logging Noise removed
2020-03-05 08:15:26.669  WARN 32960 --- [ntContainer#0-3] de.idslogistik.clearing.api.p2.Rddel     : ... Logging Noise removed
2020-03-05 08:15:26.669  INFO 32960 --- [ntContainer#0-3] de.idslogistik.clearing.api.p2.Rddel     : ... Logging Noise removed
2020-03-05 08:15:33.012 DEBUG 32960 --- [ntContainer#0-3] de.idslogistik.clearing.api.p2.Rddel     : ... Logging Noise removed
2020-03-05 08:15:33.013 DEBUG 32960 --- [ntContainer#0-3] d.i.i.util.clearing.VerSendungTools      : ... Logging Noise removed
2020-03-05 08:15:33.013 DEBUG 32960 --- [ntContainer#0-3] d.i.i.util.clearing.VerSendungTools      : ... Logging Noise removed
2020-03-05 08:15:33.013 DEBUG 32960 --- [ntContainer#0-3] d.i.i.util.clearing.VerSendungTools      : ... Logging Noise removed
2020-03-05 08:15:33.014 DEBUG 32960 --- [ntContainer#0-3] d.i.i.util.clearing.VerSendungTools      : ... Logging Noise removed

Answer 1

prefetch = 1表示当前队列中未确认的消息最多为1条。当这条消息没有被确认时，下一条消息无法进入消费队列。

或者可以考虑增加消费者数量来加快消息的处理速度 https://www.rabbitmq.com/tutorials/tutorial-two-java.html