前言

使用静态网络连接(static networkConnector)可以将多个broker连接起来,作为一个集群对外提供服务。
我们假定有amq1和amq2配置了静态网络连接,那么当一个消费者连接到amq2,当生产者发送消息到amq1时,amq2将会作为amq1的一个消费者,将消息转移到amq2,然后投递给连接到amq2的消费者;反之亦然。对于queue,由于一条消息只能被一个消费者消费,所以在上面的情形下,消息将被转移到amq2,然后投递给连接到amq2的消费者;对于topic,amq2从amq1复制消息,然后投递给消费者。

配置

这里准备2台activemq,这里分别以activemq1和activemq2称之。

activemq1修改

1.修改brokerName为amq1;
2.修改openwire端口为61616,其他的transportConnector删除;
3.在destinationPolicy增加下面的配置

1
2
3
4
5
<policyEntry queue=">" enableAudit="false">
    <networkBridgeFilterFactory>
        <conditionalNetworkBridgeFilterFactory replayWhenNoConsumers="true"/>
    </networkBridgeFilterFactory>
</policyEntry>

这是针对queue的配置。该配置是为了解决消息回流的问题。replayWhenNoConsumers=true表示没有消费者时将消息回流到原始的broker。enableAudit=false为了防止消息回流后被当做重复消息而不被分发。

activemq2修改

1.修改brokerName为amq2,2个不要冲突。
2.修改openwire端口为61617,其他的transportConnector删除;
3.修改netty端口为8162;
4.增加下面的配置

1
2
3
<networkConnectors>
    <networkConnector name="bridge" uri="static:(tcp://localhost:61616)" duplex="true" conduitSubscriptions="false"/>
</networkConnectors>

duplex默认为false,单向连接。如果配置为false,那么发送到amq2的消息可以从amq1消费,反之则不行。
conduitSubscriptions默认true,是否把同一个broker的多个consumer当做一个来处理。如果配置为true,假定有1个消费者连接amq1,3个消费者连接amq2,那么amq1的那个消费者将接收到50%的消息,amq2的3个消费者一共接收到50%的消息。所以这里设置为false,这样假定发送100条消息,那么每个消费者消费25条消息。

5.在destinationPolicy增加下面的配置

1
2
3
4
5
<policyEntry queue=">" enableAudit="false">
    <networkBridgeFilterFactory>
        <conditionalNetworkBridgeFilterFactory replayWhenNoConsumers="true"/>
    </networkBridgeFilterFactory>
</policyEntry>

代码

客户端使用failover协议来连接mq。

1
failover:(tcp://127.0.0.1:61616,tcp://127.0.0.1:61626)?randomize=false

指定randomize为false,这样会先连接127.0.0.1:61616,连接不上才会连接后面的地址。

测试

配置duplex=false,测试结果。
a.连接amq1发送消息;

b.消费者连接amq2接收消息

可以发现发送到amq1的消息amq2没法消费。

将duplex改为true,amq2就可以消费amq1的消息了。

其他的就不一一测试了。

注意事项

这里使用2个broker,实际可以根据需要配置多个。多个地址用逗号分隔即可。比如uri=”static:(tcp://localhost:61616,tcp://localhost:61636)”。