Rabbitmq集群

1、 rabbitmq介紹
RabbitMQ是一個開源的靠AMQP協議實現的服務，服務器端用Erlang語言編寫，支持多種客戶端，如：Python、Ruby、.NET、Java、JMS、C、PHP、ActionScript、XMPP、STOMP等，支持AJAX。用于在分布式系統中存儲轉發消息，在易用性、擴展性、高可用性等方面表現不俗。
AMQP，即Advanced Message Queuing Protocol，高級消息隊列協議，是應用層協議的一個開放標準，為面向消息的中間件設計。消息中間件主要用于組件之間的解耦，消息的發送者無需知道消息使用者的存在，反之亦然。
它可以使對應的客戶端（client）與對應的消息中間件（broker）進行交互。消息中間件從發布者（publisher）那里收到消息（發布消息的應用，也稱為producer），然后將他們轉發給消費者（consumers，處理消息的應用）。由于AMQP是一個網絡協議，所以發布者、消費者以及消息中間件可以部署到不同的物理機器上面
2、 消息隊列的概念
消息即是信息的載體。為了讓消息發送者和消息接收者都能夠明白消息所承載的信息（消息發送者需要知道如何構造消息；消息接收者需要知道如何解析消息），它們就需要按照一種統一的格式描述消息，這種統一的格式稱之為消息協議。所以，有效的消息一定具有某一種格式；而沒有格式的消息是沒有意義的。
3、 消息隊列應用的場景
消息隊列在實際應用中常用的使用場景。異步處理，應用解耦，流量削鋒和消息通訊四個場景
3.1異步處理
場景說明：如用戶注冊后，需要發送郵件和注冊短信，傳統的做法有兩種
1）串行方式：將注冊信息寫入數據庫成功后，發送注冊郵件，再發送注冊短信。以上三個任務全部完成后，返回給客戶端。
2）并行方式：將注冊信息寫入數據庫成功后，發送注冊郵件的同時，發送注冊短信。以上三個任務完成后，返回給客戶端。與串行的差別是，并行的方式可以提高處理的時間。
假設三個業務節點每個使用50毫秒鐘，不考慮網絡等其他開銷，則串行方式的時間是150毫秒，并行的時間可能是100毫秒。
如以上案例描述，傳統的方式系統的性能（并發量，吞吐量，響應時間）會有瓶頸。如何解決這個問題呢？
引入消息隊列，將不是必須的業務邏輯，異步處理。改造后的架構如下：
按照以上約定，用戶的響應時間相當于是注冊信息寫入數據庫的時間，也就是50毫秒。注冊郵件，發送短信寫入消息隊列后，直接返回，因此寫入消息隊列的速度很快，基本可以忽略，因此用戶的響應時間可能是50毫秒。
3.2應用解耦
場景說明：用戶下單后，訂單系統需要通知庫存系統。傳統的做法是，訂單系統調用庫存系統的接口

傳統模式的缺點：假如庫存系統無法訪問，則訂單減庫存將失敗，從而導致訂單失敗，訂單系統與庫存系統耦合
如何解決以上問題呢？引入應用消息隊列后的方案，如下圖：

訂單系統：用戶下單后，訂單系統完成持久化處理，將消息寫入消息隊列，返回用戶訂單下單成功
庫存系統：訂閱下單的消息，采用拉/推的方式，獲取下單信息，庫存系統根據下單信息，進行庫存操作
假如：在下單時庫存系統不能正常使用。也不影響正常下單，因為下單后，訂單系統寫入消息隊列就不再關心其他的后續操作了。實現訂單系統與庫存系統的應用解耦
3.3、流量削峰
流量削鋒也是消息隊列中的常用場景，一般在秒殺或團搶活動中使用廣泛
應用場景：秒殺活動，一般會因為流量過大，導致流量暴增，應用掛掉。為解決這個問題，一般需要在應用前端加入消息隊列。
a、可以控制活動的人數
b、可以緩解短時間內高流量壓垮應用

用戶的請求，服務器接收后，首先寫入消息隊列。假如消息隊列長度超過最大數量，則直接拋棄用戶請求或跳轉到錯誤頁面。秒殺業務根據消息隊列中的請求信息，再做后續處理
3.4、日志處理
日志處理是指將消息隊列用在日志處理中，比如Kafka的應用，解決大量日志傳輸的問題。架構簡化如下

