CentOS 6.5下搭建ActiveMq集群（静态网络连接）

1.下载安装activemq
从activemq官网http://activemq.apache.org/download.html下载相应的activemq版本。完了之后直接解压之后运行启动命令即可运行activemq服务器。过程十分简单，最简单的启动命令方式如下：[root@centos-server-yzr local]# cd /usr/local/apache-activemq-5.15.0/
[root@centos-server-yzr apache-activemq-5.15.0]# cd bin
[root@centos-server-yzr bin]# ll
total 152
-rwxr-xr-x. 1 root root 21446 Jun 28 2017 activemq
-rwxr-xr-x. 1 root root 6189 Jun 28 2017 activemq-diag
-rw-r--r--. 1 root root 16067 Jun 28 2017 activemq.jar
-rw-r--r--. 1 root root 4884 Jun 28 2017 env
drwxr-xr-x. 2 root root 4096 Feb 21 22:05 linux-x86-32
drwxr-xr-x. 2 root root 4096 Feb 21 22:05 linux-x86-64
drwxr-xr-x. 2 root root 4096 Feb 21 22:05 macosx
-rw-r--r--. 1 root root 83820 Jun 28 2017 wrapper.jar
[root@centos-server-yzr bin]# ./activemq start
INFO: Loading '/usr/local/apache-activemq-5.15.0//bin/env'
INFO: Using java '/usr/local/java/jdk1.8.0_131/bin/java'
INFO: Starting - inspect logfiles specified in logging.properties and log4j.properties to get details
INFO: pidfile created : '/usr/local/apache-activemq-5.15.0//data/activemq.pid' (pid '1255')按这种默认的启动方式，activemq会默认从根目录下的conf目录找到activemq.xml文件，并根据该xml文件的配置进行启动。所以说这种方式的启动等价于./activemq start ../conf/activemq.xml。
另外activemq运行时的日志文件默认是存放在根目录下data/activemq.log下，如果想指定activemq的日志输出位置，可以在activemq启动时使用./activemq start > /tmp/activemqlog 命令。
启动之后可以通过查看activemq的输出日志来判断是否启动成功，又或者可以直接通过浏览器直接访问，那activemq默认使用61616端口提供JMS服务，使用8161端口提供管理控制台服务，这样子说来activemq的管理页面就可以直接通过访问http://192.168.137.201:8161/admin来登陆，第一次访问的时候会弹出登录窗口，activemq默认的账户密码是admin/admin
关闭activemq也简单，直接执行bin目录下的stop即可，或者杀死activemq进程号也行。
2.启动多个broker实例
在一台服务器上如果要启动多个activemq，可以按照以下步骤进行更改：
1)首先复制一份conf目录，重命名为conf2，并且将conf2目录下的activemq.xml和jetty.xml更改一下内容[root@centos-server-yzr conf2]# cd /usr/local/apache-activemq-5.15.0/
[root@centos-server-yzr apache-activemq-5.15.0]# cp -r conf conf2打开conf2/activemq.xml文件之后做一下修改：<broker xmlns="http://activemq.apache.org/schema/core" brokerName="broker2" dataDirectory="${activemq.data}">更改此处brokerName的值，不能和其他broker同名，随便取名为broker2<persistenceAdapter>
<kahaDB directory="${activemq.data}/kahadb2"/>
</persistenceAdapter>activemq数据存放的路径文件不能相同，activemq默认持久化的方式是kahaDB，所以只要将kahaDB中的directory属性值改为${activemq.data}/kahadb2即可。<transportConnectors>
<!-- DOS protection, limit concurrent connections to 1000 and frame size to 100MB -->
<transportConnector name="openwire" uri="tcp://0.0.0.0:61716?maximumConnections=1000&wireFormat.maxFrameSize=104857600"/>
<transportConnector name="amqp" uri="amqp://0.0.0.0:5772?maximumConnections=1000&wireFormat.maxFrameSize=104857600"/>
<transportConnector name="stomp" uri="stomp://0.0.0.0:61663?maximumConnections=1000&wireFormat.maxFrameSize=104857600"/>
<transportConnector name="mqtt" uri="mqtt://0.0.0.0:1873?maximumConnections=1000&wireFormat.maxFrameSize=104857600"/>
<transportConnector name="ws" uri="ws://0.0.0.0:67614?maximumConnections=1000&wireFormat.maxFrameSize=104857600"/>
</transportConnectors>所有<transportConnectors>下的的端口都不能和其他broker一样，更改一下每一个transportConnector的端口即可以保持和其他的broker不一样即可。
打开conf2/jetty.xml文件之后做一下修改：
<bean id="jettyPort" class="org.apache.activemq.web.WebConsolePort" init-method="start">
<!-- the default port number for the web console -->
<property name="host" value="0.0.0.0"/>
<property name="port" value="8181"/>
</bean>jetty.xml里面默认的端口是8161，将此处的端口改为和其他broker不一样即可，此处随便改为8181.
2)在根目录下的bin目录下，复制一份activemq，重命名为activemq2即可
更改ACTIVEMQ_PIDFILE="$ACTIVEMQ_DATA/activemq2.pid"# Location of the pidfile
if [ -z "$ACTIVEMQ_PIDFILE" ]; then
ACTIVEMQ_PIDFILE="$ACTIVEMQ_DATA/activemq2.pid"
fi修改ACTIVEMQ_CONF="$ACTIVEMQ_BASE/conf2"# Active MQ configuration directory
if [ -z "$ACTIVEMQ_CONF" ] ; then

