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的版本問題，其實是沒有影響的，你可以使用下面的命令進(jìn)行安裝：

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

啟動

#service rabbitmq-server start --后臺方式運(yùn)行

#service rabbitmq-server stop  --停止運(yùn)行

#service rabbitmq-server status --查看狀態(tài)

#rabbitmq-server start 

可以看到使用的日志文件

日志目錄

/var/log/rabbitmq

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

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

卸載

管理

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

1）. 服務(wù)器啟動與關(guān)閉

      啟動: rabbitmq-server –detached

      關(guān)閉:rabbitmqctl stop

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

2）. 插件管理

      開啟某個插件：rabbitmq-pluginsenable xxx

      關(guān)閉某個插件：rabbitmq-pluginsdisablexxx

      注意：重啟服務(wù)器后生效。

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}

      設(shè)置用戶角色：rabbitmqctlset_user_tags {username} {tag ...}

              Tag可以為 administrator,monitoring, management

5）. 權(quán)限管理

      權(quán)限設(shè)置：set_permissions [-pvhostpath] {user} {conf} {write} {read}

               Vhostpath

               Vhost路徑

               user

      用戶名

              Conf

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

              Write

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

               Read

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

6）. 獲取服務(wù)器狀態(tài)信息

       服務(wù)器狀態(tài)：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

關(guān)閉應(yīng)用

# rabbitmqctl stop_app

啟動應(yīng)用，和上述關(guān)閉命令配合使用，達(dá)到清空隊列的目的

# rabbitmqctl start_app

清除所有隊列

# rabbitmqctl reset

更多用法及參數(shù)，可以執(zhí)行如下命令查看

# rabbitmqctl

rabbitmq常用命令

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

rabbitmqctl list_exchanges 

rabbitmqctl list_bindings

rabbitmqctl list_queues #分別查看當(dāng)前系統(tǒng)種存在的Exchange和Exchange上綁定的Queue信息。

rabbitmqctl status  #查看運(yùn)行信息

rabbitmqctl stop     #停止運(yùn)行rabbitmq

rabbitmq-plugins enable rabbitmq_management  

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

什么是復(fù)制?

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

主機(jī)名解析需求

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

主機(jī)名解析可使用任何一種標(biāo)準(zhǔn)的操作系統(tǒng)提供方法:

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

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

集群構(gòu)成

集群可以通過多種方式來構(gòu)建:

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

一個集群的構(gòu)成可以動態(tài)修改. 所有RabbitMQ brokers開始都是以單個節(jié)點(diǎn)來運(yùn)行的. 這些節(jié)點(diǎn)可以加入到集群中, 隨后也可以脫離集群再次成為單一節(jié)點(diǎn)。

故障處理

RabbitMQ brokers 可以容忍個別節(jié)點(diǎn)故障. 節(jié)點(diǎn)可以隨意地啟動和關(guān)閉，只要在已知關(guān)閉的時間內(nèi)能夠聯(lián)系到集群節(jié)點(diǎn).

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

磁盤和內(nèi)存節(jié)點(diǎn)

節(jié)點(diǎn)可以是磁盤節(jié)點(diǎn)，也可以是內(nèi)存節(jié)點(diǎn)。多數(shù)情況下，你希望所有的節(jié)點(diǎn)都是磁盤節(jié)點(diǎn)，但RAM節(jié)點(diǎn)是一種特殊情況，它可以提高集群中隊列和，交換機(jī)，綁定的性能. 當(dāng)有疑問時，最好只使用磁盤節(jié)點(diǎn)。

節(jié)點(diǎn)(以及CLI工具)之間如何來認(rèn)證: Erlang Cookie

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

當(dāng)RabbitMQ 服務(wù)器啟動時，Erlang VM 會自動地創(chuàng)建一個隨機(jī)的cookie文件. 最簡單的處理方式是允許一個節(jié)點(diǎn)來創(chuàng)建文件，然后再將這個文件拷貝到集群的其它節(jié)點(diǎn)中。

在 Unix 系統(tǒng)中, 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 腳本中調(diào)用erl時,插入"-setcookie cookie"選項.

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

啟動獨(dú)立節(jié)點(diǎn)

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

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

