Master/Slave + Broker Cluster的Ac

我们在上一篇文章中已经实现了ActiveMQ的高可用集群,但是无法做到负载均衡.接下来我们通过修改配置文件实现支持负载均衡的集群.

步骤一:

基于上一篇文章的环境下,我们还是在同一台CentOS中来完成这个实验.需要创建6个ActiveMQ的节点.要实现多主多从,我们把他们分成两组(Group)
group1
Master Broker(61616)+Slave Borker(61617)+Slave Borker(61618)
group2
Master Broker(61619)+Slave Borker(61620)+Slave Borker(61621)
目录创建如下:

步骤二:

因为我们是在一台机器上做实验,所以不同的ActiveMQ需要有不同的端口配置,具体分配如下:

节点	openwire端口	amqp端口	stomp端口	mqtt端口	ws端口	admin端口
mq1	61616	5672	61613	1883	61614	8161
mq2	61617	5682	61623	1903	61634	8162
mq3	61618	5692	61633	1923	61654	8163
mq4	61619	5702	61643	1933	61664	8165
mq5	61620	5712	61653	1943	61674	8166
mq6	61621	5722	61673	1963	61694	8167

具体配置的修改同学们就自己去修改,我在文章中就不贴出来了.

步骤三:

我们要实现的效果是,我们可以往两个Group中的Master Broker节点发送任意消息.
然后消费端连接任意一台机器都可以获取到所有的消息.
这时候我们需要在配置文件中配置,让两组的Group之间的机器是可见的.
1 ) 在group1中的所有节点需要修改activemq.xml文件,添加如下配置:

vi /usr/local/mqcluster/mq1/conf/activemq.xml:这段配置添加在节点persistenceAdapter前

<networkConnectors>
                <networkConnector uri="static:(tcp://192.168.122.129:61619,tcp://192.168.122.129:61620,tcp://192.168.122.129:61621)" duplex="true"/>
</networkConnectors>

2. 在group2中的所有节点需要修改activemq.xml文件,添加如下配置:

vi /usr/local/mqcluster/mq4/conf/activemq.xml:这段配置添加在节点persistenceAdapter前

<networkConnectors>
                <networkConnector uri="static:(tcp://192.168.122.129:61616,tcp://192.168.122.129:61617,tcp://192.168.122.129:61618)" duplex="true"/>
</networkConnectors>

在group2中的所有节点还需要修改activemq.xml中的persistenceAdapter内容:

<persistenceAdapter>
  <!--<kahaDB directory="${activemq.data}/kahadb"/>-->
   <replicatedLevelDB
      directory="${activemq.data}/leveldb"
      replicas="3"
      bind="tcp://0.0.0.0:0"
      zkAddress="192.168.122.129:2181"
      zkPassword=""
      hostname="localhost"
      sync="local_disk"
      zkPath="/activemq/leveldb-stores/group2"
    />
</persistenceAdapter>

修改zkPath="/activemq/leveldb-stores/group2",其他不用变.
好,到这一步我们所有的配置都配置好了.接下来就做测试了,测试的代码还是原来那套.

测试:

1. 将6台ActiveMQ的示例启动起来.
2. 使用group1的生产者发送3条消息,生产者的地址配置:

ConnectionFactory factory = new ActiveMQConnectionFactory(
                "wolfcode",
                "wolfcode",
                "failover:(tcp://192.168.122.129:61616,tcp://192.168.122.129:61617,tcp://192.168.122.129:61618)?randomize=false"
        );

3. 使用group2的生产者发送3条消息,生产者地址配置:

ConnectionFactory factory = new ActiveMQConnectionFactory(
                "wolfcode",
                "wolfcode",
                "failover:(tcp://192.168.122.129:61619,tcp://192.168.122.129:61620,tcp://192.168.122.129:61621)?randomize=false"
        );

4.运行消费端,是可以把group1和group2的消息消费掉的.消费端地址配置:

ConnectionFactory factory = new ActiveMQConnectionFactory(
                "wolfcode",
                "wolfcode",
                "failover:(tcp://192.168.122.129:61616,tcp://192.168.122.129:61617,tcp://192.168.122.129:61618," +
                        "tcp://192.168.122.129:61619,tcp://192.168.122.129:61620,tcp://192.168.122.129:61621)?randomize=false");