Spring Amqp消费者在运行一段时间后暂停

Question

我们有一个具有Ha all策略的2节点RabbitMQ集群。我们在应用程序中使用Spring AMQP与RabbitMQ对话。生产者部分工作正常，但消费者工作了一段时间并暂停。生产者和消费者作为不同的应用程序运行。更多关于消费者部分的信息。

• 我们将SimpleMessageListenerContainer与ChannelAwareMessageListener一起使用，使用手动ack模式和默认预取（1）
• 在我们的应用程序中，我们创建队列（按需）并将其添加到侦听器中
• 当我们从10个ConcurrentConsumers和20个MaxConcurrentConsumers开始时，消费大约持续15个小时并暂停。当我们将MaxConcurrentConsumers增加到75时，这种情况会在1小时内发生

在RabbitMQ UI上，当出现这种情况时，我们会在通道选项卡上看到带有3/4 unacked消息的通道，在此之前，只有1条unacked消息。

我们的线程转储类似于此。但是将心跳设置为60无助于改善这种情况。

大多数线程转储都有以下消息。如果需要，我将附加整个线程转储。如果我缺少任何可能导致消费者暂停的设置，请告知我？

"pool-6-thread-16" #86 prio=5 os_prio=0 tid=0x00007f4db09cb000 nid=0x3b33 waiting on condition [0x00007f4ebebec000]
   java.lang.Thread.State: WAITING (parking)
    at sun.misc.Unsafe.park(Native Method)
    - parking to wait for  <0x00000007b9930b68> (a java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject)
    at java.util.concurrent.locks.LockSupport.park(LockSupport.java:175)
    at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.await(AbstractQueuedSynchronizer.java:2039)
    at java.util.concurrent.LinkedBlockingQueue.put(LinkedBlockingQueue.java:350)
    at org.springframework.amqp.rabbit.listener.BlockingQueueConsumer$InternalConsumer.handleDelivery(BlockingQueueConsumer.java:660)
    at com.rabbitmq.client.impl.ConsumerDispatcher$5.run(ConsumerDispatcher.java:144)
    at com.rabbitmq.client.impl.ConsumerWorkService$WorkPoolRunnable.run(ConsumerWorkService.java:99)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
    at java.lang.Thread.run(Thread.java:745)

更多信息我们动态地向SimpleMessageListenerContainer添加和删除队列，我们怀疑这会导致一些问题，因为每次我们从侦听器添加或删除队列时，所有BlockingQueueConsumer都会被删除并再次创建。你认为这会导致这个问题吗？

Answer 1

AMQP-621现在合并为master；我们将发布1.6.1。在接下来的几天发布。

Answer 2

您的问题在目标侦听器的下游某处。

看，预取（1）导致：

this.queue = new LinkedBlockingQueue<Delivery>(prefetchCount);

如果我们不调查队列，我们这里有什么？

BlockingQueueConsumer.this.queue.put(new Delivery(consumerTag, envelope, properties, body));

对，锁着停车。