Error: Package: rabbitmq-server-3.6.1-1.noarch (/rabbitmq-server-3.6.1-1.noarch) 
Requires: erlang >= R16B-3 
You could try using --skip-broken to work around the problem 
You could try running: rpm -Va --nofiles --nodigest

這是由于erlang的版本問題，其實是沒有影響的，你可以使用下面的命令進行安裝：

#rpm -ivh --nodeps rabbitmq-server-3.6.1-1.noarch.rpm

啟動

#service rabbitmq-server start --后臺方式運行

#service rabbitmq-server stop  --停止運行

#service rabbitmq-server status --查看狀態

#rabbitmq-server start 

可以看到使用的日志文件

日志目錄

/var/log/rabbitmq

#cat /var/log/rabbitmq/rabbit@iZ94nxslz66Z.log 可以看到下面的日志記錄

...................................................................................................................................................................................................................................................

卸載

管理

Rabbitmq服務器的主要通過rabbitmqctl和rabbimq-plugins兩個工具來管理，以下是一些常用功能。

1）. 服務器啟動與關閉

      啟動: rabbitmq-server –detached

      關閉:rabbitmqctl stop

      若單機有多個實例，則在rabbitmqctlh后加–n 指定名稱

2）. 插件管理

      開啟某個插件：rabbitmq-pluginsenable xxx

      關閉某個插件：rabbitmq-pluginsdisablexxx

      注意：重啟服務器后生效。

3）.virtual_host管理

      新建virtual_host: rabbitmqctladd_vhost  xxx

      撤銷virtual_host:rabbitmqctl  delete_vhost xxx

4）. 用戶管理

      新建用戶：rabbitmqctl add_user xxxpwd

      刪除用戶:   rabbitmqctl delete_user xxx

      改密碼: rabbimqctlchange_password {username} {newpassword}

      設置用戶角色：rabbitmqctlset_user_tags {username} {tag ...}

              Tag可以為 administrator,monitoring, management

5）. 權限管理

      權限設置：set_permissions [-pvhostpath] {user} {conf} {write} {read}

               Vhostpath

               Vhost路徑

               user

      用戶名

              Conf

      一個正則表達式match哪些配置資源能夠被該用戶訪問。

              Write

      一個正則表達式match哪些配置資源能夠被該用戶讀。

               Read

      一個正則表達式match哪些配置資源能夠被該用戶訪問。

6）. 獲取服務器狀態信息

       服務器狀態：rabbitmqctl status

       隊列信息：rabbitmqctl list_queues[-p vhostpath] [queueinfoitem ...]

                Queueinfoitem可以為：name，durable，auto_delete，arguments，messages_ready，

                messages_unacknowledged，messages，consumers，memory

       Exchange信息：rabbitmqctllist_exchanges[-p vhostpath] [exchangeinfoitem ...]

                 Exchangeinfoitem有：name，type，durable，auto_delete，internal，arguments.

       Binding信息：rabbitmqctllist_bindings[-p vhostpath] [bindinginfoitem ...]       

                 Bindinginfoitem有：source_name，source_kind，destination_name，destination_kind，routing_key，arguments

       Connection信息：rabbitmqctllist_connections [connectioninfoitem ...]

       Connectioninfoitem有：recv_oct，recv_cnt，send_oct，send_cnt，send_pend等。

       Channel信息：rabbitmqctl  list_channels[channelinfoitem ...]

      Channelinfoitem有consumer_count，messages_unacknowledged，messages_uncommitted，acks_uncommitted，messages_unconfirmed，prefetch_count，client_flow_blocked

常用命令：

查看所有隊列信息

# rabbitmqctl list_queues

關閉應用

# rabbitmqctl stop_app

啟動應用，和上述關閉命令配合使用，達到清空隊列的目的

# rabbitmqctl start_app

清除所有隊列

# rabbitmqctl reset

更多用法及參數，可以執行如下命令查看

# rabbitmqctl

rabbitmq常用命令

rabbitmq-server start  或者   service rabbitmq-server start     #啟動rabbitmq

rabbitmqctl list_exchanges 

rabbitmqctl list_bindings

rabbitmqctl list_queues #分別查看當前系統種存在的Exchange和Exchange上綁定的Queue信息。

