一。 rabbitmq 叢集高可用介紹

   rabbitmq的broker是由一個或者多個物理節點組成 rabbtmq程式之間共享使用者 虛擬主機 佇列 交換機，繫結和引數，也可以將多個節點

組合成一個叢集，所有的資料都會被拷貝到叢集中不同的節點 除了佇列資料  佇列資料可以通過一些策略分配到不同的節點 具體參考高可用佇列

對映（http://www.rabbitmq.com/ha.html）

二。 rabbitmq叢集

 參考文件

   http://www.rabbitmq.com/clustering.html

   http://www.rabbitmq.com/ha.html

1》叢集安裝

模擬環境 

192.168.58.149 node2
192.168.58.150 node3
192.168.58.151 node4

rabbitmq叢集需要通過主機名互相訪問 所以必須在三臺機器 /etc/hosts配置以上內容 確保主機名正確 

三臺機器使用rabbitmq進行通訊 使用的安全策略是 erlang.cookie 就是一個隨機字串 三個必須保持一致 否則 無法新增節點到叢集
該cookie檔案 可能位於  /var/lib/rabbitmq/.erlang.cookie 或者 ~/.erlang.cookie任何一臺機器(比如node1) 啟動

 service rabbitmq-server start 
 

拷貝生成的cookie檔案到node2和node3

scp ~/.erlang.cookie [email protected]:~/
scp /var/lib/rabbitmq/.erlang.cookie [email protected]:/var/lib/rabbitmq/
scp ~/.erlang.cookie [email protected]:~/
scp /var/lib/rabbitmq/.erlang.cookie [email protected]:/var/lib/rabbitmq/

分別三臺機器安裝 rabbitmq-server 具體參考（http://blog.csdn.net/liaomin416100569/article/details/78476783）
分別啟動三臺機器 三臺機器都執行

rabbitmq-server -detached
 

分別檢視叢集狀態 每個機器單獨的沒有關聯起來 所有各自就是各自的叢集

node2上（running_nodes表示執行的叢集 只有自己）

[[email protected] rabbitmq]# rabbitmqctl cluster_status
Cluster status of node [email protected] ...
[{nodes,[{disc,[[email protected]]}]},
 {running_nodes,[[email protected]]},
 {cluster_name,<<"[email protected]">>},
 {partitions,[]}]
...done.

node3上

[[email protected] ~]# rabbitmqctl cluster_status
Cluster status of node [email protected] ...
[{nodes,[{disc,[[email protected]]}]},
 {running_nodes,[[email protected]]},
 {cluster_name,<<"[email protected]">>},
 {partitions,[]}]
...done.

node4上

[[email protected] ~]# rabbitmqctl cluster_status
Cluster status of node [email protected] ...
[{nodes,[{disc,[[email protected]]}]},
 {running_nodes,[[email protected]]},
 {cluster_name,<<"[email protected]">>},
 {partitions,[]}]
...done.

建立叢集 叢集必須有一箇中心節點 所有的其他節點會被新增到這個中心節點 比如我這裡是node2 叢集有個叢集名稱 如果三個節點被新增到一個叢集中 叢集名稱必須是相同 預設的名字是rabbit 比如node2 一般在叢集中的叢集名 就是 叢集名@機器名 node2就是[email protected]將node3和node4 停止應用 新增到node2叢集中 重啟服務

[[email protected] ~]# rabbitmqctl stop_app
Stopping node [email protected] ...
...done.
[[email protected] ~]# rabbitmqctl join_cluster [email protected]
Clustering node [email protected] with [email protected] ...
...done.
[[email protected] ~]# rabbitmqctl start_app
Starting node [email protected] ...
...done.

任何一臺機器檢視叢集狀態 發現三臺叢集都加入了執行叢集中

[[email protected] ~]# rabbitmqctl cluster_status
Cluster status of node [email protected] ...
[{nodes,[{disc,[[email protected],[email protected],[email protected]]}]},
 {running_nodes,[[email protected],[email protected],[email protected]]},
 {cluster_name,<<"[email protected]">>},
 {partitions,[]}]
...done.

