ActiveMQ筆記(4)：搭建Broker集群(cluster) - yjmyzz－IT工程師數位筆記本

文章出處

上一篇介紹了基于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、端口規劃

activemq1: 61616 (broker1)
activemq2: 61626 (broker2)
activemq3: 61636 (broker3)
activemq4: 61646 (broker-hub1)
activemq5: 61656 (broker-hub2)

共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>

View Code

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>

View Code

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>

View Code

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>