rabbitmqctl status  #查看運行信息

rabbitmqctl stop     #停止運行rabbitmq

rabbitmq-plugins enable rabbitmq_management  

#啟動rabbitmq的圖形管理界面，這個操作必須重啟rabbitmq， 然后在web中 http://127.0.0.1:15672 用戶名和密碼都是guest guest。如果局域網無法訪問設置防火墻過濾規則或關閉防火墻。

什么是復制?

RabbitMQ broker操作所需的所有數據/狀態都可以在多個節點間復制. 例外是消息隊列,默認情況下它駐留在一個節點， 盡管它們對所有節點來說，是可見的，可達的.要在集群中跨節點復制隊列，可參考high availability 文檔(注意，你仍然先需要一個工作集群).

主機名解析需求

RabbitMQ節點彼此之間使用域名，要么是簡短的，要么是全限定的(FQDNs). 因此，集群中所有成員的主機名都必須是可解析的，也可用于機器上的命令行工具，如rabbitmqctl.

主機名解析可使用任何一種標準的操作系統提供方法:

• DNS 記錄
• 本地主機文件(e.g. /etc/hosts)

要使用FQDNs, 參考RABBITMQ_USE_LONGNAME in the Configuration guide.

集群構成

集群可以通過多種方式來構建:

• 手動使用rabbitmqctl
• 通過在配置文件列舉集群節點來聲明
• 通過使用rabbitmq-autocluster (插件)來聲明
• 使用rabbitmq-clusterer（插件）來聲明

一個集群的構成可以動態修改. 所有RabbitMQ brokers開始都是以單個節點來運行的. 這些節點可以加入到集群中, 隨后也可以脫離集群再次成為單一節點。

故障處理

RabbitMQ brokers 可以容忍個別節點故障. 節點可以隨意地啟動和關閉，只要在已知關閉的時間內能夠聯系到集群節點.

RabbitMQ 集群有多種模式來處理網絡分化, 主要是一致性方向. 集群是在LAN中使用的，不推薦在WAN中運行集群. Shovel 或 Federation 插件對于跨WAN連接brokers ，有更好的解決方案. 注意 Shovel 和 Federation 不等同于集群.

磁盤和內存節點

節點可以是磁盤節點，也可以是內存節點。多數情況下，你希望所有的節點都是磁盤節點，但RAM節點是一種特殊情況，它可以提高集群中隊列和，交換機，綁定的性能. 當有疑問時，最好只使用磁盤節點。

節點(以及CLI工具)之間如何來認證: Erlang Cookie

RabbitMQ 節點和CLI 工具(如rabbitmqctl) 使用cookie來確定每個節點之間是否可以通信. 兩個節點之間要能通信，它們必須要有相同的共享密鑰Erlang cookie. cookie只是具有字母數字特征的字符串。只要你喜歡，它可長可短. 每個集群節點必須有相同的cookie.

當RabbitMQ 服務器啟動時，Erlang VM 會自動地創建一個隨機的cookie文件. 最簡單的處理方式是允許一個節點來創建文件，然后再將這個文件拷貝到集群的其它節點中。

在 Unix 系統中, cookie的通常位于/var/lib/rabbitmq/.erlang.cookie 或$HOME/.erlang.cookie.

在Windows中, 其位置在C:\Users\Current User\.erlang.cookie(%HOMEDRIVE% + %HOMEPATH%\.erlang.cookie) 或C:\Documents and Settings\Current User\.erlang.cookie, 對于RabbitMQ Windows service其位置在C:\Windows\.erlang.cookie。如果使用了Windows service ,  cookie可被放于這兩個位置中.

作為替代方案，你可以在 rabbitmq-server 和 rabbitmqctl 腳本中調用erl時,插入"-setcookie cookie"選項.

當cookie未配置時 (例如，不相同), RabbitMQ 會記錄這樣的錯誤"Connection attempt from disallowed node" and "Could not auto-cluster".

啟動獨立節點

集群可通過重新配置，而將現有RabbitMQ 節點加入到集群配置中. 因此第一步是以正常的方式在所有節點上啟動RabbitMQ:

rabbit1$ rabbitmq-server -detached 
rabbit2$ rabbitmq-server -detached 
rabbit3$ rabbitmq-server -detached