# For backwards compat with old variables we let ACTIVEMQ_CONFIG_DIR set ACTIVEMQ_CONF
if [ -z "$ACTIVEMQ_CONFIG_DIR" ] ; then
ACTIVEMQ_CONF="$ACTIVEMQ_BASE/conf2"
else
ACTIVEMQ_CONF="$ACTIVEMQ_CONFIG_DIR"
fi
fi到此就完成了一个不一样的broker了，分别使用./activemq start 和./activemq2 start启动两个broker。





3.静态网络链接
经过上面的操作之后得到了两个互不关联的broker，他们之间是相互独立的。activemq中有一个static networks静态网络的概念，它可用于创建一个静态的配置对于网络中 的多个broker。这种协议 用于复合url，一个url包括多个url地址，格式为如下写法:static:(uri1,uri2,uri3)?key=value<networkConnectors>
<networkConnector name="local network"
uri="static://(tcp://192.168.137.201:61616,tcp://192.168.137.201:61716)" />
</networkConnectors>在localhost的broker上的activemq.xml添加上面配置之后，就可以达到将broker2关联到localhost的broker上，并且这种关联是单向的，只能从配置方即localhost指向broker2.这就意味着localhost上的消息可以被broker2上的消费者消费，反之则不行，broker2上的消息不能被localhost的消费者消费。
为了校验上面的说法，通过代码来验证一下：先启动两个broker实例，然后往localhost 61616这个broker发送消息，然后再61716端口的broker2接收消息。log4j:WARN No appenders could be found for logger (org.apache.activemq.transport.WireFormatNegotiator).
log4j:WARN Please initialize the log4j system properly.
log4j:WARN See http://logging.apache.org/log4j/1.2/faq.html#noconfig for more info.
发送消息：ActiveMq MessageContent 1
发送消息：ActiveMq MessageContent 2
发送消息：ActiveMq MessageContent 3
发送消息：ActiveMq MessageContent 4
发送消息：ActiveMq MessageContent 5

Process finished with exit code 0

