一、rabbitmq架構(gòu)

RabbitMQ是一個(gè)流行的開源消息隊(duì)列系統(tǒng)，是AMQP（高級消息隊(duì)列協(xié)議）標(biāo)準(zhǔn)的實(shí)現(xiàn)，由以高性能、健壯、可伸縮性出名的Erlang語言開發(fā)，并繼承了這些優(yōu)點(diǎn)。

rabbitmq簡單架構(gòu)如下：

上圖簡單展示了rabbitmq的架構(gòu)，從圖中看到幾個(gè)關(guān)鍵字：vhost、exchange、route key、queue等，后面會(huì)介紹這些概念。

下面看下rabbitmq的進(jìn)程模型：

看到這個(gè)圖，相信大家應(yīng)該很熟悉，沒錯(cuò)就是事件驅(qū)動(dòng)模型（或者說反應(yīng)堆模型），這是一種高性能的非阻塞io線程模型，不過在Erlang中稱為進(jìn)程模型。

tcp_acceptor進(jìn)程接收客戶端連接，創(chuàng)建rabbit_reader、rabbit_writer、rabbit_channel進(jìn)程。

• rabbit_reader接收客戶端連接，解析AMQP幀；rabbit_writer向客戶端返回?cái)?shù)據(jù)；
• rabbit_channel解析AMQP方法，對消息進(jìn)行路由，然后發(fā)給相應(yīng)隊(duì)列進(jìn)程。
• rabbit_amqqueue_process是隊(duì)列進(jìn)程，在RabbitMQ啟動(dòng)（恢復(fù)durable類型隊(duì)列）或創(chuàng)建隊(duì)列時(shí)創(chuàng)建。
• rabbit_msg_store是負(fù)責(zé)消息持久化的進(jìn)程。

在整個(gè)系統(tǒng)中，存在一個(gè)tcp_accepter進(jìn)程，一個(gè)rabbit_msg_store進(jìn)程，有多少個(gè)隊(duì)列就有多少個(gè)rabbit_amqqueue_process進(jìn)程，每個(gè)客戶端連接對應(yīng)一個(gè)rabbit_reader和rabbit_writer進(jìn)程。

二、關(guān)于AMQP協(xié)議

1.AMQP幀組件

AMQP幀由五個(gè)不同的組件組成：

幀類型
信道編號
以字節(jié)為單位的幀大小
幀有效載荷payload
結(jié)束字節(jié)標(biāo)志（ASCII值206）

2.幀類型

AMQP規(guī)范定義了五種類型的幀：協(xié)議頭幀、方法幀、內(nèi)容幀、消息體幀及心跳幀。每種幀類型都有明確的目的，有些幀的使用頻率比其他的高很多：

協(xié)議頭幀用于連接到rabbitmq，進(jìn)使用一次。
方法幀攜帶發(fā)送給rabbitmq或者從rabbitmq接收到的rpc請求或者響應(yīng)
內(nèi)容頭包含一條消息的大小和屬性。
消息體幀包含消息的內(nèi)容
心跳幀在客戶端與rabbitmq直接進(jìn)行傳遞，作為一種校驗(yàn)機(jī)制確保連接的兩端都可用并且正常工作。

3.將消息編組成幀

我們使用方法幀、內(nèi)容頭幀和消息體幀組成一個(gè)完整的rabbitmq消息。方法頭幀攜帶命令和執(zhí)行它所需要的參數(shù)（如交換器和路由鍵）、內(nèi)容幀包含消息的基本屬性以及消息的大小，消息體幀也就是攜帶我們真正需要發(fā)送的消息內(nèi)容。

4.方法幀結(jié)構(gòu)

5.內(nèi)容頭幀結(jié)構(gòu)

內(nèi)容頭包含的具體屬性如下：

• content-type:消息體的報(bào)文編碼，如application/json
• expiration：消息過期時(shí)間
• reply-to：響應(yīng)消息的隊(duì)列名
• content-encoding：報(bào)文壓縮的編碼，如gzip
• message-id:消息的編號
• correlation-id：鏈路id
• deliver-mode:告訴rabbitmq將消息寫入磁盤還是內(nèi)存
• user-id：投遞消息的用戶（發(fā)送消息時(shí)不要設(shè)置該值）
• timestamp：投遞消息的時(shí)間
• headers:定義一些屬性，可用于實(shí)現(xiàn)rabbitmq路由（比如exchange類型是headers的時(shí)候用到）

6.消息體幀結(jié)構(gòu)