這會創建三個獨立的RabbitMQ brokers, 每個節點一個，可通過cluster_status命令來驗證:

rabbit1$ rabbitmqctl cluster_status 
Cluster status of node rabbit@rabbit1 ... [{nodes,[{disc,[rabbit@rabbit1]}]},{running_nodes,[rabbit@rabbit1]}] ...done. 
rabbit2$ rabbitmqctl cluster_status 
Cluster status of node rabbit@rabbit2 ... [{nodes,[{disc,[rabbit@rabbit2]}]},{running_nodes,[rabbit@rabbit2]}] ...done. 
rabbit3$ rabbitmqctl cluster_status 
Cluster status of node rabbit@rabbit3 ... [{nodes,[{disc,[rabbit@rabbit3]}]},{running_nodes,[rabbit@rabbit3]}] ...done.

以rabbitmq-server shell腳本來啟動RabbitMQ broker的節點名稱是rabbit@shorthostname,在這里，短節點名稱是小寫的(如上面的rabbit@rabbit1). 如果在windows上，你使用rabbitmq-server.bat批處理文件來啟動，短節點名稱是大寫的（如：rabbit@RABBIT1). 當你輸入節點名稱時，不論是大寫還是小寫的，這些字符串都必須精確匹配。

創建集群

為了把這三個節點構建到一個集群中，我們可以告訴其中的兩個節點, 假設為rabbit@rabbit2 和 rabbit@rabbit3, 將加入到第三個節點的集群中，這第三個節點假設為rabbit@rabbit1.

首先我們將rabbit@rabbit2加入到rabbit@rabbit1的集群中. 要做到這一點，我們必須在rabbit@rabbit2 上停止RabbitMQ應用程序，并將其加入到rabbit@rabbit1 集群中, 然后再重啟RabbitMQ 應用程序. 

注意：加入集群會隱式地重置節點, 因此這會刪除此節點上先前存在的所有資源和數據.（如何備份數據）

rabbit2$ rabbitmqctl stop_app 
Stopping node rabbit@rabbit2 ...done. 
rabbit2$ rabbitmqctl join_cluster rabbit@rabbit1 
Clustering node rabbit@rabbit2 with [rabbit@rabbit1] ...done. 
rabbit2$ rabbitmqctl start_app 
Starting node rabbit@rabbit2 ...done.

在每個節點上通過運行cluster_status 命令，我們可以看到兩個節點已經加入了集群:

rabbit1$ rabbitmqctl cluster_status 
Cluster status of node rabbit@rabbit1 ... [{nodes,[{disc,[rabbit@rabbit1,rabbit@rabbit2]}]},  {running_nodes,[rabbit@rabbit2,rabbit@rabbit1]}] ...done. 
rabbit2$ rabbitmqctl cluster_status 
Cluster status of node rabbit@rabbit2 ... [{nodes,[{disc,[rabbit@rabbit1,rabbit@rabbit2]}]},  {running_nodes,[rabbit@rabbit1,rabbit@rabbit2]}] ...done. 

現在我們將rabbit@rabbit3節點加入到同一個集群中. 操作步驟同上面的一致,除了這次我們選擇rabbit2來加入集群，但這并不重要：

rabbit3$ rabbitmqctl stop_app 
Stopping node rabbit@rabbit3 ...done. 
rabbit3$ rabbitmqctl join_cluster rabbit@rabbit2 
Clustering node rabbit@rabbit3 with rabbit@rabbit2 ...done. 
rabbit3$ rabbitmqctl start_app 
Starting node rabbit@rabbit3 ...done.

在任何一個節點上通過運行cluster_status命令，我們可以看到三個節點已經加入了集群:

