ActiveMQ Artemis 2.31.2高连接负载下线程CPU过高及Broker不稳定问题

问题背景 #

搭建了1主1从的ActiveMQ Artemis 2.31.2集群，进行连接接受速率负载测试时，当每秒客户端连接数超过100，出现单线程CPU占用极高，一段时间后Broker完全无响应，同时抛出以下错误栈：

线程CPU占用过高错误栈 #

"Thread-1 (ActiveMQ-remoting-threads-ActiveMQServerImpl::name=0.0.0.0-595556720)" #229 prio=5 os_prio=0 cpu=273210.52ms elapsed=279.75s tid=0x00007ff7b000aee0 nid=0x3287db runnable  [0x00007ff9589e2000]
   java.lang.Thread.State: RUNNABLE
        at org.apache.activemq.artemis.core.server.impl.QueueImpl$SynchronizedIterator.hasNext(QueueImpl.java:4433)
        - locked <0x00000003427b4788> (a org.apache.activemq.artemis.core.server.impl.QueueImpl)
        at org.apache.activemq.artemis.core.server.impl.QueueImpl.iterQueue(QueueImpl.java:2234)
        - locked <0x00000003427b4788> (a org.apache.activemq.artemis.core.server.impl.QueueImpl)
        at org.apache.activemq.artemis.core.server.impl.QueueImpl.deleteMatchingReferences(QueueImpl.java:2189)
        at org.apache.activemq.artemis.core.server.Queue.deleteMatchingReferences(Queue.java:325)
        at org.apache.activemq.artemis.core.server.impl.QueueImpl.deleteMatchingReferences(QueueImpl.java:2184)
        at org.apache.activemq.artemis.core.protocol.mqtt.MQTTStateManager.removeDurableSessionState(MQTTStateManager.java:145)
        at org.apache.activemq.artemis.core.protocol.mqtt.MQTTStateManager.removeSessionState(MQTTStateManager.java:140)
        at org.apache.activemq.artemis.core.protocol.mqtt.MQTTSession.stop(MQTTSession.java:145)
        - locked <0x000000035e6b7608> (a org.apache.activemq.artemis.core.protocol.mqtt.MQTTSession)
        at org.apache.activemq.artemis.core.protocol.mqtt.MQTTConnectionManager.disconnect(MQTTConnectionManager.java:186)
        - locked <0x000000035e6b80e0> (a org.apache.activemq.artemis.core.protocol.mqtt.MQTTSessionState)
        at org.apache.activemq.artemis.core.protocol.mqtt.MQTTFailureListener.connectionFailed(MQTTFailureListener.java:41)
        at org.apache.activemq.artemis.core.protocol.mqtt.MQTTFailureListener.connectionFailed(MQTTFailureListener.java:36)
        at org.apache.activemq.artemis.core.protocol.mqtt.MQTTConnection.fail(MQTTConnection.java:51)
        at org.apache.activemq.artemis.core.remoting.server.impl.RemotingServiceImpl.issueFailure(RemotingServiceImpl.java:617)
        at org.apache.activemq.artemis.core.remoting.server.impl.RemotingServiceImpl.connectionDestroyed(RemotingServiceImpl.java:598)
        at org.apache.activemq.artemis.core.remoting.impl.netty.NettyAcceptor$Listener.connectionDestroyed(NettyAcceptor.java:974)
        at org.apache.activemq.artemis.core.remoting.impl.netty.ActiveMQChannelHandler.lambda$channelInactive$0(ActiveMQChannelHandler.java:89)
        at org.apache.activemq.artemis.core.remoting.impl.netty.ActiveMQChannelHandler$$Lambda$696/0x00007ff90c62d4b8.run(Unknown Source)
        at org.apache.activemq.artemis.utils.actors.OrderedExecutor.doTask(OrderedExecutor.java:57)
        at org.apache.activemq.artemis.utils.actors.OrderedExecutor.doTask(OrderedExecutor.java:32)
        at org.apache.activemq.artemis.utils.actors.ProcessorBase.executePendingTasks(ProcessorBase.java:68)
        at org.apache.activemq.artemis.utils.actors.ProcessorBase$$Lambda$261/0x00007ff90c2b5de0.run(Unknown Source)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(java.base@17.0.13/ThreadPoolExecutor.java:1136)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(java.base@17.0.13/ThreadPoolExecutor.java:635)
        at org.apache.activemq.artemis.utils.ActiveMQThreadFactory$1.run(ActiveMQThreadFactory.java:118)
 
