ActiveMQ笔记(4)：搭建Broker集群(cluster)

上一篇介绍了基于Networks of Borkers的2节点HA方案，这一篇继续来折腾Networks of Brokers，当应用规模日渐增长时，2节点的broker可能仍然抗不住访问压力，这时候就需要多加一些broker，弄一个更大规模的Broker集群，但是怎么合理设置broker之间的网络桥接，却是有讲究的，先来看一种不太好的设计：

这个架构看上去没瑕疵，没毛病，3个broker之间两两互通，整体可用性极高，但是从消息的路由角度来看，却不是一个好的设计，当PRoducer向broker1发送一条消息时，Consumer得到消息的路径可能有如下2条：

a） producer -> broker1 -> broker2 

b） producer -> broker1 -> broker3 -> broker2  

当broker更多时，情况会更复杂，比如下面这张图：

消息的路由途径将会更多：

a) producer -> broker1 -> broker4

b) producer -> broker1 -> broker2 -> broker4

c) producer -> broker1 -> broker2 -> broker3 -> broker4

d) producer -> broker1 -> broker3 -> broker4

不难想像，每多经过一个节点，消息处理的延时将会增加一些，如果Broker越多，情况越复杂，最终系统对外表现为消息处理有时很快，有时很慢，整体性能很不稳定，所以实际生产中，不要采用所有Broker之间两两互连的方案。

合理的方案如下：

这张图的灵感，应该来自组建局域网中的星形网络，在中心放置一个Borker充当Hub，与其它所有Broker互连，这样不管Consumer连接到外围的哪个Broker，消息的路由途径都比较稳定(最多经过3个Broker)，这种架构性能虽然稳定了，但是中心的Hub就变成单点隐患，如果中间的DockerHub挂了，整个系统也就废了。

改进后的架构如下：

本质上仍然是一个星形网络，只不过将hub弄成二个互备，然后每个hub都与其它外围的broker相连，消费者连接到broker1/broker2/broker3，生产者(Producer)连接到hub1/hub2，消息的最长路径不超过3个broker (注：生产者也可以连接到broker1/2/3，与消费者一样，但是消息经过的最长路径会变成4)

如果以后要扩张，比如增加了Broker4,Broker5…，直接修改hub1/2上的配置，增加与新的broker的连接即可，不影响消息的路由途径长度。

最后，在本机演练一把，给出一些配置示例：

1、端口规划

共5个activemq实例，端口61616、61626、61636为broker1、broker2、broker3，61645、61656为broker-hub1、broker-hub2

2、activemq.xml配置

以boker1为例，配置文件内容如下：

 1 <beans 2         xmlns="http://www.springframework.org/schema/beans" 3         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" 4         xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd 5   http://activemq.apache.org/schema/core http://activemq.apache.org/schema/core/activemq-core.xsd"> 6  7   <bean class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer"> 8     <property name="locations"> 9       <value>file:${activemq.conf}/credentials.properties</value>10     </property>11   </bean>12 13   <broker xmlns="http://activemq.apache.org/schema/core" brokerName="broker1">  14     <persistenceAdapter>15       <kahaDB directory="${activemq.data}/kahadb"/>16     </persistenceAdapter>17     <transportConnectors>18       <transportConnector name="openwire"19                           uri="tcp://0.0.0.0:61616?maximumConnections=1000&amp;wireFormat.maxFrameSize=104857600"/>20     </transportConnectors>21   </broker>22 23   <import resource="jetty.xml"/>24 </beans>
View Code
broker2及broker3，大家参考该配置修改端口号及brokerName即可。
broker-hub1的配置：
 1 <beans 2         xmlns="http://www.springframework.org/schema/beans" 3         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" 4         xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd 5   http://activemq.apache.org/schema/core http://activemq.apache.org/schema/core/activemq-core.xsd"> 6  7   <bean class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer"> 8     <property name="locations"> 9       <value>file:${activemq.conf}/credentials.properties</value>10     </property>11   </bean>12 13   <broker xmlns="http://activemq.apache.org/schema/core" brokerName="broker-hub1">14     <networkConnectors>15       <networkConnector uri="static:(tcp://127.0.0.1:61656,tcp://127.0.0.1:61616,tcp://127.0.0.1:61626,tcp://127.0.0.1:61636)" duplex="true"/>      16     </networkConnectors>17     <persistenceAdapter>18       <kahaDB directory="${activemq.data}/kahadb"/>19     </persistenceAdapter>20     <transportConnectors>21       <transportConnector name="openwire"22                           uri="tcp://0.0.0.0:61646?maximumConnections=1000&amp;wireFormat.maxFrameSize=104857600"/>23     </transportConnectors>24   </broker>25 26   <import resource="jetty.xml"/>27 </beans>
broker-hub2的配置：
 1 <beans 2         xmlns="http://www.springframework.org/schema/beans" 3         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" 4         xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd 5   http://activemq.apache.org/schema/core http://activemq.apache.org/schema/core/activemq-core.xsd"> 6  7   <bean class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer"> 8     <property name="locations"> 9       <value>file:${activemq.conf}/credentials.properties</value>10     </property>11   </bean>12 13   <broker xmlns="http://activemq.apache.org/schema/core" brokerName="broker-hub2">14     <networkConnectors>   15       <networkConnector uri="static:(tcp://127.0.0.1:61616,tcp://127.0.0.1:61626,tcp://127.0.0.1:61636)" duplex="true"/>16     </networkConnectors>17     <persistenceAdapter>18       <kahaDB directory="${activemq.data}/kahadb"/>19     </persistenceAdapter>20     <transportConnectors>21       <transportConnector name="openwire"22                           uri="tcp://0.0.0.0:61656?maximumConnections=1000&amp;wireFormat.maxFrameSize=104857600"/>23     </transportConnectors>24   </broker>25 26   <import resource="jetty.xml"/>27 </beans>
3、java代码中spring配置文件
a) 生产者
 1 <bean id="jmsFactory" class="org.apache.activemq.pool.PooledConnectionFactory" destroy-method="stop"> 2     <property name="connectionFactory"> 3         <bean class="org.apache.activemq.ActiveMQConnectionFactory"> 4             <!--broker服务的地址--> 5             <property name="brokerURL" value="failover:(tcp://localhost:61646,tcp://localhost:61656)"/> 6             <!--默认值为1000,如果不需要这么大,可以调小--> 7             <property name="maxThreadPoolSize" value="100"/> 8             <!--<property name="userName" value="system"/>--> 9             <!--<property name="passWord" value="manager"/>-->10         </bean>11     </property>12 </bean>
b) 消费者
 1 <bean id="jmsFactory" class="org.apache.activemq.pool.PooledConnectionFactory" destroy-method="stop"> 2     <property name="connectionFactory"> 3         <bean class="org.apache.activemq.ActiveMQConnectionFactory"> 4             <!--broker服务的地址--> 5             <property name="brokerURL" value="failover:(tcp://localhost:61616,tcp://localhost:61626,tcp://localhost:61636)"/> 6             <!--默认值为1000,如果不需要这么大,可以调小--> 7             <property name="maxThreadPoolSize" value="100"/> 8             <!--<property name="userName" value="system"/>--> 9             <!--<property name="password" value="manager"/>-->10         </bean>11     </property>12 </bean>