日志采集客戶端，負責日志數據采集，定時寫受寫入Kafka隊列
Kafka消息隊列，負責日志數據的接收，存儲和轉發
日志處理應用：訂閱并消費kafka隊列中的日志數據
3.5、消息通訊
消息通訊是指，消息隊列一般都內置了高效的通信機制，因此也可以用在純的消息通訊。比如實現點對點消息隊列，或者聊天室等
點對點通訊：

客戶端A和客戶端B使用同一隊列，進行消息通訊。
聊天室通訊：

客戶端A，客戶端B，客戶端N訂閱同一主題，進行消息發布和接收。實現類似聊天室效果
4、 常見的消息隊列產品
4.1、redis
是一個Key-Value的NoSQL數據庫，開發維護很活躍，雖然它是一個Key-Value數據庫存儲系統，但它本身支持MQ功能，所以完成可以當做一個輕量級的隊列服務來使用。對于RabbitMQ和Redis的入隊和出隊操作，各執行100萬次，每10萬次記錄一次執行時間。測試數據分為128Bytes、512Bytes、1K和10K四個不同大小的數據。實驗表明：入隊時，當數據比較小時，Redis的性能要高于RabbitMQ，而如否數據大小超過了10K，Redis則慢的無法忍受；出隊時，無論數據大小，Redis都表現出非常好的性能，而RabbitMQ的出隊性能則遠低于Redis。
4.2、 mecahceq
持久化消息隊列（簡稱mcq）是一個輕量級的消息隊列，特性如下：
簡單易用
處理速度快
多條隊列
并發性能好
與memcache的協議兼容。意味著只要裝了前者的extension即可不需要額外的插件
在zend framework中使用很方便 php開發框架
4.3、 MSMQ
這是微軟的產品里唯一被認為有價值的東西關鍵是它并不復雜，除了接收和發送，沒有別的，它有一些硬性限制，比如最大消息體積是4MB。
4.4、 ZeroMQ
ZeroMQ是一個非常輕量級的消息系統，號稱最快的消息隊列系統，專門為高吞吐量/低延遲的場景開發，在金融界的應用中經?？梢园l現它。
與RabbitMQ相比，ZeroMQ支持許多高級消息場景，能夠實現RabbitMQ不擅長的高級/復雜的隊列，但是你必須實現ZeroMQ框架中的各個塊（比如Socket或Device等）。
ZeroMQ具有一個獨特的非中間件的模式
你不需要安裝和運行一個消息服務器或中間件，因為你的應用程序將扮演這個服務角色。你只需要簡單地引用ZeroMQ程序庫，可以使用NuGet安裝（微軟開發的.Net平臺），然后你就可以愉快地在應用程序之間發送消息了。
但是ZeroMQ僅提供非持久性的隊列，即沒有地方可以觀察它是否有問題出現，也就是說如果down機，數據將會丟失。
4.5、 Jafka/Kafka
Jafka/Kafka（能將消息分散到不同的節點上）是LinkedIn于2010年12月開發并開源的一個分布式MQ系統，現在是Apache的一個孵化項目，是一個高性能跨語言分布式Publish/Subscribe消息隊列系統，而Jafka是在Kafka之上孵化而來的，即Kafka的一個升級版。具有以下特性：
1）快速持久化，可以在O(1)的系統開銷下進行消息持久化；
2）高吞吐，在一臺普通的服務器上既可以打到10W/s的吞吐速率；
3）完全的分布式系統，Broker、Producer、Consumer都原生自動支持分布式，自動實現復雜均衡；
4）支持Hadoop數據并行加載，統一了在線和離線的消息處理，對于像Hadoop一樣的日志數據和離線分析系統，但又要求實時處理的限制，這是一個可行的解決方案。
5）相對于ActiveMQ是一個非常輕量級的消息系統，除了性能非常好之外，還是一個工作良好的分布式系統
4.6、 Apache ActiveMQ
ActiveMQ居于（RabbitMQ&ZeroMQ）之間，類似于ZemoMQ，支持高級消息場景，支持數據的持久化
ActiveMQ被譽為Java世界的中堅力量。它有很長的歷史，且被廣泛使用。它還是跨平臺的，給那些非微軟平臺的產品提供了一個天然的集成接入點。
然而它只有跑過了MSMQ才有可能被考慮。如需配置ActiveMQ則需要在目標機器上安裝Java環境。
類似于RabbitMQ，它易于實現高級場景，而且只需付出低消耗。它被譽為消息中間件的“瑞士軍刀”。
4.7、 RabbitMQ
RabbitMQ是使用Erlang編寫的一個開源消息隊列，本身支持很多的協議：AMQP（高級消息隊列協議）， XMPP(可擴展消息處理現場協議)， SMTP（簡單郵件傳輸協議）， STONP（簡單文本定向消息協議），也正是如此，使的它變的非常重量級，更適合于企業級的開發。
它實現了代理(Broker)架構，意味著消息在發送到客戶端之前可以在中央節點上排隊。此特性使得RabbitMQ易于使用和部署，適宜于很多場景如路由、負載均衡或消息持久化等，用消息隊列只需幾行代碼即可搞定。但是，這使得它的可擴展性差，速度較慢，因為中央節點增加了延遲，消息封裝后也比較大。如需配置RabbitMQ則需要在目標機器上安裝Erlang環境
最終，上述同類產品：
1． 都有各自客戶端API或支持多種編程語言
2． 都有大量的文檔
3． 都提供了積極的支持
4． ActiveMQ、RabbitMQ、MSMQ、Redis都需要啟動服務進程，這些都可以監控 和配置，其他幾個就有問題了
5． 都相對提供了良好的可靠性（一致性）、擴展性和負載均衡，當然還有性能

