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文件，加入域名

一、中止rabbit@salve1的rabbitmq應用

二、加入rabbit@master1集羣

三、啓動rabbit@salve1

四、在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上：

一、中止RabbitMQ應用程序

二、重置節點

三、從新啓用RabbitMQ應用程序

 

./rabbitmqctl stop_app
./rabbitmqctl reset
./rabbitmqctl start_app
./rabbitmqctl cluster_status

四、遠程刪除節點，在處理無響應的節點服務

例如咱們在master1上將slave1從集羣中移除

1. 中止節點slave1
2. 在master1上經過forget_cluster_node命令移除slave1
3. 重置slave1節點
4. 啓動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集羣