ActiveMQ的静态网络链接
ActiveMQ的networkConnector是什么?
在某些场景下,需要多个ActiveMQ的Broker做集群,那么就涉及到Broker到Broker的通信,这个被称为ActiveMQ的networkConnector。
ActiveMQ的networkConnector默认是单向的,一个Broker在一端发送消息,另一Broker在另一端接收消息。这就是所谓的“桥接”。 ActiveMQ也支持双向链接,创建一个双向的通道对于两个Broker,不仅发送消息而且也能从相同的通道来接收消息,通常作为duplex connector来映射,如下:
“discovery”的概念
一般情况下,discovery是被用来发现远程的服务,客户端通常想去发现所有可利用的brokers;另一层意思,它是基于现有的网络Broker去发现其他可用的Brokers。
有两种配置Client到Broker的链接方式,一种方式:Client通过Statically配置的方式去连接Broker,一种方式:Client通过discovery agents来dynamically的发现Brokers。
Static networks
Static networkConnector是用于创建一个静态的配置对于网络中的多个Broker。这种协议用于复合url,一个复合url包括多个url地址。格式如下:
static:(uri1,uri2,uri3,...)?key=value
配置示例如下:
<networkConnectors>
<networkConnector name="local network"
uri="static://(tcp://remotehost1:61616,tcp://remotehost2:61616)"/>
</networkConnectors>
Static networkConnector的基本原理示意图:
上图中,两个Brokers是通过一个static的协议来网络链接的。一个Consumer链接到brokerB的一个地址上 ,当Producer在brokerA上以相同的地址发送消息时,此时它将被转移到brokerB上。也就是,BrokerA会转发消息到BrokerB上。
networkConnector配置的可用属性:
1:name:默认是bridge
2:dynamicOnly:默认是false,如果为true, 持久订阅被激活时才创建对应的网路持久订阅。默认是启动时激活
3:decreaseNetworkConsumerPriority:默认是false。设定消费者优先权,如果为true,网络的消费者优先级降低
为-5。如果为false,则默认跟本地消费者一样为0
4:networkTTL :默认是1 ,网络中用于消息和订阅消费的broker数量
5:messageTTL :默认是1 ,网络中用于消息的broker数量
6:consumerTTL:默认是1 ,网络中用于消费的broker数量
7:conduitSubscriptions :默认true,是否把同一个broker的多个consumer当做一个来处理
8:dynamicallyIncludedDestinations :默认为空,要包括的动态消息地址,类似于excludedDestinations,如:
<dynamicallyIncludedDestinations>
<queue physicalName="include.test.foo"/>
<topic physicalName="include.test.bar"/>
</dynamicallyIncludedDestinations>
9:staticallyIncludedDestinations :默认为空,要包括的静态消息地址。类似于excludedDestinations,如:
<staticallyIncludedDestinations>
<queue physicalName="always.include.queue"/>
</staticallyIncludedDestinations>
10:excludedDestinations :默认为空,指定排除的地址,示例如下:
<networkConnectors>
<networkConnector uri="static://(tcp://localhost:61617)"
name="bridge" dynamicOnly="false" conduitSubscriptions="true"
decreaseNetworkConsumerPriority="false">
<excludedDestinations>
<queue physicalName="exclude.test.foo"/>
<topic physicalName="exclude.test.bar"/>
</excludedDestinations>
<dynamicallyIncludedDestinations>
<queue physicalName="include.test.foo"/>
<topic physicalName="include.test.bar"/>
</dynamicallyIncludedDestinations>
<staticallyIncludedDestinations>
<queue physicalName="always.include.queue"/>
<topic physicalName="always.include.topic"/>
</staticallyIncludedDestinations>
</networkConnector>
</networkConnectors>
11:duplex :默认false,设置是否能双向通信。
12:prefetchSize :默认是1000,持有的未确认的最大消息数量,必须大于0,因为网络消费者不能自己轮询消息。
13:suppressDuplicateQueueSubscriptions:默认false,如果为true, 重复的订阅关系一产生即被阻止。
14:bridgeTempDestinations :默认true,是否广播advisory messages来创建临时destination。
15:alwaysSyncSend :默认false,如果为true,非持久化消息也将使用request/reply方式代替oneway方式发送到远程broker。
16:staticBridge :默认false,如果为true,只有staticallyIncludedDestinations中配置的destination可以被处理。
“丢失”的消息
有这样的场景,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=">" enableAudit="false">
<networkBridgeFilterFactory>
<conditionalNetworkBridgeFilterFactory replayWhenNoConsumers="true"/>
</networkBridgeFilterFactory>
</policyEntry>
</policyEntries>
</policyMap>
</destinationPolicy>