ActiveMQ中Network的示例分析

這篇文章給大家分享的是有關(guān)ActiveMQ中Network的示例分析的內(nèi)容。小編覺得挺實(shí)用的，因此分享給大家做個參考，一起跟隨小編過來看看吧。

一、在一臺服務(wù)器上啟動多個Broker

• 步驟如下（為集群做準(zhǔn)備）：
  1：把整個conf文件夾復(fù)制一份，比如叫做conf2
  2：修改里面的activemq.xml文件
  （1）里面的brokerName不能跟原來的重復(fù)
  （2）數(shù)據(jù)存放的文件名稱不能重復(fù)，比如：
  <hahaDB directory=”${activemq.data}/kahadb_2”/>
  （3）所有設(shè)計的transportConnectors的端口，都要跟前面的不一樣
  3：修改jetty.xml，主要就是修改端口，比如：
  <property name=”port” value=”8181”/>端口必須和前面的不一樣
  4：到bin下面，復(fù)制一個activemq，比如叫做activemq2；
  （1）修改程序的id，不能和前面的重復(fù)
  ACTIVEMQ_PIDFILE=”$ACTIVEMQ_DATA/activemq2-hostname.pid”
  （2）修改配置文件路徑
  ACTIVEMQ_CONF=”$ACTIVEMQ_BASE/conf2”
  （3）修改端口，里面有個tcp的61616的端口，要改成不一樣的，最好跟    activemq.xml里面的ctp的端口一致
  （4）然后就可以執(zhí)行了。

二、ActiveMQ的靜態(tài)網(wǎng)絡(luò)鏈接

• ActiveMQ的networkConnector是什么
  在某些場景下，需要多個ActiveMQ的Broker做集群，那么久涉及到Broker的通信，這個被稱為ActiveMQ的networkConnector。
      ActiveMQ的networkConnector默認(rèn)是單向的，一個Broker在一端發(fā)送消息，另一個Broker在另一端接收消息。這就是所謂的“橋接”。ActiveMQ也支持雙向鏈接，創(chuàng)建一個雙向的通道對于兩個Broker，不僅發(fā)送消息而且也能從相同的通道來接收消息，通常作為duplex connector來映射，如下：
  

• “discovery”的概念
      一般情況下，discovery是被用來發(fā)現(xiàn)遠(yuǎn)程的服務(wù)，客戶端通常想去發(fā)現(xiàn)所有可利用的brokers；另一層意思，它是基于現(xiàn)有的網(wǎng)絡(luò)Broker去發(fā)現(xiàn)其他可用的Brokers。
      有兩種配置Client到Broker的鏈接方式，一種方式：Client通過Statically配置的方式去連接Broker，另一種方式：Client通過discovery agents來dynamically發(fā)現(xiàn)Brokers

• Static networks
      Static networkConnector是用于創(chuàng)建一個靜態(tài)的配置對于網(wǎng)絡(luò)中的多個Broker。這種協(xié)議用于復(fù)合url，一個復(fù)合url包括多個url地址。格式如下：
  static:(uri1,uri2,uri3,...)?Key=value
  1：配置示例如下（activemq.xml--注釋掉persistenceFactory節(jié)點(diǎn)）：

  <networkConnectors>
  <networkConnector name="local network" uri="static://(tcp://remotehost1:61616,tcp://remotehost2:61616)"/>
  </networkConnectors>

• Static networkConnector的基本原來示意圖：
  
      上圖中，兩個Brokers是通過一個static的協(xié)議來網(wǎng)絡(luò)鏈接的。一個Consumer鏈接到BrokerB的一個地址上，當(dāng)Producer在BrokerA上以相同的地址發(fā)送消息時，此時它將被轉(zhuǎn)移到BrokerB上。也就是，BrokerA會轉(zhuǎn)發(fā)消息到BrokerB上。

