ActiveMQ集群主队列与镜像队列入队消息数量不一致问题求助

针对你遇到的ActiveMQ集群中镜像队列消息数多于主队列、存在重复消息的问题，结合你的配置和使用场景，以下是具体分析和解决办法：

核心原因 #

1. 复合队列转发与集群网络转发的叠加重复 #

你配置的复合队列（compositeQueue）会将主队列myqueue的消息转发到myqueueMirror，同时集群的networkConnector会在节点间转发myqueue的消息（当其他节点存在该队列的消费者时）。如果两个节点都配置了相同的复合队列规则，那么转发到另一个节点的myqueue消息会再次被该节点的复合队列转发到本地的myqueueMirror，导致镜像队列的消息数叠加（比如主节点的10000条 + 跨节点转发的350条 = 10350条）。

2. 消息循环转发风险 #

当前networkTTL=2的配置允许消息在集群中最多经过2次网络跳转，可能导致消息在节点间循环转发（例如Node1→Node2→Node1），循环回来的消息会再次触发复合队列的转发逻辑，增加镜像队列的重复消息。

3. STOMP协议的生产者重发可能 #

如果Python生产者使用自动确认（auto-ack）机制，当网络波动导致生产者未收到Broker的ACK时，会重新发送消息。虽然主队列可能因某些机制（如消息ID去重，但ActiveMQ默认不开启）避免重复，但镜像队列会因转发逻辑存储重复消息。

解决方案 #

1. 调整复合队列与集群转发的冲突 #

• 排除主队列的跨节点转发：在networkConnector中添加excludedDestinations配置，禁止myqueue在节点间转发，确保所有消息仅在单个节点的myqueue中处理，避免跨节点的二次转发。修改后的networkConnector示例：

<networkConnectors>
  <networkConnector uri="static://(tcp://node2:61617)?useExponentialBackOff=false"
    name="node1-node2"
    userName="cluster"
    password="12345678"
    duplex="false"
    networkTTL="1"
    decreaseNetworkConsumerPriority="true"
    conduitSubscriptions="true"
    suppressDuplicateQueueSubscriptions="true">
    <excludedDestinations>
      <queue physicalName="myqueue"/>
    </excludedDestinations>
  </networkConnector>
</networkConnectors>

• 仅在单个节点配置复合队列：如果不需要跨节点的镜像队列同步，只在其中一个节点配置复合队列规则，避免两个节点同时触发转发逻辑。

2. 使用原生镜像队列替代复合队列 #

ActiveMQ原生支持镜像队列功能，无需手动配置复合队列，可避免自定义转发带来的重复问题。在Broker配置中添加以下规则：

<destinationPolicy>
  <policyMap>
    <policyEntries>
      <policyEntry queue="myqueue" mirrorQueue="true"/>
    </policyEntries>
  </policyMap>
</destinationPolicy>

此配置会自动创建myqueue.mirror作为镜像队列，Broker会负责消息的镜像同步，确保主队列与镜像队列的消息数一致。

3. 优化网络连接器配置 #

• 将networkTTL改为1，限制消息仅能进行一次跨节点转发，避免循环转发的风险。
• 启用duplex="true"，单个networkConnector即可实现双向通信，简化配置同时减少转发逻辑冲突。

4. 优化STOMP生产者的可靠性 #

在Python的STOMP客户端中使用客户端确认机制（client-ack），避免因网络问题导致的消息重发：

import stomp

class MyListener(stomp.ConnectionListener):
    def on_message(self, frame):
        # 处理消息后手动确认
        conn.ack(frame.headers['message-id'], frame.headers['subscription'])

conn = stomp.Connection([('node1', 61613)])
conn.set_listener('', MyListener())
conn.connect('user', 'pass', wait=True)

# 发送消息时指定客户端确认
conn.send(destination='/queue/myqueue', body='test message', ack='client')

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