5、Rabbitmq基礎概念
應用場景框架

RabbitMQ Server：也叫broker server，它不是運送食物的卡車，而是一種傳輸服務。原話是RabbitMQ isn’t a food truck, it’s a delivery service. 他的角色就是維護一條從Producer到Consumer的路線，保證數據能夠按照指定的方式進行傳輸。但是這個保證也不是100%的保證，但是對于普通的應用來說這已經足夠了。當然對于商業系統來說，可以再做一層數據一致性的guard（監控是否執行成功），就可以徹底保證系統的一致性了。
Client A & B：也叫Producer，數據的發送方。Create messages and Publish (Send) them to a broker server (RabbitMQ)。一個Message有兩個部分：Payload（有效載荷）和Label（標簽）。Payload顧名思義就是傳輸的數據，Label是Exchange的名字或者說是一個tag，它描述了payload，而且RabbitMQ也是通過這個label來決定把這個Message發給哪個Consumer。AMQP僅僅描述了label，而RabbitMQ決定了如何使用這個label的規則。
Client 1，2，3：也叫Consumer，數據的接收方。Consumers attach to a broker server (RabbitMQ) and subscribe to a queue。把queue比作是一個有名字的郵箱。當有Message到達某個郵箱后，RabbitMQ把它發送給它的某個訂閱者即Consumer。當然可能會把同一個Message發送給很多的Consumer。在這個Message中，只有payload，label已經被刪掉了。對于Consumer來說，它是不知道誰發送的這個信息的。就是協議本身不支持。但是當然了如果Producer發送的payload包含了Producer的信息就另當別論了。
對于一個數據從Producer到Consumer的正確傳遞，還有三個概念需要明確：exchanges, queues and bindings。
Exchanges：消息交換機，它指定消息按什么規則，路由到哪個隊列
Queues：消息隊列載體，每個消息都會被投入到一個或多個隊列
Bindings：它的作用就是把exchange和queue按照路由規則綁定起來
Routing Key：路由關鍵字，exchange根據這個關鍵字進行消息投遞
Connection：就是一個TCP的連接。
Producer和Consumer都是通過TCP連接到RabbitMQ Server的。以后我們可以看到，程序的起始處就是建立這個TCP連接。
Channel：虛擬連接。它建立在上述的TCP連接中。數據流動都是在Channel中進行的。也就是說，一般情況是程序起始建立TCP連接，第二步就是建立這個Channel。
Vhost：虛擬主機，一個broker里可以開設多個vhost，用作不同用戶的權限分離。每個virtual host本質上都是一個RabbitMQ Server，擁有它自己的queue，exchagne，和bings rule等等。這保證了你可以在多個不同的application中使用RabbitMQ。
6、Channel的選擇
那么，為什么使用Channel，而不是直接使用TCP連接？
對于OS來說，建立和關閉TCP連接是有代價的，頻繁的建立關閉TCP連接對于系統的性能有很大的影響，而且TCP的連接數也有限制，這也限制了系統處理高并發的能力。但是，在TCP連接中建立Channel是沒有上述代價的。對于Producer或者Consumer來說，可以并發的使用多個Channel進行Publish或者Receive。
7、消息隊列執行過程
客戶端連接到消息隊列服務器，打開一個Channel。
客戶端聲明一個Exchange，并設置相關屬性。
客戶端聲明一個Queue，并設置相關屬性。
客戶端使用Routing key，在Exchange和Queue之間建立好綁定關系。
客戶端投遞消息到Exchange。
Exchange接收到消息后，就根據消息的key和已經設置的Binding，進行消息路由，將消息投遞到一個或多個隊列里。
8、消息持久化
RabbitMQ支持消息的持久化，也就是數據寫在磁盤上，為了數據安全考慮，大多數用戶都會選擇持久化。消息隊列持久化包括3個部分：
Exchange持久化，在聲明時指定durable => 1
Queue持久化，在聲明時指定durable => 1
消息持久化，在投遞時指定delivery_mode => 2（1是非持久化）
若Exchange和Queue都是持久化的，那么它們之間的Binding也是持久化的；而Exchange和Queue兩者之間有一個持久化，一個非持久化，就不允許建立綁定。
Consumer從durable queue中取回一條消息之后并發回了ack消息，RabbitMQ就會將其標記，方便后續垃圾回收。如果一條持久化的消息沒有被consumer取走，RabbitMQ重啟之后會自動重建exchange和queue(以及bingding關系)，消息通過持久化日志重建再次進入對應的queues，exchanges。
9、安裝rabbitmq

 24  yum  -y   localinstall  erlang-18.1-1.el6.x86_64.rpm   rabbitmq-server-3.6.6-1.el6.noarch.rpm   socat-1.7.3.2-2.el7.x86_64.rpm   （安裝rabbitmq）
  22  chkconfig  --add  rabbitmq-server     (加入開機自啟)
   25  chkconfig  rabbitmq-server on        （開啟rabbitmq開機自啟）
   26  /etc/init.d/rabbitmq-server  start        （開啟rabbitmq）
