Spring Amqp Consumer 运行一段时间后暂停答案

【问题标题】：Spring Amqp Consumer pauses after running for sometimeSpring Amqp Consumer 运行一段时间后暂停
【发布时间】：2016-06-30 23:11:25
【问题描述】：

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

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

在 RabbitMQ UI 上，当这种情况发生时，我们会在频道选项卡上看到具有 3/4 条 unacked 消息的频道，在此之前它只有 1 条 unacked 消息。

我们的线程转储类似于this。但是将心跳设置为 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 都会被删除并重新创建。你认为这是否会导致这个问题？

【问题讨论】：

你需要在某个地方发布完整的线程转储，可能不在这里，因为它太大了；可能是 pastebin 或 github gist 之类的东西。很可能容器线程卡在您的代码中的某个地方。
@GaryRussell: pastebin.com/UrBLfn2C 是包含完整线程转储的粘贴箱。
线程转储看起来不错 - 所有容器线程都在 nextMessage() 中等待，所以看起来您当前的理论是正确的 - 网络中的某些东西已经默默地断开了连接 - 一些路由器在空闲时这样做连接。设置请求的心跳应该保持连接活跃 - 你需要使用网络监视器（tcpdump、wireshark 等）来解决这个问题。
@GaryRussell ：我们尝试了心跳选项，但它没有用（尽管我们没有尝试使用任何网络监视器）而且我们也没有空闲连接，我们有连续的流量。我在问题中添加了更多信息，我们认为这可能是原因。
嗯-这是一种可能性-您问题中的线程正在尝试插入消息；看起来队列已满，但也许旧的消费者已经离开了——这不应该发生，因为我们在停止消费者之前取消了消费者，但可能存在竞争条件。请打开JIRA issue，我们会看看。

标签： rabbitmq spring-amqp consumer spring-rabbit

【解决方案1】：

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

看，prefetch(1) 原因是：

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

如果我们不轮询该队列，我们这里有什么？

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

对，停车锁上。

【讨论】：

【解决方案2】：

AMQP-621 现已合并到 master；我们将在接下来的几天内发布 1.6.1.RELEASE。

【讨论】：