ActiveMQ中Network的示例分析

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

一、在一臺服務器上啟動多個Broker

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

二、ActiveMQ的靜態網絡鏈接

• ActiveMQ的networkConnector是什么
  在某些場景下，需要多個ActiveMQ的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
  1：配置示例如下（activemq.xml--注釋掉persistenceFactory節點）：

  <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：默認為空，指定排除的地址，示例如下：
  
  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從broke2上消費掉，然后轉發給這些consumers。不幸的是轉發部分消息的時候broker1重啟了，這些consumers發現broker1連接失敗，通過failover連接到broker2上去了，但是又一部分他們還沒有消費的消息被broker2已經分發到broker1上去了。這些消息，就好像是消失了，除非有消費者重新連接到broker1上來消費。
     從5.6版起，在destinationPolicy上新增的選項replayWhenNoConsumers。這個選項使得broker1上有需要轉發的消息但是沒有消費者時，把消息回流到它原始的broker。同時把enableAudit設置為false，為了防止消息回流后被當做重復消息而不被分發，示例如下：

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

四、容錯的連接

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

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

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

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

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

五、動態網絡連接（純理論）
多播協議multicast
    ActiveMQ使用Multicast協議將一個Service和其他的Broker的Service連接起來。IP multicast是一個被用于網絡中傳輸數據到其他一組接收者的技術。
    IP multicast傳統的概念成為組地址。組地址是ip地址在224.0.0.0到239.255.255.255之間的ip地址。ActiveMQ broker使用multicast協議去建立服務與遠程的broker的服務的網絡連接。

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

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

• ActiveMQ使用multicast協議的配置格式如下：

  <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默認的ip，默認動態 的尋找地址
  2："discoveryUri"是指在transport中用multicast的default的地址傳遞
  3："uri"指動態尋找可利用的地址
  4：如果防止自動的尋找地址？
  （1）名稱為openwire的transport，移除discoveryUri="multicast:
  //default"即可。傳輸連接用默認的名稱openwire來配置broker的tcp多點連接，這將允許其他broker能夠自動發現和連接到可用的broker中。
  （2）名稱為"default-nc"的networkConnector，注釋掉或者刪除即可。
  ActiveMQ默認的networkConnector基于multicast協議的連接的默認名稱是default-nc，而且自動的去發現其他broker。去停止這種行為，只需要注銷或者刪除掉default-nc網絡連接。
  （3）使brokerName的名字唯一，可以唯一識別Broker的實例，默認是localhost

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

Discovery協議
    Discovery是在multicast協議的功能上定義的。功能類似與Failover功能。它將動態的發現multicast協議的broker的連接并且隨機的連接其中一個broker。

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

• Discovery協議的配置示例

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

Peer協議
    ActiveMQ提供了peer transport connector提供了更加容易的去嵌入broker網絡中。它創建一個優于vm連接的p2p網絡連接。默認格式如下：
peer://peergroup/brokerName?Key=value

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

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

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

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

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