Springboot 1.5.x 整合基於Centos7的RabbitMQ叢集安裝及配置
RabbitMQ簡介
RabbitMQ是實現了高階訊息佇列協議（AMQP）的開源訊息代理軟體（亦稱面向訊息的中介軟體）。
RabbitMQ是一套開源（MPL）的訊息佇列服務軟體，是由LShift提供的一個Advanced Message Queuing Protocol（AMQP）的開源實現，由以高效能、健壯以及可伸縮性出名的Erlang寫成。
選擇RabbitMQ
市面上有很多MQ可以選擇，如：ActiveMQ、ZeroMQ、Apache Qpid及RocketMQ，為什麼要選擇RabbitMQ呢？
1. 除了Qpid，RabbitMQ是唯一一個實現了AMQP標準的訊息伺服器；
2. 可靠性，RabbitMQ的持久化支援，保證了訊息的穩定性；
3. 高併發，RabbitMQ使用了Erlang開發語言，Erlang是為電話交換機開發的語言，天生自帶高併發光環和高可用特性；
4. 叢集部署簡單，正是應為Erlang使得RabbitMQ叢集部署變的超級簡單；
5. 社群活躍度高，從網上資料來看，RabbitMQ也是首選
工作機制
1. 訊息模型
生產者、消費者和代理
生產者（producer）：訊息的建立者，負責建立和推送資料到訊息伺服器；
消費者（consumer）：訊息的接收方，用於處理資料和確認訊息；
代理（proxy）：就是RabbitMQ本身，用於扮演“快遞”的角色，本身不生產訊息，只是扮演“快遞”的角色。
圖片來源： 訊息佇列之 RabbitMQ
2. 基本概念
RabbitMQ內部結構：
1. Message
訊息是不具名的，它由訊息頭和訊息體組成。訊息體是不透明的，而訊息頭則由一系列的可選屬性組成，這些屬性包括routing-key（路由鍵）、priority（相對於其他訊息的優先權）、delivery-mode（指出該訊息可能需要永續性儲存）等。
2. Publisher
訊息得生產者，也是一個向交換器釋出訊息的客戶端應用程式。
3. Exchange
交換器，用來接收生產者傳送的訊息並將這些訊息路由給伺服器中的佇列。
4. Binding
繫結，用於訊息佇列和交換器之間的關聯，一個繫結就是基於路由鍵將交換器和訊息佇列連線起來的路由規則，所以可以將交換器理解成一個由繫結構成的路由表。
5. Queue
訊息佇列，用來儲存訊息知道傳送給消費者。他是訊息的容器，也是訊息的終點。一個訊息可投入一個或多個佇列。訊息一直在佇列裡面，等待消費者連線到這個佇列將其取走。
6. Connection
網路連線，比如一個TCP連線
7. Channel
通道，多路複用連線中的一條獨立的雙向資料流通道。通道是建立在真實的TCP連線內地虛擬連線，AMQP命令都是通過通道發出去的，不管是釋出訊息、訂閱佇列還是接收訊息，這些動作都是通過通道完成。因為對於作業系統來說建立和銷燬TCP都是非常昂貴的開銷，所以引入了通道的概念，以複用一條TCP連線。
8. Consumer
訊息的消費者，表示一個從訊息佇列中取得訊息的客戶端應用程式。
9. Virtual Host
虛擬主機，表示一批交換器、訊息佇列和相關物件。虛擬主機是共享相同的身份認證和加密環境的獨立伺服器域。每個vhost本質上就是一個mini版的RabbitMQ伺服器，擁有自己的佇列、交換器、繫結和許可權機制。vhost是AMQP概念的基礎，必須在連線時指定，RabbitMQ預設的vhost是 / 。
10. Broker
表示訊息佇列伺服器實體。
3. AMQP中的訊息路由
AMQP中訊息的路由過程和Java開發者熟悉的JMS存在一些差異，AMQP中增加了Exchange和Binding的角色。生產者把訊息釋出到Exchange上，訊息最終達到佇列並被消費者接收，而Binding決定交換器的訊息應該傳送到哪個佇列。
4. 交換機Exchange
Exchange分發訊息時根據型別的不同分發策略有區別，目前存在四種類型：direct、fanout、topic、headers。headers匹配AMQP訊息的header而不是路由主鍵，此外headers交換機和direct交換機完全一致，但效能差很多，目前基本上用不到了，所以直接看其他型別。
1. direct
訊息中的路由主鍵（routing key）如何和Binding中的binding key一致，交換機就將訊息發到對應的佇列中。路由鍵和佇列名完全匹配，如果一個佇列繫結到交換機要求路由鍵為“dog”，則只能轉發routing key標記為“dog”的訊息。它是完全匹配，單播的模式。
2. fanout
每個發到fanout型別交換器的訊息都會分到所有繫結的佇列上去。fanout交換機不處理路由鍵，只是簡單的將佇列繫結到交換機上，每個傳送到交換機的訊息都會被轉發到與該交換機繫結的所有佇列上。很像子網廣播，每臺子網內的主機都獲得了一份複製的訊息。它是轉發訊息最快的。
3. topic
topic交換機通過模式匹配分配訊息的路由鍵屬性，將路由鍵和某個模式進行匹配，此時佇列需要繫結到一個模式上。它將路由鍵和繫結鍵的字串切分成單詞，這些單詞之間用點隔開。它同樣也會識別兩個萬用字元：符號“#”和符號“”。#匹配0個活多個單詞，“”匹配不多不少一個單詞。
5. 訊息持久化
RabbitMQ佇列和交換器在預設情況下，重啟伺服器會導致訊息丟失，如何保證RabbitMQ在重啟的時候訊息不丟失呢？答案：訊息持久化
RabbitMQ持久化會將你的持久化訊息寫入到磁碟中的持久化日誌檔案中，等訊息被消費後，RabbitMQ會把這條訊息標識為等待垃圾回收。
當你把訊息傳送到RabbitMQ伺服器的時候，你需要選擇是否要進行持久化，但這並不能保證RabbitMQ能從崩潰中恢復，想要訊息恢復必須滿足以下3個條件：
1. 投遞訊息的時候durable設定為true，訊息持久化，程式碼：channel.queueDeclare(x, true, false, false, null)，引數2設定為true持久化；
2. 設定投遞模式deliveryMode設定為2（持久），程式碼：channel.basicPublish(x, x, MessageProperties.PERSISTENT_TEXT_PLAIN,x)，引數3設定為儲存純文字到磁碟；
3. 訊息已經到達持久化交換器上；
4. 訊息已經到達持久化的佇列；
4. 持久化缺點
訊息持久化的優點顯而易見，但缺點也很明顯，那就是效能，因為要寫入硬碟要比寫入記憶體效能較低很多，從而降低了伺服器的吞吐量，儘管使用SSD硬碟可以使事情得到緩解，但他仍然吸乾了Rabbit的效能，當訊息成千上萬條要寫入磁碟的時候，效能是很低的。所以需要根據自己專案的實際情況，選擇適合自己的方式。
5. 虛擬主機
每個Rabbit都能建立很多vhost，我們稱之為虛擬主機，每個虛擬主機其實都是mini版的RabbitMQ，擁有自己的佇列，交換機和繫結，擁有自己的許可權機制。
vhost 特性：
1. RabbitMQ預設的vhost是“/”開箱即用
2. 多個vhost是隔離的，多個vhost無法通訊，並且不用擔心命名衝突（佇列和交換機進行繫結），實現了多層分離
3. 建立使用者的時候必須指定vhost
vhost操作：
1. 可以通過rabbitmqctl工具命令建立：
rabbitmqctl add_vhost[vhost_name]
2. 刪除vhost：
rabbitmqctl delete_vhost[vhost_name]
3. 檢視所有的vhost：
rabbitmqctl list_vhosts
RabbitMQ安裝及搭建
1. 選擇版本
選擇Erlang與RabbitMQ對應版本 RabbitMQ Erlang Version Requirements
2. 安裝Erlang
2.1 下載Erlang需要依賴的包
cd / mkdir downloads
2.2 將Erlang需要的依賴通過yum下載並安裝
yum -y install make gcc gcc-c++ kernel-devel m4 ncurses-devel openssl-devel unixODBC-devel
注：Erlang所需依賴下載安裝完成後，開始下載Erlang。
2.3 開始下載Erlang
cd /usr/local/ mkdir rabbitmq cd rabbitmq
wget http://www.rabbitmq.com/releases/erlang/erlang-19.0.4-1.el7.centos.x86_64.rpm
2.4 開始安裝Erlang
rpm -ivh erlang-19.0.4-1.el7.centos.x86_64.rpm
3. 安裝RabbitMQ
3.1 下載RabbitMQ
cd /usr/local/rabbitmq/
wget http://www.rabbitmq.com/releases/rabbitmq-server/v3.6.6/rabbitmq-server-3.6.6-1.el7.noarch.rpm
rpm -ivh rabbitmq-server-3.6.6-1.el7.noarch.rpm
3.2 在安裝RabbitMQ過程中提示依賴socat軟體
yum install socat
3.3 安裝完socat，重新執行安裝RabbitMQ
3.4 查詢安裝RabbitMQ bin的位置
3.5 啟動RabbitMQ
cd /sbin/ ./rabbitmq-server -detached # 啟動服務命令 ./rabbitmqctl stop # 停止服務命令 ./rabbitmq-plugins enable rabbitmq_management # 安裝管理外掛 ./rabbitmq-server -detached # 重新啟動服務命令 ./rabbitmqctl status # 檢視服務執行狀態 wget http://master1:15672 # 驗證是否成功安裝
3.6 建立使用者
./rabbitmqctl add_user admin admin ./rabbitmqctl set_user_tags admin administrator ./rabbitmqctl set_permissions -p "/" admin ".*" ".*" ".*" ./rabbitmqctl list_users
如果發現遠端訪問不了： http ： //192.168.56.101:15672
應該是 iptables 的原因
1.關閉firewall [root@localhost ~]# systemctl stop firewalld.service //停止firewall [root@localhost ~]# systemctl disable firewalld.service //禁止firewall開機啟動 2.安裝iptables [root@localhost ~]# yum install iptables-services //安裝 [root@localhost ~]# systemctl restart iptables.service #重啟防火牆使配置生效 [root@localhost ~]# systemctl enable iptables.service #設定防火牆開機啟動 [root@localhost ~]# systemctl disable iptables.service #禁止防火牆開機啟動
4. RabbitMQ叢集配置
4.1 伺服器
192.168.56.101，192.168.56.102
4.2 在102機器按照之上101模式重新安裝RabbitMQ即可
4.3 兩臺RabbitMQ服務重啟後，通過cluster_status命令檢視叢集狀態
4.4 連線叢集
為了連線叢集中的兩個節點，我們把rabbit@slave1加入到rabbit@master1節點叢集中，調整兩臺伺服器hosts檔案，加入域名
1、 停止 rabbit@salve1 的 rabbitmq 應用
2、 加入 rabbit@master1 叢集
3、 啟動 rabbit@salve1
4、 在 master1/slave1 任意一個節點檢視叢集狀態
cd /sbin/ ./rabbitmqctl stop_app ./rabbitmqctl join_cluster rabbit@master1 ./rabbitmqctl start_app ./rabbitmqctl cluster_status
4.5 停止master1檢視叢集狀態
針對master1操作：
cd /sbin/ ./rabbitmqctl stop
4.6 重啟master1檢視叢集狀態
4.7 注意問題
當整個叢集關閉時，最後一個關閉的節點必須是第一個要聯機的節點。
如果要離線的最後一個節點無法恢復，可以使用forget_cluster_node命令將其從群集中刪除
如果所有叢集節點同時停止並且不受控制（例如斷電），則可能會留下所有節點都認為其他節點在其後停止的情況。在這種情況下，您可以在一個節點上使用force_boot命令使其再次可引導
4.8 移除叢集
當節點不再是節點的一部分時，需要從叢集中明確地刪除節點。我們首先從叢集中刪除rabbit@slave1，並將其返回到獨立操作。
在rabbit@slave1上：
1、停止RabbitMQ應用程式
2、重置節點
3、重新啟用RabbitMQ應用程式
./rabbitmqctl stop_app ./rabbitmqctl reset ./rabbitmqctl start_app ./rabbitmqctl cluster_status
4、遠端刪除節點，在處理無響應的節點服務
例如我們在master1上將slave1從叢集中移除
- 停止節點slave1
- 在master1上通過forget_cluster_node命令移除slave1
- 重置slave1節點
- 啟動slave1節點
Slave1伺服器：
./rabbitmqctl stop_app ./rabbitmqctl reset # 執行此命令前 先執行master1 上的forget命令 ./rabbitmqctl start_app ./rabbitmqctl cluster_status
master1伺服器：
./rabbitmqctl forget_cluster_node rabbit@slave1 # 執行此命令前 先執行slave1 停止節點 ./rabbitmqctl cluster_status
4.9 主機名更改
RabbitMQ節點使用主機名相互通訊。因此，所有節點名稱必須能夠解析所有叢集對等的名稱。像rabbitmqctl這樣的工具也是如此
除此之外，預設情況下RabbitMQ使用系統的當前主機名來命名資料庫目錄。如果主機名更改，則會建立一個新的空資料庫。為了避免資料丟失，建立一個固定和可解析的主機名至關重要。每當主機名更改時，您應該重新啟動RabbitMQ
如果要使用節點名稱的完整主機名（RabbitMQ預設為短名稱），並且可以使用DNS解析完整的主機名，則可能需要調查設定環境變數 RABBITMQ_USE_LONGNAME = true
5. 使用HAProxy實現RabbitMQ叢集的負載
5.1 資源包下載，並上傳伺服器
https://download.csdn.net/download/u014518626/10122459?web=web
cd /usr/local/ mkdir haproxy
5.2 解壓
tar -xzvf haproxy-1.7.9.tar.gz
5.3 檢視核心
uname -r
5.4 編譯安裝
make TARGET=linux2628 make install
安裝後，檢視版本
./haproxy -v
複製haproxy檔案到 /usr/sbin 下，因為下面的 haproxy.init 啟動指令碼預設會去 /usr/sbin 下找，當然你也可以修改，不過比較麻煩。
cp haproxy /usr/sbin/
複製 haproxy 指令碼，到 /etc/init.d 下
cp ./examples/haproxy.init /etc/init.d/haproxy chmod 755 /etc/init.d/haproxy
5.5 建立系統賬號
useradd -r haproxy
5.6 建立配置檔案
mkdir /etc/haproxy vi /etc/haproxy/haproxy.cfg
Haproxy.cfg內容如下：
global log 127.0.0.1 local0 info chroot /usr/local/haproxy user haproxy group haproxy daemon maxconn 4096 defaults logglobal modetcp optiontcplog optiondontlognull retries 3 optionabortonclose maxconn 4096 timeout connect5000ms timeout client3000ms timeout server3000ms balance roundrobin listen private_monitoring bind0.0.0.0:8100 modehttp optionhttplog statsrefresh5s statsuri/stats statsrealmHaproxy statsauthadmin:admin listen rabbitmq_admin bind0.0.0.0:8102 servermaster1192.168.56.101:15672 serverslave1192.168.56.102:15672 listen rabbitmq_cluster bind0.0.0.0:8101 modetcp optiontcplog balance roundrobin timeout client3h timeout server3h servermaster1 192.168.56.101:5672checkinter3000rise2fall3 serverslave1192.168.56.102:5672checkinter3000rise2fall3
由於我打開了 log 127.0.0.1 local0 info 日誌，需要調整rsyslog配置檔案。
開啟rsyslog配置
vi /etc/rsyslog.conf
去掉下面兩行前面的#號
$ModLoad imudp
$UDPServerRun 514
並新增下面一行
local0.* /var/log/haproxy.log
重新啟動rsyslog
systemctl restart rsyslog
啟動haproxy
service haproxy start
6. Springboot 整合RabbitMQ叢集
6.1 引入依賴
<dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-amqp</artifactId> </dependency>
6.2 地址配置
--- # 叢集 RabbitMQ spring: rabbitmq: host: 192.168.56.101 port: 8101 username: admin password: admin
注意：8101 是 haproxy 負載均衡器的地址，因為之上我已經使用了 haproxy作為 rabbitmq叢集的負載器
6.3 建立佇列
package com.sinosoft.config;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/**
* Created by xushuyi on 2017/7/20.
*/
@Configuration
public class Queues {
/**
* 建立佇列 hello.foo
* @return
*/
@Bean
public Queue helloQueue() {
return new Queue("hello.foo");
}
}
6.4 生產訊息
package com.sinosoft.rabbit;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import java.util.Date;
/**
* Created by xushuyi on 2017/7/20.
*/
@Component
public class RabbitSend {
@Autowired
private AmqpTemplate rabbitTemplate;
public void send(String val) {
String context = "hello " + new Date();
System.out.println("Sender : " + val);
this.rabbitTemplate.convertAndSend("hello.foo", val);
}
}
6.5 訊息消費
package com.sinosoft.rabbit;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
/**
* Created by xushuyi on 2017/7/20.
*/
@Component
public class RabbitReceive {
/**
* 接收通道
*
* @param i
*/
@RabbitListener(queues = "hello.foo")
public void process(String i) {
System.out.println("執行接收Receiver1: " + i.toString());
}
}
注：這是一個非常精簡的demo示例，已經可以達到佇列訊息的生產與消費，當然還有很多詳細的配置內容，在這裡就不多贅述了。
技術參考：
2.RabbitMQ叢集