Apache ActiveMQ 中的内存泄漏

问题描述 投票：0回答：2

我是 Apache ActiveMQ 的新手，在 ActiveMQ 中发送和存储大量大消息（最大的消息约为 100mb）时遇到一些问题。消息是持久的，因此据我所知它们存储在硬盘驱动器上，而不是内存中。但奇怪的是，当数据库大小（KahaDB 文件夹）达到 2.8Gb（4Gb jvm 堆的 70%）时，ActiveMQ 崩溃。看来 ActiveMQ 仍然将所有消息存储在内存中。我首先需要存储所有消息，然后我将消耗它们。

我的配置activemq.xml

<policyEntry queue=">" producerFlowControl="false" memoryLimit="100MB">
    <pendingQueuePolicy>
        <fileQueueCursor />
    </pendingQueuePolicy>
</policyEntry>
...
<systemUsage>
    <systemUsage>
        <memoryUsage>
            <memoryUsage percentOfJvmHeap="70" />
        </memoryUsage>
        <storeUsage>
            <storeUsage limit="100 gb"/>
        </storeUsage>
        <tempUsage>
            <tempUsage limit="50 gb"/>
        </tempUsage>
    </systemUsage>
</systemUsage>
...
<transportConnectors>
    <transportConnector name="openwire" uri="tcp://0.0.0.0:61616?maximumConnections=1000&amp;wireFormat.maxFrameSize=204857600"/>
</transportConnectors>

这就是我如何在 java 中发送消息。尝试使用资源会关闭所有内容。

try (Connection conn = queueConnFactory.createConnection();
        Session session = conn.createSession(true, Session.AUTO_ACKNOWLEDGE);
        MessageProducer producer = session.createProducer(getTargetQueue())) {

      ObjectMessage message = session.createObjectMessage(transferMessage);
      producer.send(message);
    }

这是activeMQ上次崩溃时的日志。

ERROR | Forwarding of acks failed | org.apache.activemq.store.kahadb.MessageDatabase | ActiveMQ Journal Checkpoint Worker
java.lang.OutOfMemoryError: Java heap space
2019-08-02 20:32:12,421 | INFO  | Ignoring no space left exception, java.io.IOException: Java heap space | org.apache.activemq.util.DefaultIOExceptionHandler | ActiveMQ Journal Checkpoint Worker
java.io.IOException: Java heap space
    at org.apache.activemq.util.IOExceptionSupport.create(IOExceptionSupport.java:40)[activemq-client-5.15.9.jar:5.15.9]
    at org.apache.activemq.store.kahadb.MessageDatabase$AckCompactionRunner.run(MessageDatabase.java:2075)[activemq-kahadb-store-5.15.9.jar:5.15.9]
    at java.util.concurrent.Executors$RunnableAdapter.call(Unknown Source)[:1.8.0_211]
    at java.util.concurrent.FutureTask.run(Unknown Source)[:1.8.0_211]
    at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(Unknown Source)[:1.8.0_211]
    at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(Unknown Source)[:1.8.0_211]
    at java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown Source)[:1.8.0_211]
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)[:1.8.0_211]
    at java.lang.Thread.run(Unknown Source)[:1.8.0_211]
2019-08-02 20:37:12,514 | WARN  | Async error occurred: java.lang.OutOfMemoryError: Java heap space | org.apache.activemq.broker.TransportConnection.Service | ActiveMQ Transport: tcp:///10.11.34.224:58668@61616
2019-08-02 20:37:44,899 | ERROR | Forwarding of acks failed | org.apache.activemq.store.kahadb.MessageDatabase | ActiveMQ Journal Checkpoint Worker
java.lang.IllegalArgumentException: Self-suppression not permitted
    at java.lang.Throwable.addSuppressed(Unknown Source)[:1.8.0_211]
    at org.apache.activemq.store.kahadb.MessageDatabase.forwardAllAcks(MessageDatabase.java:2132)[activemq-kahadb-store-5.15.9.jar:5.15.9]
    at org.apache.activemq.store.kahadb.MessageDatabase.access$700(MessageDatabase.java:121)[activemq-kahadb-store-5.15.9.jar:5.15.9]
    at org.apache.activemq.store.kahadb.MessageDatabase$AckCompactionRunner.run(MessageDatabase.java:2068)[activemq-kahadb-store-5.15.9.jar:5.15.9]
    at java.util.concurrent.Executors$RunnableAdapter.call(Unknown Source)[:1.8.0_211]
    at java.util.concurrent.FutureTask.run(Unknown Source)[:1.8.0_211]
    at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(Unknown Source)[:1.8.0_211]
    at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(Unknown Source)[:1.8.0_211]
    at java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown Source)[:1.8.0_211]
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)[:1.8.0_211]
    at java.lang.Thread.run(Unknown Source)[:1.8.0_211]