7.幾個(gè)概念

• Broker：簡單來說就是消息隊(duì)列服務(wù)器實(shí)體
• Exchange：消息交換機(jī)，它指定消息按什么規(guī)則，路由到哪個(gè)隊(duì)列
• Queue：消息隊(duì)列載體，每個(gè)消息都會(huì)被投入到一個(gè)或多個(gè)隊(duì)列，隊(duì)列類型又分為臨時(shí)隊(duì)列，持久化隊(duì)列，排他隊(duì)列
• Binding：綁定，它的作用就是把exchange和queue按照路由規(guī)則綁定起來
• Routing Key：路由關(guān)鍵字，exchange根據(jù)這個(gè)關(guān)鍵字進(jìn)行消息投遞
• vhost：虛擬主機(jī)，一個(gè)broker里可以開設(shè)多個(gè)vhost，用作不同用戶的權(quán)限分離
• producer：消息生產(chǎn)者，就是投遞消息的程序
• consumer：消息消費(fèi)者，就是接受消息的程序
• channel：消息通道，在客戶端的每個(gè)連接里，可建立多個(gè)channel，每個(gè)channel代表一個(gè)會(huì)話任務(wù)

四、通訊過程

1.啟動(dòng)會(huì)話

2.聲明交換器

3.聲明隊(duì)列

4.綁定隊(duì)列到exchange

5.發(fā)送消息-使用事務(wù)機(jī)制

對事務(wù)的支持是AMQP協(xié)議的一個(gè)重要特性。假設(shè)當(dāng)生產(chǎn)者將一個(gè)持久化消息發(fā)送給服務(wù)器時(shí)，假如使用no_ack模式，所以即使服務(wù)器崩潰，沒有持久化該消息，生產(chǎn)者也無法獲知該消息已經(jīng)丟失。

如果此時(shí)使用事務(wù)，即通過txSelect()開啟一個(gè)事務(wù)，然后發(fā)送消息給服務(wù)器，然后通過txCommit()提交該事務(wù)，即可以保證，如果txCommit()提交了，則該消息一定會(huì)持久化，如果txCommit()還未提交即服務(wù)器崩潰，則該消息不會(huì)服務(wù)器就收。

當(dāng)然Rabbit MQ也提供了txRollback()命令用于回滾某一個(gè)事務(wù)。但是使用事務(wù)，會(huì)導(dǎo)致性能下降，它使得生產(chǎn)者發(fā)布消息后必須等到消息真正持久化后服務(wù)端響應(yīng)了才結(jié)束本次連接，所以需要在實(shí)際應(yīng)用中平衡性能與安全的問題。

6.發(fā)送消息-非事務(wù)方式

使用事務(wù)固然可以保證只有提交的事務(wù)，才會(huì)被服務(wù)器執(zhí)行。但是這樣同時(shí)也將客戶端與消息服務(wù)器同步起來，這背離了消息隊(duì)列解耦的本質(zhì)。Rabbit MQ提供了一個(gè)更加輕量級的機(jī)制來保證生產(chǎn)者可以感知服務(wù)器消息是否已被路由到正確的隊(duì)列中——Confirm。如果設(shè)置channel為confirm狀態(tài)，則通過該channel發(fā)送的消息都會(huì)被分配一個(gè)唯一的ID，然后一旦該消息被正確的路由到匹配的隊(duì)列中后，服務(wù)器會(huì)返回給生產(chǎn)者一個(gè)Confirm，該Confirm包含該消息的ID，這樣生產(chǎn)者就會(huì)知道該消息已被正確分發(fā)。對于持久化消息，只有該消息被持久化后，才會(huì)返回Confirm。

Confirm機(jī)制的最大優(yōu)點(diǎn)在于異步，生產(chǎn)者在發(fā)送消息以后，即可繼續(xù)執(zhí)行其他任務(wù)（也就是異步監(jiān)聽服務(wù)端的ack即可）。而服務(wù)器返回Confirm后，會(huì)觸發(fā)生產(chǎn)者的回調(diào)函數(shù)，生產(chǎn)者在回調(diào)函數(shù)中處理Confirm信息。如果消息服務(wù)器發(fā)生異常，導(dǎo)致該消息丟失，會(huì)返回給生產(chǎn)者一個(gè)nack，表示消息已經(jīng)丟失，這樣生產(chǎn)者就可以通過重發(fā)消息，保證消息不丟失。Confirm機(jī)制在性能上要比事務(wù)優(yōu)越很多。

