正确配置 ActiveMQ 以避免 Producer 内存泄漏

Posted 2023-03-10

【中文标题】正确配置 ActiveMQ 以避免 Producer 内存泄漏

【英文标题】：Configuring ActiveMQ properly to avoid Producer memory leaks

【发布时间】：2019-06-06 04:16:25

【问题描述】：

我不是 ActiveMQ 专家，但是我尝试在 Internet 上搜索了很多类似的问题，但我仍然很困惑。我有以下问题。

在 Tomcat 8.x、Java 8、Spring Framework 4.3.18 中运行 Web 应用程序。 我的 Web 应用程序使用 org.apache.activemq:activemq-spring:5.11.0 依赖项使用 ActiveMQ 发送和接收消息。

我正在以这种方式设置 ActiveMQ 连接工厂：

<amq:connectionFactory id="amqJmsFactory" brokerURL="$jms.broker.url" />
<bean id="jmsConnectionFactory"
    class="org.apache.activemq.pool.PooledConnectionFactory" destroy-method="stop">
    <property name="connectionFactory" ref="amqJmsFactory" />
    <property name="maxConnections" value="2" />
    <property name="idleTimeout" value="60000" />
    <property name="timeBetweenExpirationCheckMillis" value="600000" />
    <property name="maximumActiveSessionPerConnection" value="10" />
</bean>

最后一个属性 (maximumActiveSessionPerConnection) 已设置为尝试解决以下问题（默认值似乎是 500，恕我直言，这是相当高的），但我不确定它是否真的有帮助，因为我仍然出现 OutOfMemory 错误。

这个连接工厂被一个监听容器工厂引用：

<jms:listener-container factory-id="activationJmsListenerContainerFactory"
    container-type="default" connection-factory="jmsConnectionFactory"
    concurrency="1" transaction-manager="centralTransactionManager">
</jms:listener-container>

通过一个 Spring Integration 4.3.17 入站适配器：

<int-jms:message-driven-channel-adapter id="invoiceEventJmsInboundChannelAdapter" 
    channel="incomingInvoiceEventJmsChannel"
    connection-factory="jmsConnectionFactory"
    destination-name="incomingEvent"
    max-concurrent-consumers="2"
    transaction-manager="customerTransactionManager"
    error-channel="unexpectedErrorChannel" />

并通过两个出站适配器：

<int-jms:outbound-channel-adapter id="invoiceEventJmsOutboundChannelAdapter"
    channel="outgoingInvoiceEventJmsChannel" destination-name="outgoingEvent"
    connection-factory="jmsConnectionFactory" explicit-qos-enabled="true" delivery-persistent="true" 
    session-transacted="true" />

<int-jms:outbound-channel-adapter
    id="passwordResetTokenSubmitterJmsOutboundChannelAdapter"
    channel="passwordResetTokenSubmitterJmsChannel"
    destination-name="passwordReset"
    connection-factory="jmsConnectionFactory" explicit-qos-enabled="true"
    delivery-persistent="false" session-transacted="false" />

一切运行良好，但我观察到 ActiveMQ 作为消息生产者（用于invoiceEventJmsOutboundChannelAdapter 适配器）在内存中累积了大量对象，这导致我的应用程序出现 OutOfMemory 错误。我的消息可能有几 KB 长，因为它们的有效负载是 XML 文件，但我不希望长时间持有这么多内存。

这是我对最近的 OutOfMemory 错误（使用 Eclipse MAT 进行调查）产生的堆转储的发现。发现了两个泄漏嫌疑人，都指向ConnectionStateTracker。

这是两个累加器之一：

Class Name                                                                                                  | Shallow Heap | Retained Heap
-------------------------------------------------------------------------------------------------------------------------------------------
java.util.concurrent.ConcurrentHashMap$HashEntry[4] @ 0xe295da78                                            |           32 |    58.160.312
'- table java.util.concurrent.ConcurrentHashMap$Segment @ 0xe295da30                                        |           40 |    58.160.384
   '- [15] java.util.concurrent.ConcurrentHashMap$Segment[16] @ 0xe295d9e0                                  |           80 |    68.573.384
      '- segments java.util.concurrent.ConcurrentHashMap @ 0xe295d9b0                                       |           48 |    68.573.432
         '- sessions org.apache.activemq.state.ConnectionState @ 0xe295d7e0                                 |           40 |    68.575.312
            '- value java.util.concurrent.ConcurrentHashMap$HashEntry @ 0xe295d728                          |           32 |    68.575.344
               '- [1] java.util.concurrent.ConcurrentHashMap$HashEntry[2] @ 0xe295d710                      |           24 |    68.575.368
                  '- table java.util.concurrent.ConcurrentHashMap$Segment @ 0xe295d6c8                      |           40 |    68.575.440
                     '- [12] java.util.concurrent.ConcurrentHashMap$Segment[16] @ 0xe295d678                |           80 |    68.575.616
                        '- segments java.util.concurrent.ConcurrentHashMap @ 0xe295d648                     |           48 |    68.575.664
                           '- connectionStates org.apache.activemq.state.ConnectionStateTracker @ 0xe295d620|           40 |    68.575.808
-------------------------------------------------------------------------------------------------------------------------------------------

如您所见，ConnectionStateTracker 的实例保留了大约 70 MB 的堆空间。 ConnectionStateTracker 有两个实例（我猜每个出站适配器一个），它们总共保留了大约 120 MB 的堆。他们在ConnectionState 的两个实例中累积它，而这两个实例又具有一个“会话”映射，其中包含累积总共 10 个SessionState 实例，而这些实例又具有ConcurrentHashMap 的生产者，累积总数为 1,258 @ 987654335@ 个实例。它们在 transactionState 字段中保留了这 120 MB 的堆，该字段的类型为 TransactionState，而 commands ArrayList 似乎保留了我要发送的全部消息。