2》叢集維護

 》》 叢集重置

   如果需要將某個節點從叢集中移除 可以在當前節點 使用reset命令

 比如將node3節點移除（移除後集群狀態執行節點就只有自己了）

[[email protected] ~]# rabbitmqctl stop_app
Stopping node [email protected] ...
...done.
[[email protected] ~]# rabbitmqctl reset
Resetting node [email protected] ...
...done.
[[email protected] ~]# rabbitmqctl cluster_status
Cluster status of node [email protected] ...
[{nodes,[{disc,[[email protected]]}]}]
...done.

如果某個中心節點被重置了 有可能一些其他節點無法重置可以使用

rabbitmqctl force_reset

rabbitmq支援遠端重置叢集 比如

假設 node2 網路中斷 （模擬 在node2上 執行 service network stop） 

node3執行

rabbitmqctl cluster_status

執行很長時間後 自動將叢集名稱 命名為  [email protected] 

》》其他維護命令 

  參考 （http://www.rabbitmq.com/cli.html  http://www.rabbitmq.com/man/rabbitmqctl.1.man.html）

三。 rabbitmq高可用

 rabbitmq的交換器等資料在所有叢集節點中共享比如在node2webgui上新增一個交換器test_ex

node3的webgui上檢視 發現多了test_ex交換器

試著新增使用者 bingding等 都會自動同步

但是當你試著去新增佇列時 發現新增介面有個選擇項 將佇列資訊新增到哪個節點上

新增後雖然你使用node3web檢視也能看到該佇列但是 該佇列的資料是被儲存在node2節點上

rabbitmq的高可用 也就是說 佇列以及佇列訊息資訊的高可用 防止一臺機器宕機後 佇列訊息的丟失

rabbitmq支援使用映象佇列（主從）模式將佇列通過某種策略匹配到不同的叢集節點上 交換器和繫結根據這些規則來獲取到對應的佇列

每個佇列都有主和從機（多個） 主負責讀寫 從用於備份

支援三種映象複製模式

ha-mode	ha-params	Result
all	(absent)	佇列對映到所有節點 新節點加入也會被對映
exactly	count	叢集中佇列的從機個數. 1就意味著沒從機 如果唯一節點掛了 意味著佇列無法訪問了， 數量 2意味 1主1從 如果主掛了 從自動接替稱為主，, 從機個數 = 配置引數- 1.， 如果配置引數超過了叢集總節點數 預設是all對映到所有節點，如果指定數量節點掛了 其他節點自動成為它的從節點.
nodes	node names	指定由哪些節點稱為 從節點 比如linux rabbitmqctl set_policy ha-nodes "^nodes\." \    '{"ha-mode":"nodes","ha-params":["[email protected]", "[email protected]"]}'

通過命令  rabbitmqctl set_policy 設定佇列複製的邏輯

該命令語法
rabbitmqctl set_policy ha-nodes "^nodes\." \
   '{"ha-mode":"nodes","ha-params":["[email protected]", "[email protected]"]}'

set_policy 策略名稱 匹配佇列名的正則表示式  被複制的方式（上面的ha-mode和引數）

以上三種模式舉例比如：

所有名字ha開頭的佇列都複製到所有的叢集主機

rabbitmqctl set_policy ha-all "^ha\." '{"ha-mode":"all"}'

所有名字 two.開頭的佇列隨機複製到任意一臺叢集主機 數量應該是2 還有一個主機

rabbitmqctl set_policy ha-two "^two\." \
   '{"ha-mode":"exactly","ha-params":2,"ha-sync-mode":"automatic"}'

所有nodes開頭的佇列都複製到node2和node3

rabbitmqctl set_policy ha-nodes "^nodes\." \
   '{"ha-mode":"nodes","ha-params":["[email protected]", "[email protected]"]}'

四。 rabbitmqapi測試叢集和高可用
使用高可用two.開頭的自動複製到兩個節點測試

執行

rabbitmqctl set_policy ha-two "^two\." \
   '{"ha-mode":"exactly","ha-params":2,"ha-sync-mode":"automatic"}'

使用簡單模式釋出一個訊息
Pub.java程式碼

package cn.et.p6;

import java.io.IOException;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

/**
 * 訊息傳送者 - 生產者
 * @author jiaozi
 *
 */
