業界消息總線技術分析-RabbitMQ

1????? 概述

RabbitMQ是一個由erlang開發的AMQP（Advanced Message Queue ）的開源實現，最初由RabbitMQ Technologies Ltd開發并且提供商業支持的。該公司在2010年4月被SpringSource（VMWare的一個部門）收購。在2013年5月被并入Pivotal。

1.1????? 官網介紹

為應用提供健壯的消息投遞，易于使用，能運行于多個主流的操作系統，多種類型的開發平臺。

1.2????? 亮點

可靠性：RabbitMQ提供多種功能讓你能夠在性能與可靠性之間做出權衡，包括持久化、消息投遞確認、發布者確認和高用性。

靈活的路由：消息在到達queues之前，通過exchanges進行路由。RabbitMQ為典型的路由邏輯內建了多種exchange類型。你能夠將多種exchanges綁定在一起，也能夠以插件的形式，自己實現一個exchange。

集群：在同一個本地網絡中的多個RabbitMQ server能夠組成集群，形成一個統一的邏輯broker。

聯合：如果多個server只需要一個松散的、不可靠的連接，則可以通過RabbitMQ的聯合模式實現。

高可用的queues：Queues能夠在集群內多臺機器之間鏡像，確保在某一臺機器故障時的消息安全。

多協議：RabbitMQ支持多種消息協議。核心協議是AMQP 0-9-1，另外還能夠以插件的形式，支持STOMP、MQTT、AMQP1.0等。

多客戶端：你能想到的任何語言幾乎都有RabbitMQ的客戶端。

管理界面：提供容易使用的管理界面來監控broker。

消息跟蹤：RabbitMQ提供消息跟蹤功能。

支持插件：支持以插件的形式，擴展功能。

1.3????? RabbitMQ解決什么問題？

1）信息的發送者和接收者如何維持連接，如果一方的連接中斷，這期間的數據如何防止丟失？

2）如何降低發送者和接收者的耦合度？

3）如何讓Priority高的接收者先接到數據？

4）如何做到load balance？有效均衡接收者的負載？

5）如何有效的將數據發送到相關的接收者？也就是說接收者subscribe 不同的數據，如何做有效的filter。

6）如何做到可擴展，甚至將這個通信模塊發到cluster上？

7）如何保證接收者接收到了完整，正確的數據？

AMQP協議解決了以上的問題，而RabbitMQ實現了AMQP。

2????? 性能吞吐量

2.1????? 與redis對比

測試環境

個人筆記本，Server與Client使用同一臺機器。

Processor : Inter(R) Core(TM)2 Duo CPU P8400 @2.26GHz RAM : 4.00GB Operating System : 32-bit Ubuntu 10.10。

軟件環境

RabbitMQ和Redis的Server啟動都是用默認配置，客戶端使用Java，JVM啟動使用默認參數：

RabbitMQ Server 2.2.0

RabbitMQ Java Client 2.2.0

Redis 2.0 Stable

Jedis

入隊性能

出隊性能

測試結論

對于入隊操作，當數據比較小時Redis的性能要高于RabbitMQ，而如果數據大小超過了10K，Redis慢的無法忍受。

對于出隊操作，無論數據大小，Redis都表現出非常好的性能，而RabbitMQ的出隊性能則遠低于Redis。

2.2????? 與kafka對比

測試環境

ubuntu 15.10 64位

cpu:inter core i7-4790 3.60GHZ * 8

內存:16GB

硬盤:ssd 120GB

軟件環境：rabbmitmq 3.6.0 ? kafka0.8.1 ?(均為單機本機運行)

測試結果均為單操作測試，即生產的時候沒有消費操作

測試結果

kafka :消費速度: 37,586 /s ?生產速度: 448,753 /s

rabbitmq: 消費速度: 20,807 /s ?生產速度 ?16,413 /s

測試結論

很明顯的看出kafka的性能遠超rabbitmq。