2019-08-02 20:37:44,899 | INFO  | Stopping BrokerService[localhost] due to exception, java.io.IOException: Self-suppression not permitted | org.apache.activemq.util.DefaultIOExceptionHandler | ActiveMQ Journal Checkpoint Worker
java.io.IOException: Self-suppression not permitted
    at org.apache.activemq.util.IOExceptionSupport.create(IOExceptionSupport.java:40)[activemq-client-5.15.9.jar:5.15.9]
    at org.apache.activemq.store.kahadb.MessageDatabase$AckCompactionRunner.run(MessageDatabase.java:2075)[activemq-kahadb-store-5.15.9.jar:5.15.9]
    at java.util.concurrent.Executors$RunnableAdapter.call(Unknown Source)[:1.8.0_211]
    at java.util.concurrent.FutureTask.run(Unknown Source)[:1.8.0_211]
    at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(Unknown Source)[:1.8.0_211]
    at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(Unknown Source)[:1.8.0_211]
    at java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown Source)[:1.8.0_211]
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)[:1.8.0_211]
    at java.lang.Thread.run(Unknown Source)[:1.8.0_211]
2019-08-02 20:37:44,946 | INFO  | Apache ActiveMQ 5.15.9 (localhost, ID:PVAH-WF-58622-1564761931309-0:1) is shutting down | org.apache.activemq.broker.BrokerService | IOExceptionHandler: stopping BrokerService[localhost]
2019-08-02 20:37:45,133 | INFO  | Connector openwire stopped | org.apache.activemq.broker.TransportConnector | IOExceptionHandler: stopping BrokerService[localhost]
2019-08-02 20:37:45,180 | INFO  | Connector amqp stopped | org.apache.activemq.broker.TransportConnector | IOExceptionHandler: stopping BrokerService[localhost]
2019-08-02 20:37:46,974 | INFO  | Connector stomp stopped | org.apache.activemq.broker.TransportConnector | IOExceptionHandler: stopping BrokerService[localhost]
2019-08-02 20:37:46,974 | INFO  | Connector mqtt stopped | org.apache.activemq.broker.TransportConnector | IOExceptionHandler: stopping BrokerService[localhost]
2019-08-02 20:37:48,316 | INFO  | Connector ws stopped | org.apache.activemq.broker.TransportConnector | IOExceptionHandler: stopping BrokerService[localhost]
2019-08-02 20:37:48,955 | INFO  | PListStore:[C:\ActiveMQ\bin\win64\..\..\data\localhost\tmp_storage] stopped | org.apache.activemq.store.kahadb.plist.PListStoreImpl | IOExceptionHandler: stopping BrokerService[localhost]
2019-08-02 20:37:49,002 | INFO  | Stopping async queue tasks | org.apache.activemq.store.kahadb.KahaDBStore | IOExceptionHandler: stopping BrokerService[localhost]
2019-08-02 20:37:49,002 | INFO  | Stopping async topic tasks | org.apache.activemq.store.kahadb.KahaDBStore | IOExceptionHandler: stopping BrokerService[localhost]
2019-08-02 20:37:49,002 | INFO  | Stopped KahaDB | org.apache.activemq.store.kahadb.KahaDBStore | IOExceptionHandler: stopping BrokerService[localhost]
2019-08-02 20:37:51,404 | INFO  | Apache ActiveMQ 5.15.9 (localhost, ID:PVAH-WF-58622-1564761931309-0:1) uptime 1 hour 32 minutes | org.apache.activemq.broker.BrokerService | IOExceptionHandler: stopping BrokerService[localhost]
2019-08-02 20:37:58,502 | INFO  | Apache ActiveMQ 5.15.9 (localhost, ID:PVAH-WF-58622-1564761931309-0:1) is shutdown | org.apache.activemq.broker.BrokerService | IOExceptionHandler: stopping BrokerService[localhost]
2019-08-02 20:37:58,534 | INFO  | Closing org.apache.activemq.xbean.XBeanBrokerFactory$1@14d57a4: startup date [Fri Aug 02 19:05:26 MSK 2019]; root of context hierarchy | org.apache.activemq.xbean.XBeanBrokerFactory$1 | IOExceptionHandler: stopping BrokerService[localhost]
2019-08-02 20:38:00,296 | INFO  | Destroying Spring FrameworkServlet 'dispatcher' | /admin | IOExceptionHandler: stopping BrokerService[localhost]
2019-08-02 20:38:00,764 | INFO  | Refreshing org.apache.activemq.xbean.XBeanBrokerFactory$1@71d9564d: startup date [Fri Aug 02 20:38:00 MSK 2019]; root of context hierarchy | org.apache.activemq.xbean.XBeanBrokerFactory$1 | WrapperSimpleAppMain
2019-08-02 20:38:03,619 | INFO  | Using Persistence Adapter: KahaDBPersistenceAdapter[C:\ActiveMQ\bin\win64\..\..\data\kahadb] | org.apache.activemq.broker.BrokerService | WrapperSimpleAppMain
2019-08-02 20:38:04,196 | INFO  | ignoring zero length, partially initialised journal data file: db-404.log number = 404 , length = 0 | org.apache.activemq.store.kahadb.disk.journal.Journal | WrapperSimpleAppMain
2019-08-02 20:38:05,819 | INFO  | KahaDB is version 6 | org.apache.activemq.store.kahadb.MessageDatabase | WrapperSimpleAppMain
2019-08-02 20:38:06,177 | INFO  | PListStore:[C:\ActiveMQ\bin\win64\..\..\data\localhost\tmp_storage] started | org.apache.activemq.store.kahadb.plist.PListStoreImpl | WrapperSimpleAppMain
2019-08-02 20:38:06,177 | INFO  | Apache ActiveMQ 5.15.9 (localhost, ID:PVAH-WF-58622-1564761931309-0:2) is starting | org.apache.activemq.broker.BrokerService | WrapperSimpleAppMain
2019-08-02 20:38:10,374 | INFO  | PListStore:[C:\ActiveMQ\bin\win64\..\..\data\localhost\tmp_storage] initialized | org.apache.activemq.store.kahadb.plist.PListStoreImpl | WrapperSimpleAppMain
2019-08-02 20:38:26,925 | WARN  | Exception encountered during context initialization - cancelling refresh attempt: org.springframework.beans.factory.BeanCreationException: Error creating bean with name 'org.apache.activemq.xbean.XBeanBrokerService#0' defined in class path resource [activemq.xml]: Invocation of init method failed; nested exception is java.lang.OutOfMemoryError: Java heap space | org.apache.activemq.xbean.XBeanBrokerFactory$1 | WrapperSimpleAppMain
2019-08-02 20:38:26,925 | ERROR | Failed to load: class path resource [activemq.xml], reason: Error creating bean with name 'org.apache.activemq.xbean.XBeanBrokerService#0' defined in class path resource [activemq.xml]: Invocation of init method failed; nested exception is java.lang.OutOfMemoryError: Java heap space | org.apache.activemq.xbean.XBeanBrokerFactory | WrapperSimpleAppMain
org.springframework.beans.factory.BeanCreationException: Error creating bean with name 'org.apache.activemq.xbean.XBeanBrokerService#0' defined in class path resource [activemq.xml]: Invocation of init method failed; nested exception is java.lang.OutOfMemoryError: Java heap space
    at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.initializeBean(AbstractAutowireCapableBeanFactory.java:1634)[spring-beans-4.3.18.RELEASE.jar:4.3.18.RELEASE]
    at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.doCreateBean(AbstractAutowireCapableBeanFactory.java:555)[spring-beans-4.3.18.RELEASE.jar:4.3.18.RELEASE]
    at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.createBean(AbstractAutowireCapableBeanFactory.java:483)[spring-beans-4.3.18.RELEASE.jar:4.3.18.RELEASE]
    at org.springframework.beans.factory.support.AbstractBeanFactory$1.getObject(AbstractBeanFactory.java:312)[spring-beans-4.3.18.RELEASE.jar:4.3.18.RELEASE]
    at org.springframework.beans.factory.support.DefaultSingletonBeanRegistry.getSingleton(DefaultSingletonBeanRegistry.java:230)[spring-beans-4.3.18.RELEASE.jar:4.3.18.RELEASE]
    at org.springframework.beans.factory.support.AbstractBeanFactory.doGetBean(AbstractBeanFactory.java:308)[spring-beans-4.3.18.RELEASE.jar:4.3.18.RELEASE]
    at org.springframework.beans.factory.support.AbstractBeanFactory.getBean(AbstractBeanFactory.java:197)[spring-beans-4.3.18.RELEASE.jar:4.3.18.RELEASE]
    at org.springframework.beans.factory.support.DefaultListableBeanFactory.preInstantiateSingletons(DefaultListableBeanFactory.java:761)[spring-beans-4.3.18.RELEASE.jar:4.3.18.RELEASE]
    at org.springframework.context.support.AbstractApplicationContext.finishBeanFactoryInitialization(AbstractApplicationContext.java:867)[spring-context-4.3.18.RELEASE.jar:4.3.18.RELEASE]
    at org.springframework.context.support.AbstractApplicationContext.refresh(AbstractApplicationContext.java:543)[spring-context-4.3.18.RELEASE.jar:4.3.18.RELEASE]
    at org.apache.xbean.spring.context.ResourceXmlApplicationContext.<init>(ResourceXmlApplicationContext.java:64)[xbean-spring-4.2.jar:4.2]
    at org.apache.xbean.spring.context.ResourceXmlApplicationContext.<init>(ResourceXmlApplicationContext.java:52)[xbean-spring-4.2.jar:4.2]
    at org.apache.activemq.xbean.XBeanBrokerFactory$1.<init>(XBeanBrokerFactory.java:104)[activemq-spring-5.15.9.jar:5.15.9]
    at org.apache.activemq.xbean.XBeanBrokerFactory.createApplicationContext(XBeanBrokerFactory.java:104)[activemq-spring-5.15.9.jar:5.15.9]
    at org.apache.activemq.xbean.XBeanBrokerFactory.createBroker(XBeanBrokerFactory.java:67)[activemq-spring-5.15.9.jar:5.15.9]
    at org.apache.activemq.broker.BrokerFactory.createBroker(BrokerFactory.java:71)[activemq-broker-5.15.9.jar:5.15.9]
    at org.apache.activemq.broker.BrokerFactory.createBroker(BrokerFactory.java:54)[activemq-broker-5.15.9.jar:5.15.9]
    at org.apache.activemq.console.command.StartCommand.runTask(StartCommand.java:87)[activemq-console-5.15.9.jar:5.15.9]
    at org.apache.activemq.console.command.AbstractCommand.execute(AbstractCommand.java:63)[activemq-console-5.15.9.jar:5.15.9]
    at org.apache.activemq.console.command.ShellCommand.runTask(ShellCommand.java:154)[activemq-console-5.15.9.jar:5.15.9]
    at org.apache.activemq.console.command.AbstractCommand.execute(AbstractCommand.java:63)[activemq-console-5.15.9.jar:5.15.9]
    at org.apache.activemq.console.command.ShellCommand.main(ShellCommand.java:104)[activemq-console-5.15.9.jar:5.15.9]
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)[:1.8.0_211]
    at sun.reflect.NativeMethodAccessorImpl.invoke(Unknown Source)[:1.8.0_211]
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(Unknown Source)[:1.8.0_211]
    at java.lang.reflect.Method.invoke(Unknown Source)[:1.8.0_211]
    at org.apache.activemq.console.Main.runTaskClass(Main.java:262)[activemq.jar:5.15.9]
    at org.apache.activemq.console.Main.main(Main.java:115)[activemq.jar:5.15.9]
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)[:1.8.0_211]
    at sun.reflect.NativeMethodAccessorImpl.invoke(Unknown Source)[:1.8.0_211]
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(Unknown Source)[:1.8.0_211]
    at java.lang.reflect.Method.invoke(Unknown Source)[:1.8.0_211]
    at org.tanukisoftware.wrapper.WrapperSimpleApp.run(WrapperSimpleApp.java:240)[wrapper.jar:3.2.3]
    at java.lang.Thread.run(Unknown Source)[:1.8.0_211]