public class Pub {
	/**
	 * 佇列名稱 應該以two.開頭
	 */
	private final static String QUEUE_NAME = "two.hello";
	public static void main(String[] args) throws Exception {
		//連線遠端rabbit-server伺服器
		ConnectionFactory factory = new ConnectionFactory();
		factory.setHost("192.168.58.150");
		factory.setPort(5672);
		Connection connection = factory.newConnection();
		Channel channel = connection.createChannel();
		//定義建立一個佇列
		channel.queueDeclare(QUEUE_NAME, false, false, false, null);
		String message = "Hello World!";
		//傳送訊息
		channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
		System.out.println(" [x] Sent '" + message + "'");
		channel.close();
		connection.close();
	}

}

因為連線的 150（node3）所有 150是主機 剛好佇列名稱 是two.開頭 所以應該有個從機被複制 檢視佇列訊息

點選佇列名稱進入可以看到從節點 到底是哪臺

這裡 如果掛掉主節點node3  發現node2自動接管成為主節點

這裡訊息接受者 無論是從 node2,node3,node4都是可以消費的 

消費者測試

package cn.et.p6;

import java.io.IOException;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Consumer;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
import com.rabbitmq.client.QueueingConsumer;

/**
 * 訊息接受者 - 消費者
 * 
 * @author jiaozi
 *
 */
public class Rec {
	/**
	 * 獲取訊息佇列名稱
	 */
	private final static String QUEUE_NAME = "two.hello";
	/**
	 * 非同步接收
	 * @throws Exception 
	 */
	public static void asyncRec() throws Exception{
		ConnectionFactory factory = new ConnectionFactory();
		factory.setHost("192.168.58.149");
		Connection connection = factory.newConnection();
		Channel channel = connection.createChannel();
		//消費者也需要定義佇列 有可能消費者先於生產者啟動 
		channel.queueDeclare(QUEUE_NAME, false, false, false, null);
		System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
		//定義回撥抓取訊息
		Consumer consumer = new DefaultConsumer(channel) {
			@Override
			public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
					byte[] body) throws IOException {
				String message = new String(body, "UTF-8");
				System.out.println(" [x] Received '" + message + "'");
			}
		};
		channel.basicConsume(QUEUE_NAME, true, consumer);
	}
	public static void main(String[] args) throws Exception {
		asyncRec();
	}
}

如果 150掛掉 程式碼中需要重連其他的機器 所以需要定義叢集叢集中所有機器 這裡生產者和消費者連線程式碼相同 只貼出生產者

package cn.et.p6;

import java.io.IOException;

import com.rabbitmq.client.Address;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

/**
 * 訊息傳送者 - 生產者
 * @author jiaozi
 *
 */
public class Pub {
	static Address[] addresses=new Address[]{
			new Address("192.168.58.149"),
			new Address("192.168.58.150"),	
			new Address("192.168.58.151")	
		};
	/**
	 * 佇列名稱 應該以two.開頭
	 */
	private final static String QUEUE_NAME = "two.hello";
	public static void main(String[] args) throws Exception {
		//連線遠端rabbit-server伺服器
		ConnectionFactory factory = new ConnectionFactory();
		factory.setPort(5672);
		//預設自動恢復重連 http://www.rabbitmq.com/api-guide.html#recovery
		factory.setAutomaticRecoveryEnabled(true);
		//連線不上之後 多少毫秒重連
		factory.setNetworkRecoveryInterval(10000);
		//不會負載均衡只會抓取第一個可用的連線 
		Connection connection = factory.newConnection(addresses);
		Channel channel = connection.createChannel();
		//定義建立一個佇列
		channel.queueDeclare(QUEUE_NAME, false, false, false, null);
		String message = "Hello World!";
		//傳送訊息
		channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
		System.out.println(" [x] Sent '" + message + "'");
		channel.close();
		connection.close();
	}

}

預設只是抓取可以連線的第一個伺服器 預設可以自己實現一個客戶端負責均衡器  弄成rr輪詢（連不上就輪詢下一個 重寫該類 重寫newConnection方法）

這裡我就懶得去實現了  也可以使用第三方反向代理來負載均衡 比如nginx或者haproxy或者lvs+keepalived