rabbit1$ rabbitmqctl cluster_status 
Cluster status of node rabbit@rabbit1 ... [{nodes,[{disc,[rabbit@rabbit1,rabbit@rabbit2,rabbit@rabbit3]}]},  {running_nodes,[rabbit@rabbit3,rabbit@rabbit2,rabbit@rabbit1]}] ...done. 
rabbit2$ rabbitmqctl cluster_status 
Cluster status of node rabbit@rabbit2 ... [{nodes,[{disc,[rabbit@rabbit1,rabbit@rabbit2,rabbit@rabbit3]}]},  {running_nodes,[rabbit@rabbit3,rabbit@rabbit1,rabbit@rabbit2]}] ...done. 
rabbit3$ rabbitmqctl cluster_status 
Cluster status of node rabbit@rabbit3 ... [{nodes,[{disc,[rabbit@rabbit3,rabbit@rabbit2,rabbit@rabbit1]}]},  {running_nodes,[rabbit@rabbit2,rabbit@rabbit1,rabbit@rabbit3]}] ...done.

通過上面的步驟，當集群運行的時候，我們可以在任何時候將新的節點加入到集群中.

重啟集群節點

注意，加入到集群中的節點可在任何時候停止， 對于崩潰來說也沒有問題. 在這兩種情況下，集群剩余的節點將不受影響地繼續操作，當它們重啟的時候，這些崩潰的節點會再次自動追趕上其它的集群節點。

我們關閉了節點rabbit@rabbit1和rabbit@rabbit3，并在每步觀察集群的狀態:

rabbit1$ rabbitmqctl stop 
Stopping and halting node rabbit@rabbit1 ...done. 
rabbit2$ rabbitmqctl cluster_status 
Cluster status of node rabbit@rabbit2 ... [{nodes,[{disc,[rabbit@rabbit1,rabbit@rabbit2,rabbit@rabbit3]}]},  {running_nodes,[rabbit@rabbit3,rabbit@rabbit2]}] ...done. 
rabbit3$ rabbitmqctl cluster_status 
Cluster status of node rabbit@rabbit3 ... [{nodes,[{disc,[rabbit@rabbit1,rabbit@rabbit2,rabbit@rabbit3]}]},  {running_nodes,[rabbit@rabbit2,rabbit@rabbit3]}] ...done. 
rabbit3$ rabbitmqctl stop 
Stopping and halting node rabbit@rabbit3 ...done. 
rabbit2$ rabbitmqctl cluster_status 
Cluster status of node rabbit@rabbit2 ... [{nodes,[{disc,[rabbit@rabbit1,rabbit@rabbit2,rabbit@rabbit3]}]},  {running_nodes,[rabbit@rabbit2]}] ...done.

譯者注：關閉了rabbit1節點后，運行的節點已經沒有rabbit1節點了

現在我們再次啟動節點,并檢查集群狀態:

rabbit1$ rabbitmq-server -detached 
rabbit1$ rabbitmqctl cluster_status 
Cluster status of node rabbit@rabbit1 ... [{nodes,[{disc,[rabbit@rabbit1,rabbit@rabbit2,rabbit@rabbit3]}]},  {running_nodes,[rabbit@rabbit2,rabbit@rabbit1]}] ...done. 
rabbit2$ rabbitmqctl cluster_status 
Cluster status of node rabbit@rabbit2 ... [{nodes,[{disc,[rabbit@rabbit1,rabbit@rabbit2,rabbit@rabbit3]}]},  {running_nodes,[rabbit@rabbit1,rabbit@rabbit2]}] ...done. 
rabbit3$ rabbitmq-server -detached 
rabbit1$ rabbitmqctl cluster_status 
Cluster status of node rabbit@rabbit1 ... [{nodes,[{disc,[rabbit@rabbit1,rabbit@rabbit2,rabbit@rabbit3]}]},  {running_nodes,[rabbit@rabbit2,rabbit@rabbit1,rabbit@rabbit3]}] ...done. 
rabbit2$ rabbitmqctl cluster_status 
Cluster status of node rabbit@rabbit2 ... [{nodes,[{disc,[rabbit@rabbit1,rabbit@rabbit2,rabbit@rabbit3]}]},  {running_nodes,[rabbit@rabbit1,rabbit@rabbit2,rabbit@rabbit3]}] ...done. 
rabbit3$ rabbitmqctl cluster_status 
Cluster status of node rabbit@rabbit3 ... [{nodes,[{disc,[rabbit@rabbit1,rabbit@rabbit2,rabbit@rabbit3]}]},  {running_nodes,[rabbit@rabbit2,rabbit@rabbit1,rabbit@rabbit3]}] ...done.

這里有一些重要的警告:

• 當整個集群崩潰的時候， 最后一個崩潰的節點必須第一個上線.如果不是這樣，節點將會等待最后一個磁盤節點30秒以確認其重新上線，否則就會失敗. 如果最后一個下線的節點，不能再重新上線，那么它可能會使用forget_cluster_node命令來從集群中刪除 - 查閱 rabbitmqctl頁面來了解更多信息.
• 如果所有集群節點都在同一個時間內停止且不受控制（如斷電）。在這種情況下，你可以在某個節點上使用force_boot命令使其再次成為可啟動的－查閱 rabbitmqctl頁面來了解更多信息.

脫離集群

當節點不再是集群的一部分時，可以明確地將其從集群中刪除. 首先我們將節點rabbit@rabbit3從集群中刪除, 以使其回歸獨立操作.要做到這一點，需要在rabbit@rabbit3節點上停止RabbitMQ 應用程序,重設節點，并重啟RabbitMQ應用程序.

rabbit3$ rabbitmqctl stop_app 
Stopping node rabbit@rabbit3 ...done. 
rabbit3$ rabbitmqctl reset 
Resetting node rabbit@rabbit3 ...done. 
rabbit3$ rabbitmqctl start_app 
Starting node rabbit@rabbit3 ...done.

在節點上運行cluster_status 命令來確認rabbit@rabbit3節點現在已不再是集群的一部分，并且會獨自操作:

rabbit1$ rabbitmqctl cluster_status 
Cluster status of node rabbit@rabbit1 ... [{nodes,[{disc,[rabbit@rabbit1,rabbit@rabbit2]}]},  {running_nodes,[rabbit@rabbit2,rabbit@rabbit1]}] ...done. 
rabbit2$ rabbitmqctl cluster_status 
Cluster status of node rabbit@rabbit2 ... [{nodes,[{disc,[rabbit@rabbit1,rabbit@rabbit2]}]},  {running_nodes,[rabbit@rabbit1,rabbit@rabbit2]}] ...done. 
rabbit3$ rabbitmqctl cluster_status 
Cluster status of node rabbit@rabbit3 ... [{nodes,[{disc,[rabbit@rabbit3]}]},{running_nodes,[rabbit@rabbit3]}] ...done. 

我們也可以遠程地刪除節點，這是相當有用的，舉例來說，當處理無反應的節點時.舉例來說，我們可以從 rabbit@rabbit2中刪除rabbit@rabbi1.

rabbit1$ rabbitmqctl stop_app 
Stopping node rabbit@rabbit1 ...done. 
rabbit2$ rabbitmqctl forget_cluster_node rabbit@rabbit1 
Removing node rabbit@rabbit1 from cluster ... ...done.

注意，rabbit1仍然認為它與rabbit2處在一個集群中，但嘗試啟動時會出現一個錯誤.這時，我們需要對其進行重置以使其能再次啟動.

rabbit1$ rabbitmqctl start_app 
Starting node rabbit@rabbit1 ... Error: inconsistent_cluster: Node rabbit@rabbit1 thinks it's clustered with node rabbit@rabbit2, but rabbit@rabbit2 disagrees 
rabbit1$ rabbitmqctl reset 
Resetting node rabbit@rabbit1 ...done. 
rabbit1$ rabbitmqctl start_app Starting node rabbit@mcnulty ... ...done.

現在， cluster_status 命令會顯示三個節點都是獨立節點，并且操作是獨立的:

rabbit1$ rabbitmqctl cluster_status 
Cluster status of node rabbit@rabbit1 ... [{nodes,[{disc,[rabbit@rabbit1]}]},{running_nodes,[rabbit@rabbit1]}] ...done. 
rabbit2$ rabbitmqctl cluster_status 
Cluster status of node rabbit@rabbit2 ... [{nodes,[{disc,[rabbit@rabbit2]}]},{running_nodes,[rabbit@rabbit2]}] ...done. 
rabbit3$ rabbitmqctl cluster_status 
Cluster status of node rabbit@rabbit3 ... [{nodes,[{disc,[rabbit@rabbit3]}]},{running_nodes,[rabbit@rabbit3]}] ...done.

