ActiveMQ双节点Broker集群客户端连接负载均衡异常问题咨询

阿华AIGC实验室

2026-5-14

解决ActiveMQ双节点集群客户端仅连接单一Broker的问题

看起来你的配置方向是对的，但可能在Broker配置细节、客户端故障转移URL的使用或者集群拓扑验证上有遗漏。我来一步步帮你排查和解决：

1. 确认ActiveMQ Broker的传输连接器配置

首先检查两个Broker节点的activemq.xml中，stomp+ssl传输连接器是否正确启用了集群相关参数。注意XML中&要转义为&：

<transportConnectors>
    <transportConnector name="stomp+ssl" 
        uri="stomp+ssl://0.0.0.0:61613?
            updateClusterClients=true&amp;
            rebalanceClusterClients=true&amp;
            updateClusterClientsOnRemove=true&amp;
            allowLinkStealing=true"/>
</transportConnectors>

updateClusterClients=true：Broker会主动向客户端推送集群拓扑变化
rebalanceClusterClients=true：允许客户端收到拓扑更新后重连到其他Broker
updateClusterClientsOnRemove=true：当Broker从集群移除时通知客户端
allowLinkStealing=true：帮助处理客户端重连时的连接抢占问题

同时要确保两个Broker的网络连接器（networkConnector）配置正确，确保它们能互相发现：

<networkConnectors>
    <!-- 节点1配置连接节点2 -->
    <networkConnector uri="static:(tcp://remotehost:61666)" duplex="true" />
    <!-- 节点2配置连接节点1 -->
    <networkConnector uri="static:(tcp://localhost:61666)" duplex="true" />
</networkConnectors>

重启两个Broker，查看日志确认集群拓扑已成功建立（日志中会出现类似"Established network connection to ..."的信息）。

2. 修正客户端故障转移连接方式

你当前的代码是直接传入两个主机地址，但stomp.py的Connection11默认会按顺序尝试连接，不会自动做负载均衡或随机选择。要利用ActiveMQ的故障转移协议实现初始随机连接和重平衡，需要使用故障转移URI格式：

import stomp
import ssl

# 使用故障转移URI，开启随机初始连接
conn = stomp.Connection11(
    host_and_ports=None,
    transport=stomp.transport.Transport(
        uri="failover:(stomp+ssl://localhost:61613,stomp+ssl://remotehost:61613)?randomize=true&initialReconnectDelay=1000&maxReconnectDelay=5000"
    ),
    heartbeats=(10000, 20000),
    encoding="utf-8",
    heart_beat_receive_scale=2.0
)

# 配置SSL（根据你的证书情况调整）
conn.set_ssl(
    for_hosts=[("localhost", 61613), ("remotehost", 61613)],
    ssl_version=ssl.PROTOCOL_TLS,
    # 如果需要证书验证，添加以下参数：
    # ca_certs="/path/to/ca.crt",
    # cert_file="/path/to/client.crt",
    # key_file="/path/to/client.key"
)

conn.start()

randomize=true：客户端初始连接时会随机选择集群中的Broker，避免全部连同一个节点
故障转移协议会自动处理Broker的上下线，配合Broker的rebalanceClusterClients参数，在集群拓扑变化时触发客户端重连到其他节点

3. 验证单个Broker的可连接性

先单独测试客户端能否连接到第二个Broker，排除SSL或网络问题：

import stomp
import ssl

conn = stomp.Connection11([("remotehost", 61613)], heartbeats=(10000, 20000), encoding="utf-8")
conn.set_ssl(for_hosts=[("remotehost", 61613)], ssl_version=ssl.PROTOCOL_TLS)
try:
    conn.start()
    print("成功连接到remotehost的Broker")
    conn.disconnect()
except Exception as e:
    print(f"连接失败: {e}")

如果连接失败，优先排查第二个Broker的SSL配置、防火墙规则或Broker运行状态。