RabbitMQ java.lang.OutOfMemoryError答案

【问题标题】：RabbitMQ java.lang.OutOfMemoryErrorRabbitMQ java.lang.OutOfMemoryError
【发布时间】：2013-11-28 22:28:21
【问题描述】：

首先，如果这个问题已经在其他地方处理过，我深表歉意，我只是还没有找到解决我特定问题的方法。

我有一个从外部接收消息的网关服务器。它将它放入我的订单处理服务器正在侦听的队列中。我的订单处理服务器侦听 2 个队列（在一个线程中）。队列 1 - 网关服务器，队列 2 - 清算服务器。

所以在我的订单处理器中，我有工作线程。我正在使用 ExecutorService 来管理我的线程。问题出在工作线程中。

在工作线程中，我创建了两个 MQ 实例，用于将消息发布到清算服务器或网关服务器。我基本上需要做一些处理，然后将该消息发布到这些队列。

我想知道的是，我是否应该在每次处理完消息后关闭工作线程中的通道和连接？

如果我在处理消息后没有关闭每个工作线程上的 MQ 连接，然后在处理 8-900 条消息后，我开始间歇性地收到以下异常：

java.lang.OutOfMemoryError: unable to create new native thread
    at java.lang.Thread.start0(Native Method)
    at java.lang.Thread.start(Unknown Source)
    at com.rabbitmq.client.impl.AMQConnection.start(AMQConnection.java:307)
    at com.rabbitmq.client.ConnectionFactory.newConnection(ConnectionFactory.java:516)
    at com.rabbitmq.client.ConnectionFactory.newConnection(ConnectionFactory.java:533)
    at interfaces.MQ.<init>(MQ.java:41)
    at orderProcessor.ProcessOrders.<init>(ProcessOrders.java:109)
    at orderProcessor.ProcessIncomingCSThread.spawnThread(ProcessIncomingCSThread.java:52)
    at orderProcessor.ProcessIncomingCSThread.spawnThread(ProcessIncomingCSThread.java:70)
    at orderProcessor.ProcessIncomingCSThread.spawnThread(ProcessIncomingCSThread.java:70)
    at orderProcessor.ProcessIncomingCSThread.spawnThread(ProcessIncomingCSThread.java:70)
    at orderProcessor.ProcessIncomingCSThread.spawnThread(ProcessIncomingCSThread.java:70)
    at orderProcessor.ProcessIncomingCSThread.spawnThread(ProcessIncomingCSThread.java:70)
    at orderProcessor.ProcessIncomingCSThread.routeIncoming(ProcessIncomingCSThread.java:45)
    at interfaces.ProcessIncomingThread.run(ProcessIncomingThread.java:47)

如果我在处理完一条消息后确实关闭了每个工作线程上的连接，那么一段时间后我会间歇性地收到以下异常：]

Exception in thread "AMQP Connection 127.0.0.1:5672" java.lang.OutOfMemoryError: unable to create new native thread
    at java.lang.Thread.start0(Native Method)
    at java.lang.Thread.start(Unknown Source)
    at com.rabbitmq.client.impl.ChannelManager.scheduleShutdownProcessing(ChannelManager.java:108)
    at com.rabbitmq.client.impl.ChannelManager.handleSignal(ChannelManager.java:94)
    at com.rabbitmq.client.impl.AMQConnection.finishShutdown(AMQConnection.java:696)
    at com.rabbitmq.client.impl.AMQConnection.shutdown(AMQConnection.java:669)
    at com.rabbitmq.client.impl.AMQConnection$MainLoop.run(AMQConnection.java:550)

我在创建 MQ 连接以供消费时使用 basicQos，以保持 RabbitMQ 的内部队列合理。

我以下列方式创建我的 MQ 连接以供消费：

_channel.queueDeclare(this._mqName.toString(), true, false, false, null);
_channel.basicConsume(this._mqName.toString(), true, _consumer);
_channel.basicQos(50);

感谢您查看此内容，我们将不胜感激任何建议或帮助。我很可能没有根据我的上下文正确地做事..

【问题讨论】：

您不需要在每次发送消息时都关闭连接和通道。另外，我们在这里谈论的是什么负载？发送时服务器是否进入流态？
流状态是什么意思？我尝试了关闭和不关闭队列的两种方法。我让我的网关服务器生成 2k 个虚拟订单并将其推送到订单处理器正在侦听的队列中。如果我的工作线程不需要随后将消息推送到队列中，事情会容易得多，因为这样我就可以避免为了在工作线程中发布而使用 MQ 连接。
当代理无法足够快地将消息从交换机路由到队列时，RabbitMQ 中会发生流状态。发生这种情况时，生产者会受到限制，并开始在内部累积消息，等待代理能够处理负载。如果您继续尝试发送消息但代理处于流状态，则消息将存储在 JVM 内存中，这意味着在某些时候，您将耗尽内存。要检查您是否进入流状态，您需要在发送消息时检查代理中的连接是否处于流状态或阻塞状态。
很确定它没有处于流动状态，因为我几乎立即收到消息
那么它似乎是罗宾提到的内存泄漏。

标签： java multithreading rabbitmq threadpool out-of-memory

【解决方案1】：

好像你有内存泄漏。使用分析器。

【讨论】：

【解决方案2】：

感谢您的意见。我解决了这个问题。

我在每个工作线程中创建连接。现在我在主线程上创建连接并将其传递给从该连接创建通道的工作线程。这似乎是一种享受。

然而，这意味着，我必须重新设计我的 MQ 类来处理这个工作流。

【讨论】：

【解决方案3】：

我不熟悉 RabbitMQ，但我怀疑您或 RabbitMQ 正在尝试创建更多操作系统配置为处理的线程。

也许这两个链接可以帮助你：

【讨论】：