注意：rabbit@rabbit2節點仍然殘留有集群的狀態（譯者注：怎么看出來的呢？）, 但是 rabbit@rabbit1 和rabbit@rabbit3 節點是新鮮的RabbitMQ brokers.如果我們想重新初始化rabbit@rabbit2節點，我們可以按其它節點的步驟來操作:

rabbit2$ rabbitmqctl stop_app 
Stopping node rabbit@rabbit2 ...done. 
rabbit2$ rabbitmqctl reset 
Resetting node rabbit@rabbit2 ...done. 
rabbit2$ rabbitmqctl start_app 
Starting node rabbit@rabbit2 ...done.

創建內存節點

當節點加入集群時，我們可將其聲明為內存節點. 我們可以通過使用像先前rabbitmqctl join_cluster命令再加--ram標志來達到目的:

rabbit2$ rabbitmqctl stop_app 
Stopping node rabbit@rabbit2 ...done. 
rabbit2$ rabbitmqctl join_cluster --ram rabbit@rabbit1 
Clustering node rabbit@rabbit2 with [rabbit@rabbit1] ...done. 
rabbit2$ rabbitmqctl start_app Starting node rabbit@rabbit2 ...done.

rabbit1$ rabbitmqctl cluster_status 
Cluster status of node rabbit@rabbit1 ... [{nodes,[{disc,[rabbit@rabbit1]},{ram,[rabbit@rabbit2]}]},  {running_nodes,[rabbit@rabbit2,rabbit@rabbit1]}] ...done. 
rabbit2$ rabbitmqctl cluster_status 
Cluster status of node rabbit@rabbit2 ... [{nodes,[{disc,[rabbit@rabbit1]},{ram,[rabbit@rabbit2]}]},  {running_nodes,[rabbit@rabbit1,rabbit@rabbit2]}] ...done. 

改變節點類型

我們可以將節點的類型從磁盤修改為內存，反之亦然. 假設我們想反轉rabbit@rabbit2 和 rabbit@rabbit1的節點類型,即先將內存節點轉換為磁盤節點，隨后再將其從磁盤節點轉換為內存節點.要做到這點，我們可以使用change_cluster_node_type命令. 首先節點必須先停止.

rabbit2$ rabbitmqctl stop_app 
Stopping node rabbit@rabbit2 ...done. rabbit2$ 
rabbitmqctl change_cluster_node_type disc 
Turning rabbit@rabbit2 into a disc node ... ...done. Starting node rabbit@rabbit2 ...done. 
rabbit1$ rabbitmqctl stop_app 
Stopping node rabbit@rabbit1 ...done. 
rabbit1$ rabbitmqctl change_cluster_node_type ram 
Turning rabbit@rabbit1 into a ram node ... 
rabbit1$ rabbitmqctl start_app 
Starting node rabbit@rabbit1 ...done.

1. Producer.java
2. Consumer.java
3. GetOne.java

1. 綁定到一個臨時隊列，即調用無參的channel.queueDeclare()方法
2. 綁定到一個非自動刪除的命名隊列
3. 綁定到一個非自動刪除的命名隊列，并且指定x-message-ttl ,如步驟2中展示的一樣.

1. Producer.java
2. Consumer.java

1. 使用arguments.put("x-message-ttl", 10000)來設置消息TTL ,正如如何使指定隊列上消息過期食譜中描述的一樣.
2. 使用arguments.put("x-dead-letter-exchange", exchange_dead_letter)來設置死信交換器名稱;

1. reason : 表示隊列是否過期的或駁回的(requeue =false )
2. queue : 表示隊列源，例如stat_queue_02/05
3. time : 表示消息變為死信的日期和時間
4. exchange : 表示交換器，如monitor_exchange_02/05
5. routing-keys : 表示發送消息時原先使用的路由鍵

1. ref_exchange_c1_6 (廣播消息) 與exchange綁定.
2. ref_exchange_c1_8 (使用topic來處理消息路由)與exchange綁定 .

1. Producer.java
2. Consumer.java
3. StatsConsumer.java
4. CCStatsConsumer.java
5. BCCStatsConsumer.java

AMQP client 關閉概述

AMQP 0-9-1 connection和channel 使用相同的方法來管理網絡故障,內部故障,以及顯示本地關閉.