查看rebbitmq是否運行
27  ps -ef | grep rabbitmq

30 rabbitmq-plugins enable rabbitmq_management

#開啟rabbitmq的web管理插件，用戶可以通過瀏覽器進行訪問
31 rabbitmqctl add_user admin 123.com

#創建登錄用戶admin 密碼123.com
32 rabbitmqctl set_user_tags admin administrator

#將admin用戶添加到管理員組當中

查看端口netstat -anpt | grep 15672

瀏覽器訪問IP地址:15672


10、Rabbitmq集群
（1.40-1.60的操作）
集群方式
Rabbitmq集群大概分為二種方式：

1. 普通模式：默認的集群模式。消息的實體只存在一個節點上
2. 鏡像模式：把需要的隊列做成鏡像隊列，存在于多個節點。
   ha-mode：all 列隊到所有節點上
   exatly：隨機鏡像到其他節點上
   nodes：鏡像到指定節點上
   集群節點模式：
   1、 內存節點：工作在內存上
   2、 磁盤節點：工作在磁盤上
   例外：內存節點和磁盤節點共同存在，提高了訪問速度有增加了持久化
   相比內存節點雖然不寫入磁盤，但是它執行比磁盤節點要好。集群中，只需要一個磁盤節點來保存狀態 就足夠了如果集群中只有內存節點，那么不能停止它們，否則所有的狀態，消息等都會丟失。
   安裝rabbitmq
   

   24  yum  -y   localinstall  erlang-18.1-1.el6.x86_64.rpm   rabbitmq-server-3.6.6-1.el6.noarch.rpm   socat-1.7.3.2-2.el7.x86_64.rpm   （安裝rabbitmq）
   22  chkconfig  --add  abbitmq-server     (加入開機自啟)
   25  chkconfig  rabbitmq-server on        （開啟rabbitmq開機自啟）
   26  /etc/init.d/rabbitmq-server  start        （開啟rabbitmq）
   查看rebbitmq是否運行
   27  ps -ef | grep rabbitmq

   

   [root@localhost ~]# vim /etc/hosts   （四臺主機都要操作）
   192.168.1.30  rabbitmq1
   192.168.1.40  rabbitmq2
   192.168.1.50  rabbitmq3
   192.168.1.60  rabbitmq4
   ##不想每臺都手寫可以使用scp
   [root@rabbitmq1 ~]# scp   /etc/hosts root@192.168.1.40:/etc/hosts ##提示輸入yes

在rabbitmq1上查看cookie節點信息并復制
##安裝集群的時候需要節點cookie信息一致

rabbitmq2，3和4上將文件cookie信息和rabbitmq1改成相同

