上一篇介绍了基于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为例,配置文件内容如下:
![](https://yqfile.alicdn.com/img_1c53668bcee393edac0d7b3b3daff1ae.gif?x-oss-process=image/resize,w_1400/format,webp)
![](https://yqfile.alicdn.com/img_405b18b4b6584ae338e0f6ecaf736533.gif?x-oss-process=image/resize,w_1400/format,webp)
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&wireFormat.maxFrameSize=104857600"/> 20 </transportConnectors> 21 </broker> 22 23 <import resource="jetty.xml"/> 24 </beans>
broker2及broker3,大家参考该配置修改端口号及brokerName即可。
broker-hub1的配置:
![](https://yqfile.alicdn.com/img_1c53668bcee393edac0d7b3b3daff1ae.gif?x-oss-process=image/resize,w_1400/format,webp)
![](https://yqfile.alicdn.com/img_405b18b4b6584ae338e0f6ecaf736533.gif?x-oss-process=image/resize,w_1400/format,webp)
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&wireFormat.maxFrameSize=104857600"/> 23 </transportConnectors> 24 </broker> 25 26 <import resource="jetty.xml"/> 27 </beans>
broker-hub2的配置:
![](https://yqfile.alicdn.com/img_1c53668bcee393edac0d7b3b3daff1ae.gif?x-oss-process=image/resize,w_1400/format,webp)
![](https://yqfile.alicdn.com/img_405b18b4b6584ae338e0f6ecaf736533.gif?x-oss-process=image/resize,w_1400/format,webp)
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&wireFormat.maxFrameSize=104857600"/> 23 </transportConnectors> 24 </broker> 25 26 <import resource="jetty.xml"/> 27 </beans>
3、java代码中spring配置文件
a) 生产者
![](https://yqfile.alicdn.com/img_1c53668bcee393edac0d7b3b3daff1ae.gif?x-oss-process=image/resize,w_1400/format,webp)
![](https://yqfile.alicdn.com/img_405b18b4b6584ae338e0f6ecaf736533.gif?x-oss-process=image/resize,w_1400/format,webp)
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) 消费者
![](https://yqfile.alicdn.com/img_1c53668bcee393edac0d7b3b3daff1ae.gif?x-oss-process=image/resize,w_1400/format,webp)
![](https://yqfile.alicdn.com/img_405b18b4b6584ae338e0f6ecaf736533.gif?x-oss-process=image/resize,w_1400/format,webp)
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>