spring - ActiveMQ 未释放队列的磁盘存储空间

Question

当向 ActiveMQ 队列施加大量负载时，它似乎按预期将消息入队/出队，但存储仍然已满。

这会导致生产者在一段时间后(当存储达到 100% 时)阻塞，进而导致应用程序停止响应请求。

这是我们正在使用的相关 activeMQ Spring 配置:

<beans
    xmlns="http://www.springframework.org/schema/beans"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xsi:schemaLocation="http://www.springframework.org/schema/beans
                        http://www.springframework.org/schema/beans/spring-beans.xsd
                        http://activemq.apache.org/schema/core
                        http://activemq.apache.org/schema/core/activemq-core.xsd
                        http://activemq.apache.org/schema/core
                        http://activemq.apache.org/schema/core/activemq-core.xsd">

<!-- =====================================================
     Broker Configuration
     ===================================================== -->
<broker id="appEmbeddedBroker"
        xmlns="http://activemq.apache.org/schema/core"
        brokerName="${msg.embedded.broker.name}"
        persistent="true"
        dataDirectory="${msg.embedded.broker.data.directory}"
        useJmx="true" >

    <destinationPolicy>
        <policyMap>
            <policyEntries>
                <policyEntry topic=">" >
                    <pendingMessageLimitStrategy>
                        <constantPendingMessageLimitStrategy limit="1000"/>
                    </pendingMessageLimitStrategy>
                </policyEntry>
            </policyEntries>
        </policyMap>
    </destinationPolicy>

    <managementContext>
        <managementContext connectorPort="${msg.embedded.broker.jmx.port}" createConnector="false"/>
    </managementContext>

    <persistenceAdapter>
        <levelDB directory="${msg.embedded.broker.db.directory}" />
    </persistenceAdapter>

    <systemUsage>
        <systemUsage>
            <memoryUsage>
                <memoryUsage percentOfJvmHeap="10"/>
            </memoryUsage>
            <storeUsage>
                <storeUsage limit="${msg.embedded.broker.system.usage.store.usage}"/> <!-- Configured for 200Mb -->
            </storeUsage>
            <tempUsage>
                <tempUsage limit="${msg.embedded.broker.system.usage.temp.usage}"/> <!-- Configured for 40Mb -->
            </tempUsage>
        </systemUsage>
    </systemUsage>

    <plugins>
        <!-- Configure authentication; Username, passwords and groups -->
        <simpleAuthenticationPlugin anonymousAccessAllowed="false">
            <users>
                <authenticationUser username="app" password="${msg.embedded.broker.app.password}"
                                    groups="users"/>
            </users>
        </simpleAuthenticationPlugin>
    </plugins>

    <transportConnectors>
        <transportConnector name="tcp" uri="tcp://0.0.0.0:${msg.embedded.broker.port}?maximumConnections=1000&amp;wireFormat.maxFrameSize=104857600"/>
    </transportConnectors>

    <shutdownHooks>
        <bean xmlns="http://www.springframework.org/schema/beans" class="org.apache.activemq.hooks.SpringContextHook" />
    </shutdownHooks>

</broker>


<!-- =====================================================
     Client Connections
     ===================================================== -->

<bean id="embeddedAmqConnectionFactory"  class="org.apache.activemq.ActiveMQConnectionFactory" depends-on="appEmbeddedBroker" >
    <property name="brokerURL" value="${msg.embedded.broker.url}" />
    <property name="userName" value="${msg.embedded.client.app.username}" />
    <property name="password" value="${msg.embedded.broker.app.password}" />
</bean>

<bean id="embeddedAmqPooledConnectionFactory" class="org.apache.activemq.pool.PooledConnectionFactory" >
    <property name="connectionFactory" ref="embeddedAmqConnectionFactory" />
    <property name="maxConnections" value="${msg.embedded.client.connection.pool.max}" />
</bean>

查询bean时:

$>get QueueSize EnqueueCount DequeueCount                                                                                  

#mbean = org.apache.activemq:brokerName=app-embedded,destinationName=the.queue.name.local,destinationType=Queue,type=Broker:

QueueSize = 0;


EnqueueCount = 17821;


DequeueCount = 17821;

还有:

#mbean = org.apache.activemq:brokerName=app-embedded,type=Broker:

StoreLimit = 209715200;


StorePercentUsage = 100;


TempLimit = 41943040;


TempPercentUsage = 0;

因此，问题是:为什么即使所有消息都出列，我们仍然会看到 100% 的存储使用率？

Answer 1

造成这种情况的原因可能有很多，具体取决于您的代理配置和使用情况。首先要检查 DLQ 的内容是什么，然后在控制台上查找保留消息的其他目标。由于事物通过 KahaDB 日志日志传播的方式，它们可能会创建一条无法破坏的链，除非您清除其他一些队列或回滚任何仍在进行中的事务等。

有一个guide ActiveMQ 站点上介绍了如何开始调试应该帮助您入门的情况。

另请尝试运行最新的代理版本，这些组件中已修复了一些尝试解决此类情况的问题。