log4j:WARN No appenders could be found for logger (org.apache.activemq.transport.WireFormatNegotiator).
log4j:WARN Please initialize the log4j system properly.
log4j:WARN See http://logging.apache.org/log4j/1.2/faq.html#noconfig for more info.
61716收到消息：ActiveMq MessageContent 1
61716收到消息：ActiveMq MessageContent 2
61716收到消息：ActiveMq MessageContent 3
61716收到消息：ActiveMq MessageContent 4
61716收到消息：ActiveMq MessageContent 5

此时在http://192.168.137.201:8161/admin/network.jsp中能查看到桥接的信息：



经过上面的校验之后，这里总结一下结论：
此时localhost broker能单向的关联到在静态网络中配置的其他broker，这当中启到的作用是localhost broker的消息能够被关联到的broker的消费者消费。除此之外，经过多次测试一下我们会发现存在两个问题：
1）localhost broker的消息基本上都会流向其他关联到的broker，这导致localhost broker的消费者基本处于闲置状态
2）如果其他关联到的broker上没有消费者，那么此时localhost broker的消息将会在永远停留在关联到的且没有消费者的broker上，直至该broker上有可用的消费者为止。也就是说在其他broker上的消息不会流回localhost broker
为了解决这两个问题，需要将静态网络连接配置成双向的，能让消息能够回流。
配置静态网络连接为双向：
在上图中可以发现静态网络连接的create by duplex参数，正是此参数用来设置关联中的broker是双向的。<networkConnectors>
<networkConnector name="local network" duplex="true"
uri="static://(tcp://192.168.137.201:61616,tcp://192.168.137.201:61716)" />
</networkConnectors>并且在其他的broker中也加入该配置。这里就在broker2中的conf2下的activemq.xml上也加入上面的配置。
此时在两个broker中发送的消息都能在关联的其他broker中消费。
配置静态网络消息回流：<policyEntry queue=">" enableAudit="false">
<networkBridgeFilterFactory>
<conditionalNetworkBridgeFilterFactory replayWhenNoConsumers ="true" />
</networkBridgeFilterFactory>
</policyEntry>enableAudit="false"是防止将回流消息当做重复消息而不被分发，replayWhenNoConsumers="true"是设置当没有消费者时将消息回流。
经过测试之后，activemq正常一般情况下会将消息分成两半，一半给localhost broker，一半给broker2.不管两个broker下有多少个consumer，都会被当做一个，这显然不太合理，在networkConnector节点上有conduitSubscriptions属性，默认值为true，其作用是否把同一个broker的多个consumer当做一个来处理。在这里只需要将这个设置为false即可
4.容错的连接
在上面的探讨中已经初步实现了消费者的集群，但是在发送端中还是最原始的往一点地方发，万一当某一个broker宕机之后，将会导致发送消息失败，而整个消费者集群再强大也没有任何用处，所以，activemq提供了一个支持生产消费的容错链接属性failover。
Failover协议实现了自动重新链接的逻辑，提供一个static的可用的Broker列表，配置方式为:failover:(uri1,uri2...)?key=value
默认情况下，这种协议用于随机的去选择一个链接去链接，如果链接失败了，那么会链接到其他的broker上，默认的配置定义了延迟重新链接，意味着传输将会在10秒后自动的去重新链接可用的broker，当然所有的重新链接参数都可以根据应用的需要而配置。connectionFactory = new ActiveMQConnectionFactory(user,password,"failover:(tcp://192.168.137.201:61616,tcp://192.168.137.201:61716)");把61616端口的localhost broker关闭，然后测试上面的容错链接是否生效。
此时会发现即使61616端的broker断开之后，在加上容错链接之后还是能够成功的往activemq集群中发送消息，且这些消息能够被关联到的broker下的消费者进行消费，即使被断开的broker的消费者处于连接失败状态，也不会影响到集群其他生产者和消费者的运行，从而不会影响到业务，一个高可用的消息中间件集群就初步完成了。
5.代码
本文的相关代码可以参考我之前的两篇文章：
http://blog.csdn.net/yzr_java/article/details/63704695

http://blog.csdn.net/yzr_java/article/details/65450025