3? ? ?RabbitMQ的架構

架構圖如下：

從架構圖可以看出，Procuder Publish的Message進入了Exchange。接著通過“routing keys”， RabbitMQ會找到應該把這個Message放到哪個queue里。

有三種類型的Exchanges：direct， fanout，topic。每個實現了不同的路由算法（routing algorithm）。

Direct exchange: 如果routing key 匹配, 那么Message就會被傳遞到相應的queue中。

Fanout exchange: 會向相應的queue廣播。

Topic exchange: 對key進行模式匹配，比如ab*可以傳遞到所有ab*的queue。

4? ? ?分布式模型

4.1????? 三種模式

集群模式（Clustering）：

連接多臺機器，作為一個邏輯的broker。集群節點之間通過Erlang的消息進行通信，同一個集群內的每一個節點，都必須有相同的Erlang cookie。集群內的所有節點，都必須有相同版本的RabbitMR與Erlang。

聯合模式（Federation）：

聯合模式允許一個broker上的exchange或queue接收消息，然后發布到另外一個broker上的exchange或queue。這里的broker可以是獨立的機器，也可以是一個集群。聯合模式broker之間通過AMQP進行通信。

鏟子模式（The Shovel）：

使用鏟子連接brokers在概念上與聯合模式類似，不一樣的是，鏟子模式工作在更低的level。鏟子只是簡單地從一個broker的queue消費消息，然后轉發到另外一個broker的exchange上。

對照

4.2????? 集群模式

RabbitMQ邏輯視圖

RabbitMQ物理視圖

?? 分為磁盤節點與內存節點。

?? 隊列數據只能夠存儲在磁盤節點上。

?? 狀態數據存儲在內存節點與磁盤節點上，內存節點只把數據保存在內存上，因此能夠提高速度，但僅僅影響到資源管理。

?? 節點之間通過Erlang Cookie進行鑒權，也就是說，互相通信的兩個節點，一定要有相同的Erlang Cookie。

?? 集群可以隨時增加或減少節點，但需要注意的是，如果整個集群都宕掉，則恢復集群時最后宕掉的節點需要最先啟動。

4.3????? 普通集群

?? queue在集群中對于contents只存儲一份，其他節點只存儲meta信息。

?? 對于publish，客戶端任意連接集群的一個節點，轉發給創建queue的節點存儲消息的所有信息；

?? 對于consumer，客戶端任意連接集群中的一個節點，如果數據不在該節點中，則從存儲該消息data的節點拉取。

?? 可見當存儲有queue內容的節點失效后，只要等待該節點恢復后，queue中存在的消息才可以獲取消費的到。

?? 顯然增加集群的節點，可以提高整個集群的吞吐量，但是在高可用方面要稍微差一些。

4.4??集群鏡像隊列

?? mirror queue是為RabbitMQ高可用的一種方案，相對于普通的集群方案來講，queue中的消息每個節點都會存在一份或多份副本，具體有多少個副本，由策略進行配置。

?? 在配置了鏡像隊列之后，單個節點失效的情況下，整個集群仍舊可以提供服務。但是由于數據需要在多個節點復制，在增加可用性的同時，系統的吞吐量會有所下降。

?? 在實現機制上，mirror queue內部實現了一套選舉算法，有一個master和多個slave，queue中的消息以master為主。

?? 對于publish，可以選擇任意一個節點進行連接，若該節點不是master，則RabbitMQ將消息轉發給master，然后由master向其他slave節點發送該消息。

?? 對于consumer，可以選擇任意一個節點進行連接，消費的請求會轉發給master。為保證消息的可靠性，consumer需要進行ack確認，master收到ack后，才會刪除消息，ack消息會同步(默認異步方式)到其他各個節點，slave節點收到從master同步過來的ack，也會將對應的消息刪除。

?? 若master節點失效，則mirror queue會自動選舉出一個節點（slave中消息隊列最長者）作為master。

