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 log global mode tcp option tcplog option dontlognull retries 3 option abortonclose maxconn 4096 timeout connect 5000ms timeout client 3000ms timeout server 3000ms balance roundrobin listen private_monitoring bind 0.0.0.0:8100 mode http option httplog stats refresh 5s stats uri /stats stats realm Haproxy stats auth admin:admin listen rabbitmq_admin bind 0.0.0.0:8102 server master1 192.168.56.101:15672 server slave1 192.168.56.102:15672 listen rabbitmq_cluster bind 0.0.0.0:8101 mode tcp option tcplog balance roundrobin timeout client 3h timeout server 3h server master1 192.168.56.101:5672 check inter 3000 rise 2 fall 3 server slave1 192.168.56.102:5672 check inter 3000 rise 2 fall 3
由於我打開了 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示例，已經可以達到隊列消息的生產與消費，當然還有很多詳細的配置內容，在這裏就不多贅述了。
技術參考：
1. Centos7 安裝RabbitMQ
2. RabbitMQ集群
Springboot 1.5.x 集成基於Centos7的RabbitMQ集群安裝及配置