架构图

image.png
搭建高可用，请点击

Broker负载均衡

static networkConnector是用于创建一个静态的配置对于网络中的多个Broker做集群，
这种协议用于复合url，一个复合url包括多个url地址。
集群一broker下的配置，配置集群二的IP地址
<networkConnectors><networkConnector name="local network"  duplex="true" uri="static://(tcp://172.18.87.147:61616,tcp://172.18.87.148:61616,tcp://172.18.87.146:61616)"/>
        </networkConnectors>

集群二broker下的配置，配置集群一的IP地址
<networkConnectors>
             <networkConnector name="local network"  duplex="true"
              uri="static://(tcp://172.18.87.147:61617,tcp://172.18.87.148:61617,tcp://172.18.87.146:61617)"/>
        </networkConnectors>

提醒

如果是三台机器搭建，需要修改点如下：

1. jetty端口
2. activemq集群内部端口bind

 <persistenceAdapter>
            <!--<kahaDB directory="${activemq.data}/kahadb"/>-->
                        <replicatedLevelDB 
            directory="${activemq.data}/leveldb" 
            replicas="3" 
            bind="tcp://0.0.0.0:62623" 
            zkAddress="172.18.87.147:2181,172.18.87.148:2181,172.18.87.146:2181" 
            hostname="172.18.87.147" 
            zkPath="/root/activemq/leveldb-stores"/> 
        </persistenceAdapter>

3. 协议端口

<transportConnector name="openwire" uri="tcp://0.0.0.0:61616?maximumConnections=1000&amp;wireFormat.maxFrameSize=104857600"/>
<transportConnector name="amqp" uri="amqp://0.0.0.0:5672?maximumConnections=1000&amp;wireFormat.maxFrameSize=104857600"/>
<transportConnector name="stomp" uri="stomp://0.0.0.0:61613?maximumConnections=1000&amp;wireFormat.maxFrameSize=104857600"/>
<transportConnector name="mqtt" uri="mqtt://0.0.0.0:1883?maximumConnections=1000&amp;wireFormat.maxFrameSize=104857600"/>
<transportConnector name="ws" uri="ws://0.0.0.0:61614?maximumConnections=1000&amp;wireFormat.maxFrameSize=104857600"/>

代码实现

https://www.jianshu.com/p/b29441bb80ac

“丢失”的消息

针对“丢失”的消息，配置replayWhenNoConsumers选项

<policyEntries>
        <policyEntry queue=">" enableAudit="false">
                <networkBridgeFilterFactory>
                        <conditionalNetworkBridgeFilterFactory replayWhenNoConsumers="true"/>
                </networkBridgeFilterFactory>
        </policyEntry>
</policyEntries>

参考资料
“丢失”的消息