但是Confirm機(jī)制，無法進(jìn)行回滾，就是一旦服務(wù)器崩潰，生產(chǎn)者無法得到Confirm信息，生產(chǎn)者其實(shí)本身也不知道該消息吃否已經(jīng)被持久化，只有繼續(xù)重發(fā)來保證消息不丟失，但是如果原先已經(jīng)持久化的消息，并不會(huì)被回滾，這樣隊(duì)列中就會(huì)存在兩條相同的消息，系統(tǒng)需要支持去重。

7.消費(fèi)消息

五、使用delivery-mode平衡速度和安全

delivery-mode有兩個(gè)值：1表示非持久化，2表示持久化消息

1.發(fā)送消息到純內(nèi)存隊(duì)列中

delivery-mode = 1

特點(diǎn)：非持久化的消息在服務(wù)宕機(jī)的時(shí)候會(huì)丟失數(shù)據(jù)，但是由于不需要磁盤io，盡可能地降低消息投遞的延遲性，性能較高。

2.發(fā)布消息到支持磁盤存儲(chǔ)的隊(duì)列

delivery-mode = 2

特點(diǎn)：持久化的消息安全性較高，盡管服務(wù)宕機(jī)，數(shù)據(jù)也不會(huì)丟失，但是在投遞消息的過程中需要發(fā)生磁盤io，性能相對純內(nèi)存投遞的方式低，但是盡管是產(chǎn)生了磁盤io，由于日志的記錄方式是直接追加到消息日志文件的末尾，屬于順序io，沒有隨機(jī)io，所以性能還是可以接受的。

大概原理：

• 所有隊(duì)列中的消息都以append的方式寫到一個(gè)文件中，當(dāng)這個(gè)文件的大小超過指定的限制大小后，關(guān)閉這個(gè)文件再創(chuàng)建一個(gè)新的文件供消息的寫入。文件名（*.rdq）從0開始然后依次累加。當(dāng)某個(gè)消息被刪除時(shí)，并不立即從文件中刪除相關(guān)信息，而是做一些記錄，當(dāng)垃圾數(shù)據(jù)達(dá)到一定比例時(shí)，啟動(dòng)垃圾回收處理，將邏輯相鄰的文件中的數(shù)據(jù)合并到一個(gè)文件中。

消息的讀寫及刪除：

• rabbitmq在啟動(dòng)時(shí)會(huì)創(chuàng)建msg_store_persistent,msg_store_transient兩個(gè)進(jìn)程，一個(gè)用于持久消息的存儲(chǔ)，一個(gè)用于內(nèi)存不夠時(shí)，將存儲(chǔ)在內(nèi)存中的非持久化數(shù)據(jù)轉(zhuǎn)存到磁盤中。所有隊(duì)列的消息的寫入和刪除最終都由這兩個(gè)進(jìn)程負(fù)責(zé)處理，而消息的讀取則可能是隊(duì)列本身直接打開文件進(jìn)行讀取，也可能是發(fā)送請求由msg_store_persisteng/msg_store_transient進(jìn)程進(jìn)行處理。

在進(jìn)行消息的存儲(chǔ)時(shí)，rabbitmq會(huì)在ets表中記錄消息在文件中的映射，以及文件的相關(guān)信息。消息讀取時(shí)，根據(jù)消息ID找到該消息所存儲(chǔ)的文件，在文件中的偏移量，然后打開文件進(jìn)行讀取。消息的刪除只是從ets表刪除指定消息的相關(guān)信息，同時(shí)更新消息對應(yīng)存儲(chǔ)的文件的相關(guān)信息（更新文件有效數(shù)據(jù)大?。?/p>

六、消息路由模式

1.fanout模式

fanout類型的Exchange路由規(guī)則非常簡單，它會(huì)把所有發(fā)送到該Exchange的消息路由到所有與它綁定的Queue中。

上圖中，生產(chǎn)者發(fā)送到Exchange的所有消息都會(huì)路由到圖中的兩個(gè)Queue，并最終被兩個(gè)消費(fèi)者（C1與C2）消費(fèi)。

2.direct模式

direct類型的Exchange路由規(guī)則也很簡單，它會(huì)把消息路由到那些binding key與routing key完全匹配的Queue中。如圖，生產(chǎn)者發(fā)送消息的routing key=key1的時(shí)候，只有綁定了key1的queue才能收到信息

3.topic模式

topic類型的Exchange在匹配規(guī)則上進(jìn)行了擴(kuò)展，它與direct類型的Exchage相似，也是將消息路由到binding key與routing key相匹配的Queue中，但這里的匹配規(guī)則有些不同，它約定：