我的问题是：为什么 ActiveMQ 将已经发出的消息保存在内存中？将所有这些消息保存在内存中也存在一些安全问题。

【参考方案1】：

这就是我最终解决这个问题的方法。

我认为这里的主要问题是bug AMQ-6603。所以，我们做的第一件事就是升级到 ActiveMQ 5.15.8。我认为这足以修复泄漏。

但是，在发现不鼓励将池连接工厂与侦听器容器工厂一起使用后，我们也稍微更改了配置。我认为 ActiveMQ 文档令人困惑，并且正确的 JMS 配置比应有的复杂。无论如何，如果您阅读 DefaultMessageListenerContainer 文档，您将阅读：

不要将 Spring 的 org.springframework.jms.connection.CachingConnectionFactory 与动态缩放结合使用。理想情况下，根本不要将它与消息侦听器容器一起使用，因为通常最好让侦听器容器本身在其生命周期内处理适当的缓存。此外，停止和重新启动侦听器容器仅适用于独立的、本地缓存的连接，而不适用于外部缓存的连接。

同样的情况也必须适用于 ActiveMQ PooledConnectionFactory。然而，ActiveMQ 文档却说：

Spring 的MessagListenerContainer 应该用于消息消费。这提供了 MDB 的所有功能 - 高效的 JMS 使用和消息侦听器池 - 但不需要完整的 EJB 容器。

您可以使用 activemq-pool org.apache.activemq.pool.PooledConnectionFactory 为您的消费者集合有效地汇集连接和会话，或者您可以使用 Spring JMS org.springframework.jms.connection.CachingConnectionFactory 来达到相同的效果。

因此，ActiveMQ 文档提出了相反的建议。我为此打开了bug AMQ-7140。出于这个原因，我现在只向 JMS clients 注入 PooledConnectionFactory（或 Spring CachingConnectionFactory），并且在使用 Spring <jms:listener-container> 或 Spring Integration <int-jms:message-driven-channel-adapter> 构建侦听器容器工厂，我宁愿设置与并发和缓存级别相关的属性。

另外的困难是我们将本地构造的事务管理器传递给侦听器容器工厂，以便将 JMS 消息提交与数据库提交同步。这确实会导致侦听器容器工厂在默认情况下完全禁用其连接缓存机制，除非您还设置了显式缓存级别（请参阅org.springframework.jms.listener.DefaultMessageListenerContainer.setCacheLevel(int)Javadoc）。

我通过说很难找到解决方案来结束这个答案，特别是因为我根本没有收到任何 ActiveMQ 开发人员的反馈，无论是在 mailing list 或问题跟踪器上，即使恕我直言，这可能是被视为安全问题。这种不活跃，再加上缺乏替代方案，让我开始思考是否应该在下一个项目中考虑使用 JMS。

【讨论】：

看起来我面临相同或非常相似的问题...不确定我是否能看到真正需要做些什么来解决它...***.com/questions/61406593/…

【参考方案2】：

我们在客户的网站上遇到了同样的问题。

查看 ActiveMQ 代码，ConnectionStateTracker 中有一个类 RemoveTransactionAction，它删除条目以响应 OpenWire RemoveInfo（类型 12）消息。此消息似乎是 ActiveMQ 收到消息后生成的。

【讨论】：

我终于解决了这个问题，我认为在我们的案例中我们遇到了 AMQ-6603 错误：issues.apache.org/jira/browse/AMQ-6603。我将添加更多详细信息的回复。

【参考方案3】：

TL;DR

在您的 PooledConnectionFactory 上禁用 anonymousProducers。

---

在使用 JMS 事务和配置最多 8 个连接的 PooledConnectionFactory 时，我们遇到了一个非常相似的问题，即服务内存不足。

但是，我们没有使用 DefaultMessageListenerContainer 或 Spring，并且只发送了一个生产者。

这个生产者负责从批处理作业中发送大量消息，我们发现当连接失败时，它会将这些消息留在附加到旧连接的ConnectionStateTracker 上。在多次故障转移之后，这些消息会在旧连接上累积到我们用完堆的地步。

似乎从内存中清除这些消息的唯一方法是在提交 JMS 事务后关闭生产者。这将从ConnectionStateTracker 上的SessionState 中删除ProducerState 实例。

SimonD 在他的回答中提到的对RemoveTransactionAction 的调用（在提交 JMS 事务后自动发生）只是从ConnectionState 中删除了TransactionState，但仍将生产者及其消息留在SessionState对象。

不幸的是在生产者上调用 close() 不能与 PooledConnectionFactory 一起使用，因为默认情况下它使用匿名生产者 - 在匿名生产者上调用 close() 方法有没有效果。您必须先在 PooledConnectionFactory 上调用 setUseAnonymousProducers(false) 以关闭生产者才能生效。

还值得指出的是，您必须在生产者上调用close() - 在会话上调用close() 不会导致生产者被关闭，尽管ActiveMQSession 的JavaDoc 建议这样做。相反，它在生产者上调用dispose() 方法。

【讨论】：

感谢分享，很有趣。不幸的是，在使用 Spring 时，您通常无法直接控制这些低级组件（而且您可能不应该这样做），因此您当然可以在工厂级别禁用匿名生产者，但关闭生产者并非如此直的。我只是想知道你是否看过AMQ-6603。