java memory-leaks out-of-memory activemq heap-memory
2个回答
0
投票

此时我没有看到任何内存泄漏的证据。仅仅因为代理内存不足并不意味着存在泄漏。您可能根本没有为 JVM 提供足够的堆来满足您的用例。

此外，仅仅因为代理使用某种数据库（即 KahaDB）将消息存储到磁盘并不意味着代理很像数据库。尽管消息被持久保存到磁盘，与您的断言相反，代理仍然会在内存中存储尽可能多的消息数据。将消息数据写入磁盘后将其从内存中清除，然后在消费者需要时立即从磁盘重新加载它，效率极低。还有一个问题是代理必须跟踪非持久消息，而不将它们写入磁盘。

如果您只是要存储所有数据，然后稍后检索它，那么最好只使用数据库，因为数据库更适合这种用例。消息代理通常更适合这样的用例：消费者与生产者同时活跃，并且消息不会在代理上构建，而是在到达后立即分发给消费者。

您的代码片段可能是反模式。 JMS 连接是线程安全的并且可以重复使用。事实上，如果应用程序除了发送一条单独的消息之外还执行任何其他 JMS 工作，则为发送的每条消息打开和关闭连接都是一种反模式。 最后，您可以考虑转向

ActiveMQ Artemis

（即下一代 ActiveMQ），因为 它支持“大”消息 - ActiveMQ 5.x 缺乏的东西。

0
投票

最新问题
© www.soinside.com 2019 - 2024. All rights reserved.