• routing key為一個(gè)句點(diǎn)號“. ”分隔的字符串（我們將被句點(diǎn)號“. ”分隔開的每一段獨(dú)立的字符串稱為一個(gè)單詞），如“image.new.profile”.
• binding key與routing key一樣也是句點(diǎn)號“. ”分隔的字符串
• binding key中可以存在兩種特殊字符“”與“#”，用于做模糊匹配，其中“”用于匹配下一個(gè)據(jù)點(diǎn)前的所有字符，“#”用于匹配所有字符，包括句點(diǎn)（可以是零個(gè)）

如圖，生產(chǎn)者以routing key為image.new.profile發(fā)布消息，這key可以被image.*.profile以及image.#匹配到，所有這兩個(gè)隊(duì)列都可以收到消息。由此可見，topic的路由方式更加靈活。

3.headers模式

headers類型的Exchange不依賴于routing key與binding key的匹配規(guī)則來路由消息，而是根據(jù)發(fā)送的消息內(nèi)容中的headers屬性進(jìn)行匹配。

在綁定Queue與Exchange時(shí)指定一組鍵值對以及x-match參數(shù)，x-match參數(shù)是字符串類型，可以設(shè)置為any或者all。如果設(shè)置為any，意思就是只要匹配到了headers表中的任何一對鍵值即可，all則代表需要全部匹配。

七、rabbitmq流量控制

RabbitMQ可以對內(nèi)存和磁盤使用量設(shè)置閾值，當(dāng)達(dá)到閾值后，生產(chǎn)者將被阻塞（block），直到對應(yīng)項(xiàng)恢復(fù)正常。除了這兩個(gè)閾值，RabbitMQ在正常情況下還用流控（Flow Control）機(jī)制來確保穩(wěn)定性。Erlang進(jìn)程之間并不共享內(nèi)存（binaries類型除外），而是通過消息傳遞來通信，每個(gè)進(jìn)程都有自己的進(jìn)程郵箱。Erlang默認(rèn)沒有對進(jìn)程郵箱大小設(shè)限制，所以當(dāng)有大量消息持續(xù)發(fā)往某個(gè)進(jìn)程時(shí)，會(huì)導(dǎo)致該進(jìn)程郵箱過大，最終內(nèi)存溢出并崩潰。

在RabbitMQ中，如果生產(chǎn)者持續(xù)高速發(fā)送，而消費(fèi)者消費(fèi)速度較低時(shí)，如果沒有流控，很快就會(huì)使內(nèi)部進(jìn)程郵箱大小達(dá)到內(nèi)存閾值，阻塞生產(chǎn)者（得益于block機(jī)制，并不會(huì)崩潰）。然后RabbitMQ會(huì)進(jìn)行page操作，將內(nèi)存中的數(shù)據(jù)持久化到磁盤中。

為了解決該問題，RabbitMQ使用了一種基于信用證的流控機(jī)制。消息處理進(jìn)程有一個(gè)信用組{InitialCredit，MoreCreditAfter}，默認(rèn)值為{200, 50}。消息發(fā)送者進(jìn)程A向接收者進(jìn)程B發(fā)消息，每發(fā)一條消息，Credit數(shù)量減1，直到為0，A被block??；對于接收者B，每接收MoreCreditAfter條消息，會(huì)向A發(fā)送一條消息，給予A MoreCreditAfter個(gè)Credit，當(dāng)A的Credit>0時(shí)，A可以繼續(xù)向B發(fā)送消息。

八、 RabbitMQ 多層消息隊(duì)列

RabbitMQ完全實(shí)現(xiàn)了AMQP協(xié)議，類似于一個(gè)郵箱服務(wù)。Exchange負(fù)責(zé)根據(jù)ExchangeType和RoutingKey將消息投遞到對應(yīng)的消息隊(duì)列中，消息隊(duì)列負(fù)責(zé)在消費(fèi)者獲取消息前暫存消息。

在RabbitMQ中，MessageQueue主要由兩部分組成，一個(gè)為AMQQueue，主要負(fù)責(zé)實(shí)現(xiàn)AMQP協(xié)議的邏輯功能。另外一個(gè)是用來存儲(chǔ)消息的BackingQueue。

為了高效處理入隊(duì)和出隊(duì)的消息、避免不必要的磁盤IO，BackingQueue進(jìn)程為消息設(shè)計(jì)了4種狀態(tài)和5個(gè)內(nèi)部隊(duì)列。