?? 若slave節點失效，mirror queue集群中其他節點的狀態無需改變。

5? ? ?網絡模型

?? 支持IPv4與IPv6雙協議棧

?? 底層使用TCP

?? 支持TLS

?? 官網介紹了較多針對傳輸層的調優措施，包括TCP Buffer Size、 OS Level Tuning、 TCP Socket Options等，具體可以參考 http://www.rabbitmq.com/networking.html

6? ? ?持久化

6.1????? 消息什么時候需要持久化

消息本身在publish的時候就要求消息寫入磁盤。

內存緊張，需要將部分內存中的消息轉移到磁盤。

6.2????? 消息什么時候會刷到磁盤

寫入文件前會有一個Buffer，大小為1M（1048576），數據在寫入文件時，首先會寫入到這個Buffer，如果Buffer已滿，則會將Buffer寫入到文件（未必刷到磁盤）；

有個固定的刷盤時間：25ms，也就是不管Buffer滿不滿，每隔25ms，Buffer里的數據及未刷新到磁盤的文件內容必定會刷到磁盤；

每次消息寫入后，如果沒有后續寫入請求，則會直接將已寫入的消息刷到磁盤：使用Erlang的receive x after 0來實現，只要進程的信箱里沒有消息，則產生一個timeout消息，而timeout會觸發刷盤操作。

6.3????? 消息在磁盤文件中的格式

消息保存于$MNESIA/msg_store_persistent/x.rdq文件中，其中x為數字編號，從1開始，每個文件最大為16M（16777216），超過這個大小會生成新的文件，文件編號加1。消息以以下格式存在于文件中：

<>

MsgId為RabbitMQ通過rabbit_guid:gen()每一個消息生成的GUID，MsgBody會包含消息對應的exchange，routing_keys，消息的內容，消息對應的協議版本，消息內容格式（二進制還是其它）等等。

6.4????? 文件何時刪除？

當所有文件中的垃圾消息（已經被刪除的消息）比例大于閾值（GARBAGE_FRACTION = 0.5）時，會觸發文件合并操作（至少有三個文件存在的情況下），以提高磁盤利用率。

publish消息時寫入內容，ack消息時刪除內容（更新該文件的有用數據大小），當一個文件的有用數據等于0時，刪除該文件。

7????? 消息QoS

可靠性

支持高可靠場景，一個master加上多個slave來保證消息的可靠。

安全

支持使用SASL進行認證

支持使用SSL進行通道加密

支持ACL

消息傳遞的優先級

AMQP協議本身支持優先級隊列，具體的實現還得再分析。

8? ? ?總結

Broker部署較為復雜

基于Erlang語言實現，每一個節點都需要安裝Erlang語言運行環境。

配置非常靈活

支持多種分布式模式，集群模式又分為普通集群與集群鏡像隊列，提供多種路由策略。同時通過插件方式，可以很方便地擴展RabbitMQ的功能。

無中心節點，可以支撐大集群

組件集群是直接使用Erlang語言的特性，不像kafka需要一個zookeeper。

RabbitMQ節點也沒有

支持多種類型的協議

不僅僅支持RabbitMQ，還通過插件形式，支持MQTT、STOMP等。

RabbitMQ 存儲

版權聲明：本文內容由網絡用戶投稿，版權歸原作者所有，本站不擁有其著作權，亦不承擔相應法律責任。如果您發現本站中有涉嫌抄襲或描述失實的內容，請聯系我們jiasou666@gmail.com 處理，核實后本網站將在24小時內刪除侵權內容。

版權聲明：本文內容由網絡用戶投稿，版權歸原作者所有，本站不擁有其著作權，亦不承擔相應法律責任。如果您發現本站中有涉嫌抄襲或描述失實的內容，請聯系我們jiasou666@gmail.com 處理，核實后本網站將在24小時內刪除侵權內容。