[root@rabbitmq2 ~]# echo  MIIPQWTLIDGVGKMDWQFX > /var/lib/rabbitmq/.erlang.cookie
[root@rabbitmq3 ~]#echo  MIIPQWTLIDGVGKMDWQFX > /var/lib/rabbitmq/.erlang.cookie   
[root@rabbitmq4 ~]#echo  MIIPQWTLIDGVGKMDWQFX > /var/lib/rabbitmq/.erlang.cookie   

配置完成后reboot重啟虛擬機，
##注意：2和3需要手動重啟，重啟完成后主機名會變為rabbitmq1、2、3，啟動不了的開機界面會一直停留在這個地方5-6分鐘恢復，可以再次手動重啟一下就不會這樣了

[root@rabbitmq1 ~]# ps -ef | grep rabbit ##重啟完成后查看是否啟動

在rabbitmq1上操作
45 rabbitmqctl stop_app

46 rabbitmqctl reset

47 rabbitmqctl start_app

##設置完成后會提示節點名稱并復制

在rabbitmq2、3，4上加入節點（2，3和4操作相同）

[root@rabbitmq2 ~]# rabbitmqctl  stop_app
Stopping node rabbit@rabbitmq2 ...
[root@rabbitmq2 ~]# rabbitmqctl  reset
Resetting node rabbit@rabbitmq2 ...
[root@rabbitmq2 ~]# rabbitmqctl   join_cluster --ram rabbit@rabbitmq1
Clustering node rabbit@rabbitmq2 with rabbit@rabbitmq1 ...
##-join_cluter加入集群 --ram 以內存節點方式加入后面跟節點名rabbit@rabbitmq1
[root@rabbitmq2 ~]# rabbitmqctl  start_app
Starting node rabbit@rabbitmq2 ...
[root@rabbitmq2 ~]# rabbitmq-plugins enable rabbitmq_management
The following plugins have been enabled:
  mochiweb
  webmachine
  rabbitmq_web_dispatch
  amqp_client
  rabbitmq_management_agent
  rabbitmq_management
Applying plugin configuration to rabbit@rabbitmq2... started 6 plugins.
##開啟rabbitmq插件，不打開的話，就不能使用瀏覽器訪問rabbit頁面進行管理

回到rabbitmq1上創建管理用戶和查看集群狀態

[root@rabbitmq1 ~]# rabbitmqctl add_user admin redhat
Creating user "admin" ...
[root@rabbitmq1 ~]# rabbitmqctl set_user_tags admin administrator
Setting tags for user "admin" to [administrator] ...
##創建用戶admin密碼為redhat，并加入到管理員組中
[root@rabbitmq1 ~]# rabbitmq-plugins enable rabbitmq_management
Plugin configuration unchanged.

Applying plugin configuration to rabbit@rabbitmq1... nothing to do.
[root@rabbitmq1 ~]# rabbitmqctl cluster_status   ##查看節點狀態

##rabbit1工作模式為磁盤節點  rabbit2、3，4為ram內存節點模式
##running_nodes：正在運行的節點
##cluster_name：節點名稱
##alarms：發生問題時rabbit1、2、3，4會進行報警

瀏覽器訪問進行管理：
http://192.168.1.30:15672 用戶名為admin密碼redhat


點擊vxgp后在下面找到premissions選項設置為admin用戶權限訪問

設置完成后再次查看

設置匹配策略

發布消息：

設置發布消息內容


可以看到已經有通知了

添加rabbitmq節點：rabbitmq4 192.168.83.4
rabbitmq1、2、3添加hosts主機都添加192.168.83.4 rabbitmq4

rabbitmq4安裝同上面安裝一樣
[root@localhost ~]# chkconfig rabbitmq-server on
[root@localhost ~]# /etc/init.d/rabbitmq-server  start 
[root@localhost ~]# echo  MIIPQWTLIDGVGKMDWQFX > /var/lib/rabbitmq/.erlang.cookie
[root@localhost ~]# reboot
[root@rabbitmq4 ~]# rabbitmqctl  stop_app
[root@rabbitmq4 ~]# rabbitmqctl  reset
[root@rabbitmq4 ~]#  rabbitmqctl join_cluster --ram rabbit@rabbitmq1

刪除節點：
##在rabbitmq4上先停止節點
[root@rabbitmq1 ~]# rabbitmqctl stop_app
回到主節點上rabbitmq1
[root@rabbitmq1 ~]# rabbitmqctl -n rabbit@rabbitmq1 forget_cluster_node rabbit@rabbitmq4
Removing node rabbit@rabbitmq4 from cluster ...
##-n 指定節點名稱
#=forget_cluster_node 后面跟要刪除的節點名稱