• networkConnector配置的可用屬性
  1、name：默認(rèn)是bridge
  2、dynamicOnly：默認(rèn)是false，如果為true，持久訂閱被激活時才創(chuàng)建對應(yīng)的網(wǎng)絡(luò)持久訂閱。默認(rèn)是啟動時激活
  3、decreaseNetworkConsumerPriority：默認(rèn)是false。設(shè)定消費(fèi)者優(yōu)先權(quán)，如果為true，網(wǎng)絡(luò)的消費(fèi)者優(yōu)先級降低為-5。如果為false，則默認(rèn)跟本地消費(fèi)者一樣為0
  4、networkTTL：默認(rèn)是1，網(wǎng)絡(luò)中用于消息和訂閱消費(fèi)的broker數(shù)量
  5、messageTTL：默認(rèn)是1，網(wǎng)絡(luò)中用于消息的broker數(shù)量
  6、consumerTTL：默認(rèn)是1，網(wǎng)絡(luò)中用于消費(fèi)的broker數(shù)量
  7、conduitSubscriptions：默認(rèn)true，是否把同一個broker的多個consumer當(dāng)做一個來處理
  8、dynamicallyIncludedDestinations：默認(rèn)為空，要包括的動態(tài)消息地址，類似于excludedDestinations,如

  <dynamicallyIncludedDestinations>
  <queue physicalName="include.test.foo"/>
  <topic physicalName="include.test.bar"/>
  </dynamicallyIncludedDestinations>

  9、staticallyIncludedDestinations：默認(rèn)為空，要包括的靜態(tài)消息地址。類似于excludedDestinations，如：

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

  10、excludedDestinations：默認(rèn)為空，指定排除的地址，示例如下：
  
  11、duplex：默認(rèn)false，設(shè)置是否能雙向通信
  12、prefetchSize：默認(rèn)是1000，持有的未確認(rèn)的最大消息數(shù)量，必須大于0，因?yàn)榫W(wǎng)絡(luò)消費(fèi)者不能自己輪詢消息
  13、suppressDuplicateQueueSubscriptions：默認(rèn)false，如果為true，重復(fù)的訂閱關(guān)系一產(chǎn)生即被阻止
  14、bridgeTempDestinations：默認(rèn)true，是否廣播advisory messages來創(chuàng)建臨時destination
  15、alwaysSyncSend：默認(rèn)false，如果為true，非持久化消息也將使用request/reply方式代替oneway方式發(fā)送到遠(yuǎn)程broker
  16、staticBridge：默認(rèn)false，如果為true，只有staticallyIncludedDestinations中配置的destination可以被處理

三、“丟失”的消息
     存在這樣的場景，broker1和broker2通過networkConnector連接，一些consumers連接到broker1，消費(fèi)broker2上的消息。消息先被broker1從broke2上消費(fèi)掉，然后轉(zhuǎn)發(fā)給這些consumers。不幸的是轉(zhuǎn)發(fā)部分消息的時候broker1重啟了，這些consumers發(fā)現(xiàn)broker1連接失敗，通過failover連接到broker2上去了，但是又一部分他們還沒有消費(fèi)的消息被broker2已經(jīng)分發(fā)到broker1上去了。這些消息，就好像是消失了，除非有消費(fèi)者重新連接到broker1上來消費(fèi)。
     從5.6版起，在destinationPolicy上新增的選項(xiàng)replayWhenNoConsumers。這個選項(xiàng)使得broker1上有需要轉(zhuǎn)發(fā)的消息但是沒有消費(fèi)者時，把消息回流到它原始的broker。同時把enableAudit設(shè)置為false，為了防止消息回流后被當(dāng)做重復(fù)消息而不被分發(fā)，示例如下：

<destinationPolicy>
    <policyMap>
        <policyEntry queue=">" enableAudit="false">
            <networkBridgeFilterFactory>
                <conditionalNetworkBridgeFilterFactory replayWhenNoConsumers="true"/>
            </networkBridgeFilterFactory>
        </policyEntry>
    </policyMap>
</destinationPolicy>

四、容錯的連接

• Failover Protocol
      之前的都是Client配置鏈接到指定的broker上。但是，如果Broker的鏈接失敗怎么辦？此時，Client有兩個選項(xiàng)：要么立刻死掉，要么去連接到其它的broker上。
      Failover協(xié)議實(shí)現(xiàn)了自動重新鏈接的邏輯。這里有兩種方式提供了穩(wěn)定的brokers列表對于Client鏈接。
  第一種：提供一個static的可用的Brokers列表
  第二種：提供一個dynamic發(fā)現(xiàn)的可用Brokers

• Failover Protocol的默認(rèn)方式
  failover:(uri1,...,uriN)?key=value或者failover:uri1,...,uriN
  & Failover Protocol的默認(rèn)配置
  默認(rèn)情況下，這種協(xié)議用于隨機(jī)的去選擇一個鏈接去鏈接，如果鏈接失敗了，那么會鏈接到其它的Broker上。默認(rèn)的配置定義了延遲重新鏈接，意味著傳輸將會在10秒后自動的去重新鏈接可用的broker。所有的重新鏈接參數(shù)都是可以根據(jù)應(yīng)用的需要而配置的。

