下載 activemq 壓縮包解壓后，conf 目錄下有各種示例配置文件，紅線標(biāo)出的是靜態(tài)發(fā)現(xiàn)和動(dòng)態(tài)發(fā)現(xiàn)的配置。

?

1. 靜態(tài)配置

啟動(dòng)3個(gè) broker，端口分別為61616，61618，61620，配置如下：

<networkConnectors></networkConnectors> <transportConnectors><transportConnector name="openwire" uri="tcp://localhost:61616"/> </transportConnectors> <networkConnectors><networkConnector uri="static:(tcp://localhost:61616)" duplex="true"/> </networkConnectors> <transportConnectors><transportConnector name="openwire" uri="tcp://localhost:61618"/> </transportConnectors> <networkConnectors><networkConnector uri="static:(tcp://localhost:61616,tcp://localhost:61618)" duplex="true"/> </networkConnectors> <transportConnectors><transportConnector name="openwire" uri="tcp://localhost:61620"/> </transportConnectors>

3個(gè) broker 組成了一張網(wǎng)，當(dāng) producer 發(fā)送消息給 broker：61616 后，broker：61618 的消費(fèi)者可以收到該消息。消息從 broker：61616 流動(dòng)到 broker：61618，底層原理是 broker：61618 是 broker：61616 的一個(gè)消費(fèi)者。

2. 動(dòng)態(tài)配置

同樣地， 啟動(dòng)3個(gè) broker，端口分別為61616，61618，61620，配置如下：

<networkConnectors><networkConnector uri="multicast://default"/> </networkConnectors><transportConnector name="openwire" uri="tcp://0.0.0.0:61616" discoveryUri="multicast://default" /> </transportConnectors> <networkConnectors><networkConnector uri="multicast://default"/> </networkConnectors> <transportConnectors><transportConnector name="openwire" uri="tcp://0.0.0.0:61618" discoveryUri="multicast://default" /> </transportConnectors> <networkConnectors><networkConnector uri="multicast://default"/> </networkConnectors> <transportConnectors><transportConnector name="openwire" uri="tcp://0.0.0.0:61620" discoveryUri="multicast://default" /> </transportConnectors>

使用多播協(xié)議，把 3 個(gè) broker 動(dòng)態(tài)地組成了一張網(wǎng)。

// 省略其他代碼 public class MulticastDiscoveryAgent implements DiscoveryAgent, Runnable {public static final String DEFAULT_DISCOVERY_URI_STRING = "multicast://239.255.2.3:6155";public static final String DEFAULT_HOST_STR = "default"; public static final String DEFAULT_HOST_IP = System.getProperty("activemq.partition.discovery", "239.255.2.3"); public static final int DEFAULT_PORT = 6155; private static final Logger LOG = LoggerFactory.getLogger(MulticastDiscoveryAgent.class);private static final String TYPE_SUFFIX = "ActiveMQ-4.";private static final String ALIVE = "alive.";private static final String DEAD = "dead.";private static final String DELIMITER = "%";private static final int BUFF_SIZE = 8192;private static final int DEFAULT_IDLE_TIME = 500;private static final int HEARTBEAT_MISS_BEFORE_DEATH = 10;public void run() {byte[] buf = new byte[BUFF_SIZE];DatagramPacket packet = new DatagramPacket(buf, 0, buf.length);while (started.get()) {// 發(fā)送多播數(shù)據(jù) doTimeKeepingServices();try {// 接收多播數(shù)據(jù) mcast.receive(packet);if (packet.getLength() > 0) {String str = new String(packet.getData(), packet.getOffset(), packet.getLength());processData(str);}} catch (SocketTimeoutException se) {// ignore} catch (IOException e) {if (started.get()) {LOG.error("failed to process packet: " + e);}}}} }

?

3. broker 集群的原理

如上圖，61616和61618組成集群，當(dāng)61618加入一個(gè)consumer時(shí)，61618向61616發(fā)送一條ConsumerInfo消息，這樣61618就成為了61616的consumer。

ConsumerInfo 示例：

ConsumerInfo {commandId = 4, responseRequired = false, consumerId = dynamic-broker1->dynamic-broker2-1872-1524494145961-2:1:1:1, destination = queue://TEST.BAT, prefetchSize = 1, maximumPendingMessageLimit = 0, browser = false, dispatchAsync = true, selector = null, clientId = ID:USER-20140617MT-1882-1524494166015-0:1, subscriptionName = null, noLocal = false, exclusive = true, retroactive = false, priority = -5, brokerPath = [ID:USER-20140617MT-1877-1524494148767-0:1], optimizedAcknowledge = false, noRangeAcks = false, additionalPredicate = org.apache.activemq.command.NetworkBridgeFilter@413249b }

?

轉(zhuǎn)載于:https://www.cnblogs.com/allenwas3/p/8872822.html

總結(jié)

以上是生活随笔為你收集整理的ActiveMQ broker 集群, 静态发现和动态发现的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。