這會創(chuàng)建三個獨(dú)立的RabbitMQ brokers, 每個節(jié)點(diǎn)一個，可通過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的節(jié)點(diǎn)名稱是rabbit@shorthostname,在這里，短節(jié)點(diǎn)名稱是小寫的(如上面的rabbit@rabbit1). 如果在windows上，你使用rabbitmq-server.bat批處理文件來啟動，短節(jié)點(diǎn)名稱是大寫的（如：rabbit@RABBIT1). 當(dāng)你輸入節(jié)點(diǎn)名稱時，不論是大寫還是小寫的，這些字符串都必須精確匹配。

創(chuàng)建集群

為了把這三個節(jié)點(diǎn)構(gòu)建到一個集群中，我們可以告訴其中的兩個節(jié)點(diǎn), 假設(shè)為rabbit@rabbit2 和 rabbit@rabbit3, 將加入到第三個節(jié)點(diǎn)的集群中，這第三個節(jié)點(diǎn)假設(shè)為rabbit@rabbit1.

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

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

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.

在每個節(jié)點(diǎn)上通過運(yùn)行cluster_status 命令，我們可以看到兩個節(jié)點(diǎn)已經(jīng)加入了集群:

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. 

現(xiàn)在我們將rabbit@rabbit3節(jié)點(diǎn)加入到同一個集群中. 操作步驟同上面的一致,除了這次我們選擇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.

在任何一個節(jié)點(diǎn)上通過運(yùn)行cluster_status命令，我們可以看到三個節(jié)點(diǎn)已經(jīng)加入了集群:

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.

通過上面的步驟，當(dāng)集群運(yùn)行的時候，我們可以在任何時候?qū)⑿碌墓?jié)點(diǎn)加入到集群中.

重啟集群節(jié)點(diǎn)

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

我們關(guān)閉了節(jié)點(diǎn)rabbit@rabbit1和rabbit@rabbit3，并在每步觀察集群的狀態(tài):

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.

譯者注：關(guān)閉了rabbit1節(jié)點(diǎn)后，運(yùn)行的節(jié)點(diǎn)已經(jīng)沒有rabbit1節(jié)點(diǎn)了

現(xiàn)在我們再次啟動節(jié)點(diǎn),并檢查集群狀態(tài):

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.

這里有一些重要的警告:

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

脫離集群

當(dāng)節(jié)點(diǎn)不再是集群的一部分時，可以明確地將其從集群中刪除. 首先我們將節(jié)點(diǎn)rabbit@rabbit3從集群中刪除, 以使其回歸獨(dú)立操作.要做到這一點(diǎn)，需要在rabbit@rabbit3節(jié)點(diǎn)上停止RabbitMQ 應(yīng)用程序,重設(shè)節(jié)點(diǎn)，并重啟RabbitMQ應(yīng)用程序.

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.

在節(jié)點(diǎn)上運(yùn)行cluster_status 命令來確認(rèn)rabbit@rabbit3節(jié)點(diǎn)現(xiàn)在已不再是集群的一部分，并且會獨(dú)自操作:

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. 

我們也可以遠(yuǎn)程地刪除節(jié)點(diǎn)，這是相當(dāng)有用的，舉例來說，當(dāng)處理無反應(yīng)的節(jié)點(diǎn)時.舉例來說，我們可以從 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仍然認(rèn)為它與rabbit2處在一個集群中，但嘗試啟動時會出現(xiàn)一個錯誤.這時，我們需要對其進(jìn)行重置以使其能再次啟動.

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.

現(xiàn)在， cluster_status 命令會顯示三個節(jié)點(diǎn)都是獨(dú)立節(jié)點(diǎn)，并且操作是獨(dú)立的:

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節(jié)點(diǎn)仍然殘留有集群的狀態(tài)（譯者注：怎么看出來的呢？）, 但是 rabbit@rabbit1 和rabbit@rabbit3 節(jié)點(diǎn)是新鮮的RabbitMQ brokers.如果我們想重新初始化rabbit@rabbit2節(jié)點(diǎn)，我們可以按其它節(jié)點(diǎn)的步驟來操作:

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.

創(chuàng)建內(nèi)存節(jié)點(diǎn)

當(dāng)節(jié)點(diǎn)加入集群時，我們可將其聲明為內(nèi)存節(jié)點(diǎn). 我們可以通過使用像先前rabbitmqctl join_cluster命令再加--ram標(biāo)志來達(dá)到目的:

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. 