• Failover Protocol的使用示例，在客戶端程序里面

  ConnectionFactory connectionFactory = new ActiveMQConnectionFactory   ("failover:(tcp://localhost:61616,tcp://localhost:61617)
  ?randomize=false");

• Failover Protocol可用的配置參數(shù)：
  1、initialReconnectDelay：第一次嘗試重連之前等待的時間（毫秒）默認(rèn)10
  2、maxReconnectDelay：最長重連的時間間隔（毫秒），默認(rèn)30000
  3、useExponentialBackOff：重連時間間隔是否以指數(shù)形式增長，默認(rèn)true
  4、backOffMultiplier：遞增倍數(shù)，默認(rèn)2.0
  5、maxReconnectAttempts：默認(rèn)-1|0，自版本5.6起，-1為默認(rèn)值，代表不限重連次數(shù)；0代表從不重試（只嘗試連接一次，并不重連），5.6版本之前，0為默認(rèn)值，代表不限重試次數(shù)；大于0的數(shù)，代表最大重試次數(shù)。
  6、startupMaxReconnectAttempts：初始化時的最大重連次數(shù)。一旦連接上，將使用maxReconnectAttempts的配置，默認(rèn)0
  7、Randomize：使用隨機(jī)連接，以達(dá)到負(fù)載均衡的目的，默認(rèn)true
  8、Backup：提前初始化一個未使用連接，以便進(jìn)行快速失敗轉(zhuǎn)移，默認(rèn)false
  9、timeout：設(shè)置發(fā)送操作的超時時間（毫秒），默認(rèn)-1
  10、trackMessages：設(shè)置是否緩存[故障發(fā)生時]尚未傳送完成的消息，當(dāng)broker一旦重新連接成功，便將這些緩存中的消息刷新到新連接的代理中，使得消息可以在broker切換前后順序傳送，默認(rèn)false
  11、maxCacheSize：當(dāng)trackMessages啟用時，緩存的最大字節(jié)，默認(rèn)為128*1024 字節(jié)
  12、updateURIsSupported：設(shè)定是否可以動態(tài)修改broker rui（自版本5.4起），默認(rèn)true

五、動態(tài)網(wǎng)絡(luò)連接（純理論）
多播協(xié)議multicast
    ActiveMQ使用Multicast協(xié)議將一個Service和其他的Broker的Service連接起來。IP multicast是一個被用于網(wǎng)絡(luò)中傳輸數(shù)據(jù)到其他一組接收者的技術(shù)。
    IP multicast傳統(tǒng)的概念成為組地址。組地址是ip地址在224.0.0.0到239.255.255.255之間的ip地址。ActiveMQ broker使用multicast協(xié)議去建立服務(wù)與遠(yuǎn)程的broker的服務(wù)的網(wǎng)絡(luò)連接。

• 基本的格式配置
  multicast://ip address:port?transportOptions
  1、group：表示唯一的組名稱，缺省值default
  2、minmumWireFormatVersion：被允許的最小的wireformat版本，缺省為0
  3、trace：是否追蹤記錄日志，默認(rèn)false
  4、useLocalHost：表示本地機(jī)器的名稱是否為localhost，默認(rèn)true
  5、datagramSize：特定的數(shù)據(jù)大小，默認(rèn)值41024
  6、timeToLive：消息的聲明周期值，默認(rèn)值-1
  7、loopBackMode：是否啟用loopback模式，默認(rèn)false
  8、wireFormat：默認(rèn)用wireFormat命名
  9、wireFormat.：前綴是wireFormat

• 配置示例
  1：默認(rèn)配置，默認(rèn)情況下是不可靠的多播，數(shù)據(jù)包可能會丟失
  multicast://default
  2：特定的ip和端口
  multicast://224.1.2.3:6255
  3：特定的ip和端口以及組名
  multicast://224.1.2.3:6255?group=mygroupname

• ActiveMQ使用multicast協(xié)議的配置格式如下：

  <broker xmlns="http://activemq.apache.org/schema/core" brokerName="multicast" dataDirectory="${activemq.data}/data">
  <networkConnectors>
      <networkConnector name="default-nc" uri="multicast://default"/>
  </networkConnectors>
  <transportConnectors>
      <transportConnector name="openwire" uri="tcp://localhost:61616" discoveryUri="multicast://default"/>
  </transportConnectors>
  </broker>

  說明：
  1：uri="multicast://default"中的default是activemq默認(rèn)的ip，默認(rèn)動態(tài) 的尋找地址
  2："discoveryUri"是指在transport中用multicast的default的地址傳遞
  3："uri"指動態(tài)尋找可利用的地址
  4：如果防止自動的尋找地址？
  （1）名稱為openwire的transport，移除discoveryUri="multicast:
  //default"即可。傳輸連接用默認(rèn)的名稱openwire來配置broker的tcp多點(diǎn)連接，這將允許其他broker能夠自動發(fā)現(xiàn)和連接到可用的broker中。
  （2）名稱為"default-nc"的networkConnector，注釋掉或者刪除即可。
  ActiveMQ默認(rèn)的networkConnector基于multicast協(xié)議的連接的默認(rèn)名稱是default-nc，而且自動的去發(fā)現(xiàn)其他broker。去停止這種行為，只需要注銷或者刪除掉default-nc網(wǎng)絡(luò)連接。
  （3）使brokerName的名字唯一，可以唯一識別Broker的實(shí)例，默認(rèn)是localhost

• multicast協(xié)議和普通的tcp協(xié)議
  它們是差不多的，不同的是multicast能夠自動的發(fā)現(xiàn)其他broker，從而替代了使用static功能列表brokers。用multicast協(xié)議可以在網(wǎng)絡(luò)中頻繁的添加和刪除ip不會有影響。
  multicast協(xié)議
  好處：能夠適應(yīng)動態(tài)變化的地址。
  缺點(diǎn)：自動連接地址會過渡的消耗網(wǎng)絡(luò)資源

Discovery協(xié)議
    Discovery是在multicast協(xié)議的功能上定義的。功能類似與Failover功能。它將動態(tài)的發(fā)現(xiàn)multicast協(xié)議的broker的連接并且隨機(jī)的連接其中一個broker。

• 基本配置如下：
  discovery:(discoveryAgentURI)?transportOptions
  1、reconnectDelay：再次尋址等待時間，缺省值10
  2、initialReconnectDelay：初始化設(shè)定再次尋址等待時間，缺省值10
  3、maxReconnectDelay：最大尋址等待時間，缺省值30000
  4、useExponentialBackOff：是否嘗試BackOff重連接，默認(rèn)是true
  5、backOffMultiplier：嘗試Backoff的次數(shù)，默認(rèn)是2
  6、maxReconnectAttempts：如果異常，最大的重新連接個數(shù)，默認(rèn)是0
  7、Group：組唯一的地址，默認(rèn)是default
  示例：
  discovery:(multicast://default)?initialReconnectDelay=100

• Discovery協(xié)議的配置示例

  <broker name="foo">
  <transportConnectors>
      <transportConnector uri="tcp://localhost:0" discoveryUri="multicast://default"/>
  </transportConnectors>
  </broker>

Peer協(xié)議
    ActiveMQ提供了peer transport connector提供了更加容易的去嵌入broker網(wǎng)絡(luò)中。它創(chuàng)建一個優(yōu)于vm連接的p2p網(wǎng)絡(luò)連接。默認(rèn)格式如下：
peer://peergroup/brokerName?Key=value

• Peer協(xié)議的基本使用
      當(dāng)我們啟用了peer協(xié)議時，應(yīng)用將自動的啟動內(nèi)嵌broker，也將會自動的去配置其它broker來建立連接，前提是必須要有一個組。配置如下：
  peer://groupa/broker1?Persistent=false
      另外，生產(chǎn)者和消費(fèi)者都各自連接到嵌入到自己應(yīng)用的broker，并且在本地的同一個組名中相互訪問數(shù)據(jù)。

• Peer協(xié)議的基本原理
  
  在本地機(jī)器斷網(wǎng)的情況下，本地的client訪問本地brokerA將依然正常。
  在斷網(wǎng)的情況下發(fā)送消息到本地brokerA，然后網(wǎng)路連接正常后，所有的消息將重新發(fā)送并連接到brokerB。
  Fanout協(xié)議

Fanout協(xié)議是同時連接到多個broker，默認(rèn)的格式如下：
    fanout:(fanoutURI)?key=value
示例：fanout:(static:(tcp://host1:61616,tcp://host2:61616))
表示client將試圖連接到兩個static列表中定義的三個URI

• Fanout協(xié)議的配置參數(shù)如下：
  1、initialReconnectDelay：重新連接的等待時間，默認(rèn)是10
  2、maxReconnectDelay：最大重新連接的等待時間，默認(rèn)是30000
  3、useExponentialBackOff：是否嘗試BackOff重連接，默認(rèn)是true
  4、backOffMultiplier：嘗試Backoff的次數(shù)，默認(rèn)是2
  5、maxReconnectAttempts：如果異常，最大的重新連接個數(shù)，默認(rèn)是0
  6、fanOutQueues：是否將topic消息轉(zhuǎn)換queue消息，默認(rèn)false
  7、minAckCount：Broker連接的最小數(shù)，默認(rèn)是2
  配置示例：
  fanout:(static:(tcp://localhost:61616,tcp://remotehost:61616))?
  initialReconnectDelay=100
  注意：
  ActiveMQ并不推薦Consumer使用fanout協(xié)議。當(dāng)Provider發(fā)送消息到多個broker中，測試Consumer可能受到重復(fù)的消息。

感謝各位的閱讀！關(guān)于“ActiveMQ中Network的示例分析”這篇文章就分享到這里了，希望以上內(nèi)容可以對大家有一定的幫助，讓大家可以學(xué)到更多知識，如果覺得文章不錯，可以把它分享出去讓更多的人看到吧！