ActiveMQ集群：网络连接模式(network connector)详解

悠悠 2022-06-12 05:08 515阅读 0赞

# 网络连接模式(network connector) #

针对海量消息所要求的横向扩展性和系统的高可用性，ActiveMQ提供了网络连接模式的集群功能。简单的说，就是通过把多个不同的broker实例连接在一起，作为一个整体对外提供服务，从而提高整体对外的消息服务能力。通过这种方式连接在一起的broker实例之间，可以共享队列和消费者列表，从而达到分布式队列的目的。

## 拓扑结构 ##

几种不同的ActiveMQ部署拓扑结构（嵌入、主从复制、网络连接）：

![BrokerTopology-1.png][]

## 配置示例 ##

在activemq.xml的broker节点内添加：

<networkConnectors>

<networkConnectoruri="static:(tcp://localhost:62001)"/>

</networkConnectors>

uri也可以使用其他两种方式：

1.          multicast://default

2.          masterslave:(tcp://host1:61616,tcp://host2:61616,tcp://..)

其中masterslave方式的第一个url需要是master，其他是slave。

一个broker的实例内可以配置多个networkConnector，如果有两个以上的networkConnector指向同一个broker，则需要显式的指定name。

## 静态URI配置 ##

使用静态URI方式可以指定多个URL，networkConnector将连接到每一个broker。

<networkConnectors>

<networkConnector uri="static:(tcp://host1:61616,tcp://host2:61616,tcp://..)"/>

</networkConnectors>

URI的几个属性：

<table style="color:rgb(51,51,51); font-family:Arial; font-size:14px"> 
 <tbody> 
  <tr> 
   <td style="background:rgb(240,240,240)"> <p><strong>属性</strong></p> </td> 
   <td style="background:rgb(240,240,240)"> <p><strong>默认值</strong></p> </td> 
   <td style="background:rgb(240,240,240)"> <p><strong>描述</strong></p> </td> 
  </tr> 
  <tr> 
   <td> <p>initialReconnectDelay</p> </td> 
   <td> <p>1000</p> </td> 
   <td> <p>重连之前等待的时间(ms) (如果useExponentialBackOff为 false)</p> </td> 
  </tr> 
  <tr> 
   <td> <p>maxReconnectDelay</p> </td> 
   <td> <p>30000</p> </td> 
   <td> <p>重连之前等待的最大时间(ms)</p> </td> 
  </tr> 
  <tr> 
   <td> <p>useExponentialBackOff</p> </td> 
   <td> <p>true</p> </td> 
   <td> <p>每次重连失败时是否增大等待时间</p> </td> 
  </tr> 
  <tr> 
   <td> <p>backOffMultiplier</p> </td> 
   <td> <p>2</p> </td> 
   <td> <p>增大等待时间的系数</p> </td> 
  </tr> 
 </tbody> 
</table>

## networkConnector配置 ##

### 配置参数 ###

<table style="color:rgb(51,51,51); font-family:Arial; font-size:14px"> 
 <tbody> 
  <tr> 
   <td style="background:rgb(240,240,240)"> <p><strong>属性</strong></p> </td> 
   <td style="background:rgb(240,240,240)"> <p><strong>默认值</strong></p> </td> 
   <td style="background:rgb(240,240,240)"> <p><strong>描述</strong></p> </td> 
  </tr> 
  <tr> 
   <td> <p>name</p> </td> 
   <td> <p>bridge</p> </td> 
   <td> <p>名称</p> </td> 
  </tr> 
  <tr> 
   <td> <p>dynamicOnly</p> </td> 
   <td> <p>false</p> </td> 
   <td> <p>如果为true, 持久订阅被激活时才创建对应的网路持久订阅。默认是启动时激活。</p> </td> 
  </tr> 
  <tr> 
   <td> <p>decreaseNetworkConsumerPriority</p> </td> 
   <td> <p>false</p> </td> 
   <td> <p>如果为true，网络的消费者优先级降低为-5。如果为false，则默认跟本地消费者一样为0.</p> </td> 
  </tr> 
  <tr> 
   <td> <p>networkTTL</p> </td> 
   <td> <p>1</p> </td> 
   <td> <p>消息和订阅在网络上通过的broker数量</p> </td> 
  </tr> 
  <tr> 
   <td> <p><span style="color:red">conduitSubscriptions</span></p> </td> 
   <td> <p><span style="color:red">true</span></p> </td> 
   <td> <p><span style="color:red">多个网络消费者是否被当做一个消费者来对待。</span></p> </td> 
  </tr> 
  <tr> 
   <td> <p>excludedDestinations</p> </td> 
   <td> <p>empty</p> </td> 
   <td> <p>不通过网络转发的destination</p> </td> 
  </tr> 
  <tr> 
   <td> <p>dynamicallyIncludedDestinations</p> </td> 
   <td> <p>empty</p> </td> 
   <td> <p>通过网络转发的destinations，注意空列表代表所有的都转发。</p> </td> 
  </tr> 
  <tr> 
   <td> <p>staticallyIncludedDestinations</p> </td> 
   <td> <p>empty</p> </td> 
   <td> <p>匹配的都将通过网络转发-即使没有对应的消费者</p> </td> 
  </tr> 
  <tr> 
   <td> <p><span style="color:red">duplex</span></p> </td> 
   <td> <p><span style="color:red">false</span></p> </td> 
   <td> <p><span style="color:red">如果为true</span>，则既可消费又可生产消息到网络broker</p> </td> 
  </tr> 
  <tr> 
   <td> <p>prefetchSize</p> </td> 
   <td> <p>1000</p> </td> 
   <td> <p>设置网络消费者的<a href="http://activemq.apache.org/what-is-the-prefetch-limit-for.html" title="What is the prefetch limit for" style="color:rgb(51,102,153); text-decoration:none" rel="nofollow"><span style="color:rgb(136,0,85)">prefetch size</span></a>参数。必须大于0，因为网络消费者不能自己轮询消息。</p> </td> 
  </tr> 
  <tr> 
   <td> <p>suppressDuplicateQueueSubscriptions</p> </td> 
   <td> <p>false</p> </td> 
   <td> <p>(从5.3版本开始) 如果为true, 重复的订阅关系一产生即被阻止。</p> </td> 
  </tr> 
  <tr> 
   <td> <p>bridgeTempDestinations</p> </td> 
   <td> <p>true</p> </td> 
   <td> <p>是否广播advisory messages来创建临时destination。</p> </td> 
  </tr> 
  <tr> 
   <td> <p>alwaysSyncSend</p> </td> 
   <td> <p>false</p> </td> 
   <td> <p>(从 5.6版本开始) 如果为true，非持久化消息也将使用request/reply方式代替oneway方式发送到远程broker。</p> </td> 
  </tr> 
  <tr> 
   <td> <p>staticBridge</p> </td> 
   <td> <p>false</p> </td> 
   <td> <p>(从5.6版本开始) 如果为true，只有staticallyIncludedDestinations中配置的destination可以被处理。</p> </td> 
  </tr> 
 </tbody> 
</table>

networkConnector的实现原理是基于ActiveMQ的公告消息（AdvisoryMessage）机制的（参见[此处][Link 1]）。当broker2通过networkConnector、duplex方式指向broker1时，发生了什么呢？

假定broker1已经启动，这时候broker2开始启动。

1.         broker2先启动自己的connector

2.         然后使用一个vm的connector，创建一个connection，把自己作为一个client，连接到broker1。

3.         通过订阅Advisory Message，拿到相互的Consumer与相应的Queue列表。

至此，双方建立关系。

然后通过broker1的转发，broker1上的消费者，就可以消费broker2的queue上的消息。这个过程可以看做一个消息被消费了两次，broker1作为消费者，消费掉broker2上的消息，broker1再作为broker，把消息投递给实际的消费者。

![1356515314_5094.jpg][]

### 管道订阅(conduit subscription) ###

conduitSubscriptions选择决定网络消费者在所有消费者中的比重。假如有2个同一个远程的broker1上的网络消费者和一个broker2的本地消费者。

1.        conduitSubscriptions为true，则2个网络消费者只相当于一个消费者，broker1仅仅在broker2上注册了一个消费者。这时往broker2上发送300个消息，2个网络消费者各接收到75个消息，一个本地消费者接收到150 消息。

2.        conduitSubscriptions为false，则3个消费者平分所有消息，broker1在broker2上将注册了两个消费者。这时往broker2上发送300个消息，2个网络消费者和本地消费者一样，各接收到100个消息。

### 双向网络连接(duplex networkConnector) ###

默认NetworkConnector在需要转发消息时是单向连接的。当duplex=true时，就变成了双向连接，这时配置在broker2端的指向broker1的duplex networkConnector，相当于即配置了

broker2到broker1的网络连接，也配置了broker1到broker2的网络连接。（就是说不管broker1同意与否，都被绑架了。）当然，仅仅在broker1上配置也有同样的效果。

注意：可以在两个broker间建立两个以上的双向网络连接来增加吞吐量或对主题\\队列分区，只需要指定他们使用不同的name即可。

### 指定和限制Destination ###

通过NetworkConnector共享的destination太多，传输的Advisory Message就会变的非常多，系统的拓扑结构将变得非常复杂，所有才有多种方式来限制这些destination配置：

1.        dynamicallyIncludedDestinations

ü  这里匹配到的destination，在需要时将被转发

2.        excludedDestinations

ü  这里匹配到的destination，将不会被转发

3.        staticallyIncludedDestinations

ü  如果指定了staticBridge为true，则只有这里匹配的destination可以被转发。此时本地broker完全被其他broker代理。并且本broker不会订阅其他broker上的AdvisoryMessage，也不会获取任何远程consumer信息。

这几个配置可以使用通配符，比如“>”，详见[wildcard][]。

示例代码：

<networkConnectors>

<networkConnector uri="static:(tcp://localhost:61617)"

name="bridge"

conduitSubscriptions="true"

decreaseNetworkConsumerPriority="false">

<dynamicallyIncludedDestinations>

<queue physicalName="include.test.foo"/>

<topic physicalName="include.test.bar"/>

</dynamicallyIncludedDestinations>

<excludedDestinations>

<queue physicalName="exclude.test.foo"/>

<topic physicalName="exclude.test.bar"/>

</excludedDestinations>

<staticallyIncludedDestinations>

<queue physicalName="always.include.queue"/>

<topic physicalName="always.include.topic"/>

</staticallyIncludedDestinations>

</networkConnector>

</networkConnectors>

此外，从5.6版本起，可以在networkConnector上设置destinationFilter来指定感兴趣的Advisory Message将被传播。

<networkConnector uri="static:(tcp://host)" destinationFilter="Queue.include.test.foo,ActiveMQ.Advisory.Consumer.Topic.include.test.bar">

<dynamicallyIncludedDestinations>

<queue physicalName="include.test.foo"/>

<topic physicalName="include.test.bar"/>

</dynamicallyIncludedDestinations>

</networkConnector>

### 被卡住的消息 ###

一个很有意思的场景是，broker1和broker2通过networkConnector连接。一些个consumers连接到broker1，消费broker2上的消息。消息先被broker1从broker2上消费掉，然后转发给这些consumers。不幸的是转发部分消息的时候broker1重启了，这些consumers发现broker1连接失败，通过failover连接到broker2上去了，但是有一部分他们还没有消费的消息被broker2已经分发到了broker1上去了。这些消息，就好像是消失了，除非有消费者重新连接到broker1上来消费。怎么办呢？

办法就是从5.6版本destinationPolicy上新增的选项replayWhenNoConsumers。这个选项使得broker1上有需要转发的消息但是没有消费者时，把消息回流到它原始的broker。同时把enableAudit设置为false，为了防止消息回流后被当做重复消息而不被分发。

<destinationPolicy>

<policyMap>

<policyEntries>

<policyEntry queue="TEST.>" >

<conditionalNetworkBridgeFilterFactory replayWhenNoConsumers="true" enableAudit="false"/>

</policyEntry>

</policyEntries>

</policyMap>

</destinationPolicy>

更详细的讨论见这里：

[http://tmielke.blogspot.de/2012/03/i-have-messages-on-queue-but-they-dont.html][http_tmielke.blogspot.de_2012_03_i-have-messages-on-queue-but-they-dont.html]

### 其他说明 ###

1.        NetworkConnector基于AdvisoryMessage机制，如果broker的advisorySupport选型被禁用，则NetworkConnector将不起作用。

2.        用作转发的broker中入列出列这些统计信息只记录其转发的数据。

3.        用作转发的broker中无法看到远程broker的相同队列中的数据（browse消息列表为空，queuesize为0）。

## 使用示例 ##

下载ActiveMQ 5.7版本，其中带了包含static network connector的例子。

即配置文件activemq-static-network-broker1.xml和activemq-static-network-broker2.xml。

他们分别使用端口 tcp://localhost:61616和tcp://localhost:61618

static network connector在第二个文件里。

分别使用这两个配置文件启动两个broker实例（先启动broker1，再启动broker2）。

在broker2的控制台看到：

*INFO | Establishing network connection fromvm://static-broker2?async=false&network=true to tcp://localhost:61616*

* INFO |Connector vm://static-broker2 Started*

* INFO |Network Connector DiscoveryNetworkConnector:NC:BrokerService\[static-broker2\]Started*

* INFO |Apache ActiveMQ 5.7.0 (static-broker2, ID:kimmking-2270-1356502079016-0:1)started*

* INFO |For help or more information please see: http://activemq.apache.org*

* INFO |Network connection between vm://static-broker2\#0 andtcp://localhost/127.0.0.1:61616 @2271(static-broker1) has been established.*

在broker1的控制台看到：

*INFO | Network connection betweenvm://static-broker1\#0 and tcp:///127.0.0.1:1710@61616 (static-broker2) has beenestablished.*

在命令行输入jconsole，然后分别通过下列url连接jmx控制台来管理broker：

ü  service:jmx:rmi:///jndi/rmi://localhost:1099/jmxrmi

ü  service:jmx:rmi:///jndi/rmi://localhost:1100/jmxrmi

也可以再这两个xml中配置jetty来使用web控制台查看和管理。

然后可以通过代码在两个broker中通过static network connector存取消息。

![1356515345_1663.jpg][]![1356515352_3637.jpg][]

[测试][Link 2]代码：

**\[java\]**  [view plain][]   [copy][view plain]

1.  **package** org.qsoft.activemq.test;  
2.    
3.  **import** java.util.concurrent.atomic.AtomicInteger;  
4.    
5.  **import** javax.jms.JMSException;  
6.  **import** javax.jms.Message;  
7.  **import** javax.jms.MessageListener;  
8.  **import** javax.jms.MessageProducer;  
9.  **import** javax.jms.Queue;  
10. **import** javax.jms.QueueConnection;  
11. **import** javax.jms.QueueConnectionFactory;  
12. **import** javax.jms.QueueReceiver;  
13. **import** javax.jms.QueueSession;  
14. **import** javax.jms.Session;  
15. **import** javax.jms.TextMessage;  
16.   
17. **import** org.apache.activemq.ActiveMQConnectionFactory;  
18. **import** org.apache.activemq.command.ActiveMQQueue;  
19.   
20. **public** **class** TestReceiver \{  
21.   
22.     /\*\* 
23.      \* @param args 
24.      \*/  
25.     **public** **static** **void** main(String\[\] args) \{  
26.         **try** \{  
27.             // init connection factory with activemq  
28.             QueueConnectionFactory factoryA = **new** ActiveMQConnectionFactory("tcp://127.0.0.1:61616");  
29.             // specify the destination  
30.             Queue queueB = **new** ActiveMQQueue("kk.b");  
31.             // create connection,session,consumer and receive message  
32.             QueueConnection connA = factoryA.createQueueConnection();  
33.             connA.start();  
34.               
35.             // first receiver on broker1  
36.             QueueSession sessionA1 = connA.createQueueSession(**false**, Session.AUTO\_ACKNOWLEDGE);  
37.             QueueReceiver receiverA1 = sessionA1.createReceiver(queueB);  
38.             **final** AtomicInteger aint1 = **new** AtomicInteger(0);  
39.             MessageListener listenerA1 = **new** MessageListener()\{  
40.                 **public** **void** onMessage(Message message) \{  
41.                     **try** \{  
42.                         System.out.println(aint1.incrementAndGet()+" => A1 receive from kk.b: " + ((TextMessage)message).getText());  
43.                     \} **catch** (JMSException e) \{  
44.                         e.printStackTrace();  
45.                     \}  
46.                 \}\};  
47.             receiverA1.setMessageListener(listenerA1 );  
48.               
49.             // second receiver on broker1  
50.             QueueSession sessionA2 = connA.createQueueSession(**false**, Session.AUTO\_ACKNOWLEDGE);  
51.             QueueReceiver receiverA2 = sessionA2.createReceiver(queueB);  
52.             **final** AtomicInteger aint2 = **new** AtomicInteger(0);  
53.             MessageListener listenerA2 = **new** MessageListener()\{  
54.                 **public** **void** onMessage(Message message) \{  
55.                     **try** \{  
56.                         System.out.println(aint2.incrementAndGet()+" => A2 receive from kk.b: " + ((TextMessage)message).getText());  
57.                     \} **catch** (JMSException e) \{  
58.                         e.printStackTrace();  
59.                     \}  
60.                 \}\};  
61.             receiverA2.setMessageListener(listenerA2 );  
62.               
63.             // a fake one on broker1  
64.             QueueReceiver receiverA3 = sessionA2.createReceiver(queueB);  
65.             **final** AtomicInteger aint3 = **new** AtomicInteger(0);  
66.             MessageListener listenerA3 = **new** MessageListener()\{  
67.                 **public** **void** onMessage(Message message) \{  
68.                     **try** \{  
69.                         System.out.println(aint3.incrementAndGet()+" => A3 receive from kk.b: " + ((TextMessage)message).getText());  
70.                     \} **catch** (JMSException e) \{  
71.                         e.printStackTrace();  
72.                     \}  
73.                 \}\};  
74.             receiverA3.setMessageListener(listenerA3 );  
75.               
76.             QueueConnectionFactory factoryB = **new** ActiveMQConnectionFactory("tcp://127.0.0.1:61618");  
77.             Queue queueB1 = **new** ActiveMQQueue("kk.b");  
78.             QueueConnection connB = factoryB.createQueueConnection();  
79.             connB.start();  
80.               
81.             // one receiver on broker2  
82.             QueueSession sessionB1 = connB.createQueueSession(**false**, Session.AUTO\_ACKNOWLEDGE);  
83.             QueueReceiver receiverB1 = sessionB1.createReceiver(queueB);  
84.             **final** AtomicInteger bint1 = **new** AtomicInteger(0);  
85.             MessageListener listenerB1 = **new** MessageListener()\{  
86.                 **public** **void** onMessage(Message message) \{  
87.                     **try** \{  
88.                         System.out.println(bint1.incrementAndGet()+" => B1 receive from kk.b: " + ((TextMessage)message).getText());  
89.                     \} **catch** (JMSException e) \{  
90.                         e.printStackTrace();  
91.                     \}  
92.                 \}\};  
93.                 receiverB1.setMessageListener(listenerB1 );  
94.               
95.             // producer  on broker2  
96.             QueueSession sessionBp = connB.createQueueSession(**false**, Session.AUTO\_ACKNOWLEDGE);  
97.             MessageProducer producer = sessionBp.createProducer(queueB1);  
98.             **int** index = 0;  
99.             **while**(index++<300)\{  
100.                 TextMessage message = sessionBp.createTextMessage(index + " from kk.b on broker2");  
101.                 producer.send(message);  
102.             \}  
103.               
104.               
105.         \} **catch** (Exception e) \{  
106.             e.printStackTrace();  
107.             System.exit(1);  
108.         \}  
109.   
110.     \}  
111.   
112. \}

[BrokerTopology-1.png]: /images/20220612/13f637d68b3147c9bda32e8d9dfa3396.png
[Link 1]: http://activemq.apache.org/advisory-message.html
[1356515314_5094.jpg]: https://img-my.csdn.net/uploads/201212/26/1356515314_5094.jpg
[wildcard]: http://activemq.apache.org/wildcards.html
[http_tmielke.blogspot.de_2012_03_i-have-messages-on-queue-but-they-dont.html]: http://tmielke.blogspot.de/2012/03/i-have-messages-on-queue-but-they-dont.html
[1356515345_1663.jpg]: https://img-my.csdn.net/uploads/201212/26/1356515345_1663.jpg
[1356515352_3637.jpg]: https://img-my.csdn.net/uploads/201212/26/1356515352_3637.jpg
[Link 2]: http://lib.csdn.net/base/softwaretest
[view plain]: http://blog.csdn.net/kimmking/article/details/8440150/#

发表评论取消回复

表情：

评论列表（有 0 条评论，515人围观）

还没有评论，来说两句吧...

相关阅读

相关 redis集群模式详解

redis集群模式详解一，redis集群 1，哨兵集群 2，redis Cluster 二，redis Cluster集群的搭建

朴灿烈づ我的快乐病毒、/ 2024年04月07日 11:55/ 0 赞/ 68 阅读

相关 ActiveMQ集群模式

目录一、面试题二、多节点集群是什么三、zookeeper+replicated-leveldb-store的主从集群四、官网集群原理图五、部署规划和步骤

叁歲伎倆/ 2023年10月11日 20:18/ 0 赞/ 38 阅读

相关 Docker——Docker网络（Network）、Docker-Redis集群部署

Docker网络理解Docker0 常见的容器之间通信方式同一个宿主机内部容器网络信息查看

忘是亡心i/ 2023年01月16日 06:34/ 0 赞/ 277 阅读

相关 Redis详解（十）------ 集群模式详解

　　在上一篇博客我们介绍了------[Redis哨兵(Sentinel)模式][Redis_Sentinel],哨兵模式主要是解决高可用问题,在master节点宕机时,sla

「爱情、让人受尽委屈。」/ 2022年12月08日 04:51/ 0 赞/ 242 阅读

相关 docker network 网络模式

安装Docker时,自动创建3个网络模式(bridge,host,none),另外和可以自己指定一个(container): <table> <thead> <tr

水深无声/ 2022年11月09日 01:00/ 0 赞/ 172 阅读

相关 activemq-集群和主从模式学习笔记

activemq主从模式与集群三种masterslave模式 [三种主从模式][Link 1] <table> <thead> <tr> <t

忘是亡心i/ 2022年06月17日 05:19/ 0 赞/ 329 阅读

相关 ActiveMQ集群：网络连接模式(network connector)详解

网络连接模式(network connector) 针对海量消息所要求的横向扩展性和系统的高可用性，ActiveMQ提供了网络连接模式的集群功能。简单的说，就是通过把多个

悠悠/ 2022年06月12日 05:08/ 0 赞/ 516 阅读

相关 ActiveMQ集群

ActiveMQ集群 1.1 何为集群? 集群就是将相同的程序、功能，部署在两台或多台服务器上，这些服务器对外提供的功能是完全一样的。通过不断横向扩展增加服务器的

Love The Way You Lie/ 2022年05月31日 03:16/ 0 赞/ 244 阅读

相关 redis详解、哨兵模式、集群模式

redis 安装安装步骤： 1、下载获得redis-3.2.5.tar.gz后将它放入我们的Linux目录/opt 2、解压命令:tar -zxvf redi

今天药忘吃喽~/ 2022年03月18日 15:10/ 0 赞/ 273 阅读

相关 ActiveMQ静态网络连接

前言使用静态网络连接（static networkConnector）可以将多个broker连接起来，作为一个集群对外提供服务。我们假定有amq1和amq2配置了

淩亂°似流年/ 2021年10月14日 05:28/ 0 赞/ 406 阅读