改變節(jié)點(diǎn)類型

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

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

1. Producer.java
2. Consumer.java

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

1. reason : 表示隊列是否過期的或駁回的(requeue =false )
2. queue : 表示隊列源，例如stat_queue_02/05
3. time : 表示消息變?yōu)樗佬诺娜掌诤蜁r間
4. exchange : 表示交換器，如monitor_exchange_02/05
5. routing-keys : 表示發(fā)送消息時原先使用的路由鍵

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 關(guān)閉概述

AMQP 0-9-1 connection和channel 使用相同的方法來管理網(wǎng)絡(luò)故障,內(nèi)部故障,以及顯示本地關(guān)閉.

AMQP 0-9-1 connection  和 channel 有如下的生命周期狀態(tài):

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

這些對象總是以closed狀態(tài)結(jié)束的,不管基于什么原因引發(fā)的關(guān)閉,比如:應(yīng)用程序請求,內(nèi)部client library故障, 遠(yuǎn)程網(wǎng)絡(luò)請求或網(wǎng)絡(luò)故障.

AMQP connection 和channel 對象會持有下面與關(guān)閉相關(guān)的方法:

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

可以像這樣來簡單使用監(jiān)聽器:

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

關(guān)閉環(huán)境信息

可通過顯示調(diào)用getCloseReason()方法或通過使用ShutdownListener類中的業(yè)務(wù)方法的cause參數(shù)來從ShutdownSignalException中獲取關(guān)閉原因的有用信息.

ShutdownSignalException 類提供方法來分析關(guān)閉的原因.通過調(diào)用isHardError()方法,我們可以知道是connection錯誤還是channel錯誤.getReason()會返回相關(guān)cause的相關(guān)信息,這些引起cause的方法形式-要么是AMQP.Channel.Close方法,要么是AMQP.Connection.Close (或者是null,如果是library中引發(fā)的異常,如網(wǎng)絡(luò)通信故障,在這種情況下,可通過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()方法不建議在在生產(chǎn)代碼中使用,因為此方法的返回值依賴于shutdown cause的存在性. 下面的代碼演示了竟?fàn)帡l件的可能性:

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);     } }

相反,我們應(yīng)該忽略這種檢查,并簡單地嘗試這種操作.如果代碼執(zhí)行期間,connection的channel關(guān)閉了,那么將拋出ShutdownSignalException,這就表明對象處于一種無效狀態(tài)了.當(dāng)broker意外關(guān)閉連接時,我們也應(yīng)該捕獲由SocketException引發(fā)的IOException,或者當(dāng)broker清理關(guān)閉時,捕獲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 線程默認(rèn)是通過一個新的ExecutorService線程池來自動分配的(參考下面的Receiving).如果需要在newConnection() 方法中更好地控制ExecutorService,可以使用定制的線程池.下面的示例展示了一個比正常分配稍大的線程池:

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

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

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

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

使用Host列表

可以傳遞一個Address數(shù)組給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)中使用),那么線程池將與第一個成功連接相關(guān)聯(lián).

心跳超時

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

自定義線程工廠

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

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

Connection恢復(fù)

clients和RabbitMQ節(jié)點(diǎn)之間的連接可發(fā)生故障. RabbitMQ Java client 支持連接和拓?fù)?queues, exchanges, bindings, 和consumers)的自動恢復(fù). 大多數(shù)應(yīng)用程序的連接自動恢復(fù)過程會遵循下面的步驟:

1. 重新連接
2. 恢復(fù)連接監(jiān)聽器
3. 重新打開通道
4. 恢復(fù)通道監(jiān)聽器
5. 恢復(fù)通道basic.qos 設(shè)置,發(fā)布者確認(rèn)和事務(wù)設(shè)置

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

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);

恢復(fù)監(jiān)聽器

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

• addRecoveryListener
• removeRecoveryListener

發(fā)布影響

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

拓?fù)浠謴?fù)

拓?fù)浠謴?fù)涉及交換器,隊列,綁定以及消費(fèi)者恢復(fù).默認(rèn)是啟用的,但也可以禁用:

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

手動應(yīng)答和自動恢復(fù)

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