(1) 4種狀態(tài)包括：

alpha，消息的內(nèi)容和索引都在內(nèi)存中；
beta，消息的內(nèi)容在磁盤，索引在內(nèi)存；
gamma，消息的內(nèi)容在磁盤，索引在磁盤和內(nèi)存中都有；
delta，消息的內(nèi)容和索引都在磁盤。

對于持久化消息，RabbitMQ先將消息的內(nèi)容和索引保存在磁盤中，然后才處于上面的某種狀態(tài)（即只可能處于alpha、gamma、delta三種狀態(tài)之一）。

(2) 5個(gè)內(nèi)部隊(duì)列

包括：q1、q2、delta、q3、q4。q1和q4隊(duì)列中只有alpha狀態(tài)的消息；q2和q3包含beta和gamma狀態(tài)的消息；delta隊(duì)列是消息按序存盤后的一種邏輯隊(duì)列，只有delta狀態(tài)的消息。所以delta隊(duì)列并不在內(nèi)存中，其他4個(gè)隊(duì)列則是由erlang queue模塊實(shí)現(xiàn)。

消息從q1入隊(duì)，q4出隊(duì)，在內(nèi)部隊(duì)列中傳遞的過程一般是經(jīng)q1順序到q4。實(shí)際執(zhí)行并非必然如此：開始時(shí)所有隊(duì)列都為空，消息直接進(jìn)入q4（沒有消息堆積時(shí)）；內(nèi)存緊張時(shí)將q4隊(duì)尾部分消息轉(zhuǎn)入q3，進(jìn)而再由q3轉(zhuǎn)入delta，此時(shí)新來的消息將存入q1（有消息堆積時(shí)）。

當(dāng)內(nèi)存緊張時(shí)觸發(fā)paging，paging將大量alpha狀態(tài)的消息轉(zhuǎn)換為beta和gamma；如果內(nèi)存依然緊張，繼續(xù)將beta和gamma狀態(tài)轉(zhuǎn)換為delta狀態(tài)。Paging是一個(gè)持續(xù)過程，涉及到大量消息的多種狀態(tài)轉(zhuǎn)換，所以Paging的開銷較大，嚴(yán)重影響系統(tǒng)性能。

九、高可用隊(duì)列（HA）

在生產(chǎn)環(huán)境下，一般都不會(huì)允許rabbitmq這種消息中間件單點(diǎn)，以免單點(diǎn)故障導(dǎo)致服務(wù)不可用，那么rabbitmq同樣可以集群部署來保證服務(wù)的可用性，在rabbitmq集群中，我們可以定義HA隊(duì)列，可以在web管理平臺(tái)設(shè)置，也可以通過AMQP接口設(shè)置，當(dāng)我們定義某個(gè)HA隊(duì)列的時(shí)候，會(huì)在集群的各個(gè)節(jié)點(diǎn)上都建立該隊(duì)列，發(fā)布消息的時(shí)候，直接發(fā)送至master服務(wù)，當(dāng)master服務(wù)受到消息后，把消息同步至各個(gè)從節(jié)點(diǎn)，假如開啟事務(wù)的情況下，是需要在消息被同步到各個(gè)節(jié)點(diǎn)之后才算完成事務(wù)，所以會(huì)帶來一定的性能損耗，所以還是回到之前說的，性能和安全直接，需要根據(jù)實(shí)際業(yè)務(wù)的需要找到平衡點(diǎn)。

當(dāng)master服務(wù)宕機(jī)之后，其中一個(gè)slaver節(jié)點(diǎn)會(huì)升級為master，消息不會(huì)丟失（因?yàn)橐呀?jīng)完成了事務(wù)的消息都會(huì)在各個(gè)節(jié)點(diǎn)有備份）

ha-隊(duì)列可以跨越集群的每臺(tái)服務(wù)，或者僅使用其中一批獨(dú)立節(jié)點(diǎn)。如果是全部節(jié)點(diǎn)都為副本的時(shí)候，將x-ha-policy參數(shù)設(shè)置為all，否則設(shè)置為nodes，然后在設(shè)置另一個(gè)參數(shù)：x-ha-nodes，該參數(shù)指定ha隊(duì)列所在的節(jié)點(diǎn)列表。

思考下，rabbitmq的集群節(jié)點(diǎn)是不是越多越好？

總結(jié)

以上為個(gè)人經(jīng)驗(yàn)，希望能給大家一個(gè)參考，也希望大家多多支持腳本之家。

最新評論