"Thread-423 (ActiveMQ-remoting-threads-ActiveMQServerImpl::name=0.0.0.0-358847552)" #4898 prio=5 os_prio=0 cpu=4531660.89ms elapsed=4619.97s tid=0x00007f2978018420 nid=0x3152ae runnable  [0x00007f288617a000]
   java.lang.Thread.State: RUNNABLE
        at org.apache.activemq.artemis.core.filter.impl.FilterImpl.match(FilterImpl.java:109)
        at org.apache.activemq.artemis.core.server.impl.QueueImpl.iterQueue(QueueImpl.java:2244)
        - locked <0x000000030243f848> (a org.apache.activemq.artemis.core.server.impl.QueueImpl)
        at org.apache.activemq.artemis.core.server.impl.QueueImpl.deleteMatchingReferences(QueueImpl.java:2189)
        at org.apache.activemq.artemis.core.server.Queue.deleteMatchingReferences(Queue.java:325)
        at org.apache.activemq.artemis.core.server.impl.QueueImpl.deleteMatchingReferences(QueueImpl.java:2184)
        at org.apache.activemq.artemis.core.protocol.mqtt.MQTTStateManager.removeDurableSessionState(MQTTStateManager.java:145)
        at org.apache.activemq.artemis.core.protocol.mqtt.MQTTStateManager.removeSessionState(MQTTStateManager.java:140)
        at org.apache.activemq.artemis.core.protocol.mqtt.MQTTSession.stop(MQTTSession.java:145)
        - locked <0x00000003cd2ff8f8> (a org.apache.activemq.artemis.core.protocol.mqtt.MQTTSession)
        at org.apache.activemq.artemis.core.protocol.mqtt.MQTTConnectionManager.disconnect(MQTTConnectionManager.java:186)
        - locked <0x00000003cd301558> (a org.apache.activemq.artemis.core.protocol.mqtt.MQTTSessionState)
        at org.apache.activemq.artemis.core.protocol.mqtt.MQTTFailureListener.connectionFailed(MQTTFailureListener.java:41)
        at org.apache.activemq.artemis.core.protocol.mqtt.MQTTFailureListener.connectionFailed(MQTTFailureListener.java:36)
        at org.apache.activemq.artemis.core.protocol.mqtt.MQTTConnection.fail(MQTTConnection.java:51)
        at org.apache.activemq.artemis.core.remoting.server.impl.RemotingServiceImpl.issueFailure(RemotingServiceImpl.java:617)
        at org.apache.activemq.artemis.core.remoting.server.impl.RemotingServiceImpl.connectionDestroyed(RemotingServiceImpl.java:598)
        at org.apache.activemq.artemis.core.remoting.impl.netty.NettyAcceptor$Listener.connectionDestroyed(NettyAcceptor.java:974)
        at org.apache.activemq.artemis.core.remoting.impl.netty.ActiveMQChannelHandler.lambda$channelInactive$0(ActiveMQChannelHandler.java:89)
        at org.apache.activemq.artemis.core.remoting.impl.netty.ActiveMQChannelHandler$$Lambda$709/0x00007f2a846277b0.run(Unknown Source)
        at org.apache.activemq.artemis.utils.actors.OrderedExecutor.doTask(OrderedExecutor.java:57)
        at org.apache.activemq.artemis.utils.actors.OrderedExecutor.doTask(OrderedExecutor.java:32)
        at org.apache.activemq.artemis.utils.actors.ProcessorBase.executePendingTasks(ProcessorBase.java:68)
        at org.apache.activemq.artemis.utils.actors.ProcessorBase$$Lambda$261/0x00007f2a842b5de0.run(Unknown Source)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(java.base@17.0.13/ThreadPoolExecutor.java:1136)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(java.base@17.0.13/ThreadPoolExecutor.java:635)
        at org.apache.activemq.artemis.utils.ActiveMQThreadFactory$1.run(ActiveMQThreadFactory.java:118)

MQTT状态存储超时错误 #

java.lang.IllegalStateException: AMQ850000: Unable to store MQTT state within given timeout: 5000ms
    at org.apache.activemq.artemis.core.protocol.mqtt.MQTTStateManager.storeSessionState(MQTTStateManager.java:177) ~[artemis-mqtt-protocol-2.31.0.jar:2.31.0]
    at org.apache.activemq.artemis.core.protocol.mqtt.MQTTSubscriptionManager.removeSubscriptions(MQTTSubscriptionManager.java:291) ~[artemis-mqtt-protocol-2.31.0.jar:2.31.0]
    at org.apache.activemq.artemis.core.protocol.mqtt.MQTTSubscriptionManager.clean(MQTTSubscriptionManager.java:...

问题分析 #

1. CPU飙升根源：错误栈显示线程卡在QueueImpl.iterQueue和FilterImpl.match方法，说明在清理MQTT会话时，Broker需要遍历队列中所有消息引用并匹配过滤条件，这个操作会持有队列锁，高并发连接断开时，多个线程竞争队列锁并执行全量扫描，导致CPU资源被耗尽。
2. 状态存储超时：AMQ850000错误是因为队列锁竞争激烈，MQTT会话状态存储操作无法在默认5秒超时时间内完成，进一步加剧Broker的资源消耗。

解决方案 #

• 升级Artemis版本：2.31.x版本存在队列遍历性能瓶颈，后续版本（如2.32.0及以上）优化了会话清理逻辑，引入了批量删除和索引优化，避免全量队列扫描。
• 延长MQTT状态超时时间：在broker.xml中添加或修改<mqtt-state-timeout>参数，给会话状态存储操作更多时间：

  <configuration>
    <core>
      <!-- 其他配置 -->
      <mqtt-state-timeout>10000</mqtt-state-timeout>
    </core>
  </configuration>
  

• 优化队列与会话配置：减少MQTT持久化会话数量，或对会话关联的队列设置更高效的过滤规则，避免全量消息扫描。
• 调整线程池大小：增加remoting线程池的容量，缓解线程竞争压力：

  <configuration>
    <core>
      <!-- 其他配置 -->
      <remoting-threads>
        <pool-size>200</pool-size>
        <max-pool-size>300</max-pool-size>
      </remoting-threads>
    </core>
  </configuration>
  

• 优化测试场景：调整客户端连接逻辑，减少不必要的频繁连接断开，降低Broker的会话清理负载。

内容的提问来源于stack exchange，提问作者dhumketu