AMQP 0-9-1 connection  和 channel 有如下的生命周期狀態:

• open: 準備要使用的對象
• closing: 對象已顯示收到收到本地關閉通知, 并向任何支持的底層對象發出關閉請求,并等待其關閉程序完成
• closed: 對象已收到所有底層對象的完成關閉通知,最終將執行關閉操作

這些對象總是以closed狀態結束的,不管基于什么原因引發的關閉,比如:應用程序請求,內部client library故障, 遠程網絡請求或網絡故障.

AMQP connection 和channel 對象會持有下面與關閉相關的方法:

• addShutdownListener(ShutdownListener 監聽器)和removeShutdownListener(ShutdownListener 監聽器),用來管理監聽器,當對象轉為closed狀態時,將會觸發這些監聽器.注意,在已經關閉的對象上添加一個ShutdownListener將會立即觸發監聽器
• getCloseReason(), 允許同其交互以了解對象關閉的理由
• isOpen(), 用于測試對象是否處于open狀態
• close(int closeCode, String closeMessage), 用于顯示通知對象關閉

可以像這樣來簡單使用監聽器:

import com.rabbitmq.client.ShutdownSignalException; 
import com.rabbitmq.client.ShutdownListener;  
connection.addShutdownListener(new ShutdownListener() {     
public void shutdownCompleted(ShutdownSignalException cause)     {         ...     } }
);

關閉環境信息

可通過顯示調用getCloseReason()方法或通過使用ShutdownListener類中的業務方法的cause參數來從ShutdownSignalException中獲取關閉原因的有用信息.

ShutdownSignalException 類提供方法來分析關閉的原因.通過調用isHardError()方法,我們可以知道是connection錯誤還是channel錯誤.getReason()會返回相關cause的相關信息,這些引起cause的方法形式-要么是AMQP.Channel.Close方法,要么是AMQP.Connection.Close (或者是null,如果是library中引發的異常,如網絡通信故障,在這種情況下,可通過getCause()方法來獲取信息).

public void shutdownCompleted(ShutdownSignalException cause) {   if (cause.isHardError())   {     
Connection conn = (Connection)cause.getReference();     
if (!cause.isInitiatedByApplication())     {       
Method reason = cause.getReason();       ...     }     ...   } 
else {     Channel ch = (Channel)cause.getReference();     ...   } }

原子使用isOpen()方法

channel和connection對象的isOpen()方法不建議在在生產代碼中使用,因為此方法的返回值依賴于shutdown cause的存在性. 下面的代碼演示了竟爭條件的可能性:

