網絡連接模式

針對海量消息所要求的橫向擴展性和系統的高可用性，ActiveMQ提供了網絡連接模式的集群功能。簡單的說，就是通過把多個不同的broker實例連接在一起，作為一個整體對外提供服務，從而提高整體對外的消息服務能力。通過這種方式連接在一起的broker實例之間，可以共享隊列和消費者列表，從而達到分布式隊列的目的。

拓撲結構

幾種不同的ActiveMQ部署拓撲結構（嵌入、主從復制、網絡連接）：

配置示例

在activemq.xml的broker節點內添加：

<networkConnectors>

????? <networkConnectoruri=“static:(tcp://localhost:62001)”/>

</networkConnectors>

uri也可以使用其他兩種方式：

multicast://default 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的幾個屬性：

屬性	默認值	描述
initialReconnectDelay	1000	重連之前等待的時間(ms) (如果useExponentialBackOff為 false)
maxReconnectDelay	30000	重連之前等待的最大時間(ms)
useExponentialBackOff	true	每次重連失敗時是否增大等待時間
backOffMultiplier	2	增大等待時間的系數

networkConnector配置

配置參數

屬性	默認值	描述
name	bridge	名稱
dynamicOnly	false	如果為true, 持久訂閱被激活時才創建對應的網路持久訂閱。默認是啟動時激活。
decreaseNetworkConsumerPriority	false	如果為true，網絡的消費者優先級降低為-5。如果為false，則默認跟本地消費者一樣為0.
networkTTL	1	消息和訂閱在網絡上通過的broker數量
conduitSubscriptions	true	多個網絡消費者是否被當做一個消費者來對待。
excludedDestinations	empty	不通過網絡轉發的destination
dynamicallyIncludedDestinations	empty	通過網絡轉發的destinations，注意空列表代表所有的都轉發。
staticallyIncludedDestinations	empty	匹配的都將通過網絡轉發-即使沒有對應的消費者
duplex	false	如果為true，則既可消費又可生產消息到網絡broker
prefetchSize	1000	設置網絡消費者的prefetch size參數。必須大于0，因為網絡消費者不能自己輪詢消息。
suppressDuplicateQueueSubscriptions	false	(從5.3版本開始) 如果為true, 重復的訂閱關系一產生即被阻止。
bridgeTempDestinations	true	是否廣播advisory messages來創建臨時destination。
alwaysSyncSend	false	(從 5.6版本開始) 如果為true，非持久化消息也將使用request/reply方式代替oneway方式發送到遠程broker。
staticBridge	false	(從5.6版本開始) 如果為true，只有staticallyIncludedDestinations中配置的destination可以被處理。

networkConnector的實現原理是基于ActiveMQ的公告消息（AdvisoryMessage）機制的（參見此處）。當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，把消息投遞給實際的消費者。

管道訂閱(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>

此外，從5.6版本起，可以在networkConnector上設置destinationFilter來指定感興趣的Advisory Message將被傳播。

<networkConnector uri="static:(tcp://host)"destinationFilter="Queue.include.test.foo,ActiveMQ.Advisory.Consum er.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，為了防止消息回流后被當做重復消息而不被分發。

總結

以上是生活随笔為你收集整理的ActiveMQ的network connectors部署集群(七)的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。