public void brokenMethod(Channel channel) {     if (channel.isOpen())     {         // The following code depends on the channel being in open state.         // However there is a possibility of the change in the channel state         // between isOpen() and basicQos(1) call         ...         channel.basicQos(1);     } }

相反,我們應該忽略這種檢查,并簡單地嘗試這種操作.如果代碼執行期間,connection的channel關閉了,那么將拋出ShutdownSignalException,這就表明對象處于一種無效狀態了.當broker意外關閉連接時,我們也應該捕獲由SocketException引發的IOException,或者當broker清理關閉時,捕獲ShutdownSignalException.

public void validMethod(Channel channel) {     try {         ...         channel.basicQos(1);     } catch (ShutdownSignalException sse) {         // possibly check if channel was closed         // by the time we started action and reasons for         // closing it         ...     } catch (IOException ioe) {         // check why connection was closed         ...     } }

Consumer線程池

Consumer 線程默認是通過一個新的ExecutorService線程池來自動分配的(參考下面的Receiving).如果需要在newConnection() 方法中更好地控制ExecutorService,可以使用定制的線程池.下面的示例展示了一個比正常分配稍大的線程池:

ExecutorService es = Executors.newFixedThreadPool(20); Connection conn = factory.newConnection(es); 

當連接關閉時,默認的ExecutorService將會被shutdown(), 但用戶自定義的ExecutorService (如上面所示)將不會被shutdown(). 提供自定義ExecutorService的Clients必須確保最終它能被關閉(通過調用它的shutdown() 方法), 否則池中的線程可能會阻止JVM終止.

同一個executor service,可在多個連接之間共享,或者連續地在重新連接上重用,但在shutdown()后,則不能再使用.

使用這種特性時,唯一需要考慮的是:在消費者回調的處理過程中,是否有證據證明有嚴重的瓶頸. 如果沒有消費者執行回調,或很少,默認的配置是綽綽有余. 開銷最初是很小的,分配的全部線程資源也是有界限的,即使偶爾可能出現一陣消費活動.

使用Host列表

可以傳遞一個Address數組給newConnection(). Address只是 com.rabbitmq.client 包中包含host和port組件的簡單便利類. 例如:

Address[] addrArr = new Address[]{ new Address(hostname1, portnumber1)                                  , new Address(hostname2, portnumber2)}; Connection conn = factory.newConnection(addrArr); 

如果提供了ExecutorService(在factory.newConnection(es, addrArr)中使用),那么線程池將與第一個成功連接相關聯.

心跳超時

參考Heartbeats guide 來了解更多關于心跳及其在Java client中如何配置的更多信息.

自定義線程工廠

像Google App Engine (GAE)這樣的環境會限制線程直接實例化. 在這樣的環境中使用RabbitMQ Java client, 需要配置一個定制的ThreadFactory,即使用合適的方法來實例化線程,如: GAE's ThreadManager. 下面是Google App Engine的相關代碼.

import com.google.appengine.api.ThreadManager;  ConnectionFactory cf = new ConnectionFactory(); cf.setThreadFactory(ThreadManager.backgroundThreadFactory()); 

Connection恢復

clients和RabbitMQ節點之間的連接可發生故障. RabbitMQ Java client 支持連接和拓撲(queues, exchanges, bindings, 和consumers)的自動恢復. 大多數應用程序的連接自動恢復過程會遵循下面的步驟:

1. 重新連接
2. 恢復連接監聽器
3. 重新打開通道
4. 恢復通道監聽器
5. 恢復通道basic.qos 設置,發布者確認和事務設置

1. 重新聲明交換器(except for predefined ones)
2. 重新聲明隊列
3. 恢復所有綁定
4. 恢復所有消費者

ConnectionFactory factory = new ConnectionFactory(); factory.setUsername(userName); factory.setPassword(password); factory.setVirtualHost(virtualHost); factory.setHost(hostName); factory.setPort(portNumber); factory.setAutomaticRecoveryEnabled(true); // connection that will recover automatically Connection conn = factory.newConnection();

ConnectionFactory factory = new ConnectionFactory(); // attempt recovery every 10 seconds factory.setNetworkRecoveryInterval(10000);

ConnectionFactory factory = new ConnectionFactory();  Address[] addresses = {new Address("192.168.1.4"), new Address("192.168.1.5")}; factory.newConnection(addresses);

恢復監聽器

可在可恢復連接和通道上注冊一個或多個恢復監聽器. 當啟用了連接恢復時,ConnectionFactory#newConnection 和 Connection#createChannel 的連接已實現了com.rabbitmq.client.Recoverable,并提供了兩個方法:

• addRecoveryListener
• removeRecoveryListener

發布影響

當連接失敗時,使用Channel.basicPublish方法發送的消息將會丟失. client不會保證在連接恢復后,消息會得到分發.要確保發布的消息到達了RabbitMQ,應用程序必須使用Publisher Confirms 

拓撲恢復

拓撲恢復涉及交換器,隊列,綁定以及消費者恢復.默認是啟用的,但也可以禁用:

ConnectionFactory factory = new ConnectionFactory();  Connection conn = factory.newConnection(); factory.setAutomaticRecoveryEnabled(true); factory.setTopologyRecoveryEnabled(false);

手動應答和自動恢復

當使用手動應答時,在消息分發與應答之間可能存在網絡連接中斷. 在連接恢復后,RabbitMQ會在所有通道上重設delivery標記. 也就是說,使用舊delivery標記的basic.ack, basic.nack, 以及basic.reject將會引發channel exception. 為了避免發生這種情況, RabbitMQ Java client可以跟蹤,更新,以使它們在恢復期間單調地增長. Channel.basicAck, Channel.basicNack, 以及Channel.basicReject 然后可以轉換這些 delivery tags,并且不再發送過期的delivery tags. 使用手動應答和自動恢復的應用程序必須負責處理重新分發.