RabbitMQ

RabbitMQ各組件的功能

Broker ：一個(gè)RabbitMQ實(shí)例就是一個(gè)Broker

Virtual Host ：虛擬主機(jī)。相當(dāng)于MySQL的DataBase，一個(gè)Broker上可以存在多個(gè)vhost，vhost之間相互隔離。每個(gè)vhost都擁有自己的隊(duì)列、交換機(jī)、綁定和權(quán)限機(jī)制。vhost必須在連接時(shí)指定，默認(rèn)的vhost是/。

Exchange ：交換機(jī)，用來接收生產(chǎn)者發(fā)送的消息并將這些消息路由給服務(wù)器中的隊(duì)列。

Queue ：消息隊(duì)列，用來保存消息直到發(fā)送給消費(fèi)者。它是消息的容器。一個(gè)消息可投入一個(gè)或多個(gè)隊(duì)列。

Banding ：綁定關(guān)系，用于消息隊(duì)列和交換機(jī)之間的關(guān)聯(lián)。通過路由鍵（Routing Key）將交換機(jī)和消息隊(duì)列關(guān)聯(lián)起來。

Channel ：管道，一條雙向數(shù)據(jù)流通道。不管是發(fā)布消息、訂閱隊(duì)列還是接收消息，這些動(dòng)作都是通過管道完成。因?yàn)閷τ?a href="http://www.xsypw.cn/v/tag/527/" target="_blank">操作系統(tǒng)來說，建立和銷毀TCP都是非常昂貴的開銷，所以引入了管道的概念，以復(fù)用一條TCP連接。

Connection ：生產(chǎn)者/消費(fèi)者 與broker之間的TCP連接。

Publisher ：消息的生產(chǎn)者。

Consumer ：消息的消費(fèi)者。

Message ：消息，它是由消息頭和消息體組成。消息頭則包括Routing-Key、Priority（優(yōu)先級）等。

RabbitMQ的多種交換機(jī)類型

Exchange 分發(fā)消息給 Queue 時(shí)， Exchange 的類型對應(yīng)不同的分發(fā)策略，有3種類型的 Exchange ：Direct、Fanout、Topic。

Direct：消息中的 Routing Key 如果和 Binding 中的 Routing Key 完全一致， Exchange 就會將消息分發(fā)到對應(yīng)的隊(duì)列中。

Fanout：每個(gè)發(fā)到 Fanout 類型交換機(jī)的消息都會分發(fā)到所有綁定的隊(duì)列上去。Fanout交換機(jī)沒有 Routing Key 。它在三種類型的交換機(jī)中轉(zhuǎn)發(fā)消息是最快的。

Topic：Topic交換機(jī)通過模式匹配分配消息，將 Routing Key 和某個(gè)模式進(jìn)行匹配。它只能識別兩個(gè)通配符："#"和"*"。### 匹配0個(gè)或多個(gè)單詞， * 匹配1個(gè)單詞。

TTL

TTL（Time To Live）：生存時(shí)間。RabbitMQ支持消息的過期時(shí)間，一共2種。

在消息發(fā)送時(shí)進(jìn)行指定。通過配置消息體的 Properties ，可以指定當(dāng)前消息的過期時(shí)間。

在創(chuàng)建Exchange時(shí)指定。從進(jìn)入消息隊(duì)列開始計(jì)算，只要超過了隊(duì)列的超時(shí)時(shí)間配置，那么消息會自動(dòng)清除。

生產(chǎn)者的消息確認(rèn)機(jī)制

Confirm機(jī)制：

消息的確認(rèn)，是指生產(chǎn)者投遞消息后，如果Broker收到消息，則會給我們生產(chǎn)者一個(gè)應(yīng)答。

生產(chǎn)者進(jìn)行接受應(yīng)答，用來確認(rèn)這條消息是否正常的發(fā)送到了Broker，這種方式也是消息的可靠性投遞的核心保障！

如何實(shí)現(xiàn)Confirm確認(rèn)消息？

在channel上開啟確認(rèn)模式：channel.confirmSelect()

在channel上開啟監(jiān)聽：addConfirmListener ，監(jiān)聽成功和失敗的處理結(jié)果，根據(jù)具體的結(jié)果對消息進(jìn)行重新發(fā)送或記錄日志處理等后續(xù)操作。

Return消息機(jī)制：

Return Listener用于處理一些不可路由的消息。

我們的消息生產(chǎn)者，通過指定一個(gè)Exchange和Routing，把消息送達(dá)到某一個(gè)隊(duì)列中去，然后我們的消費(fèi)者監(jiān)聽隊(duì)列進(jìn)行消息的消費(fèi)處理操作。

但是在某些情況下，如果我們在發(fā)送消息的時(shí)候，當(dāng)前的exchange不存在或者指定的路由key路由不到，這個(gè)時(shí)候我們需要監(jiān)聽這種不可達(dá)消息，就需要使用到Returrn Listener。

基礎(chǔ)API中有個(gè)關(guān)鍵的配置項(xiàng) Mandatory ：如果為true，監(jiān)聽器會收到路由不可達(dá)的消息，然后進(jìn)行處理。如果為false，broker端會自動(dòng)刪除該消息。

同樣，通過監(jiān)聽的方式， chennel.addReturnListener(ReturnListener rl) 傳入已經(jīng)重寫過handleReturn方法的ReturnListener。

消費(fèi)端ACK與NACK

消費(fèi)端進(jìn)行消費(fèi)的時(shí)候，如果由于業(yè)務(wù)異常可以進(jìn)行日志的記錄，然后進(jìn)行補(bǔ)償。但是對于服務(wù)器宕機(jī)等嚴(yán)重問題，我們需要手動(dòng)ACK保障消費(fèi)端消費(fèi)成功。

// deliveryTag：消息在mq中的唯一標(biāo)識 // multiple：是否批量(和qos設(shè)置類似的參數(shù)) // requeue：是否需要重回隊(duì)列。或者丟棄或者重回隊(duì)首再次消費(fèi)。 publicvoidbasicNack(longdeliveryTag,booleanmultiple,booleanrequeue)

如上代碼，消息在消費(fèi)端重回隊(duì)列是為了對沒有成功處理消息，把消息重新返回到Broker。一般來說，實(shí)際應(yīng)用中都會關(guān)閉重回隊(duì)列（避免進(jìn)入死循環(huán)），也就是設(shè)置為false。

死信隊(duì)列DLX

死信隊(duì)列（DLX Dead-Letter-Exchange）：當(dāng)消息在一個(gè)隊(duì)列中變成死信之后，它會被重新推送到另一個(gè)隊(duì)列，這個(gè)隊(duì)列就是死信隊(duì)列。

DLX也是一個(gè)正常的Exchange，和一般的Exchange沒有區(qū)別，它能在任何的隊(duì)列上被指定，實(shí)際上就是設(shè)置某個(gè)隊(duì)列的屬性。

當(dāng)這個(gè)隊(duì)列中有死信時(shí)，RabbitMQ就會自動(dòng)的將這個(gè)消息重新發(fā)布到設(shè)置的Exchange上去，進(jìn)而被路由到另一個(gè)隊(duì)列。

RocketMQ

阿里巴巴雙十一官方指定消息產(chǎn)品，支撐阿里巴巴集團(tuán)所有的消息服務(wù)，歷經(jīng)十余年高可用與高可靠的嚴(yán)苛考驗(yàn)，是阿里巴巴交易鏈路的核心產(chǎn)品。

Rocket：火箭的意思。

RocketMQ的核心概念

他有以下核心概念：Broker 、 Topic 、 Tag 、 MessageQueue 、 NameServer 、 Group 、 Offset 、 Producer 以及 Consumer 。

下面來詳細(xì)介紹。

Broker：消息中轉(zhuǎn)角色，負(fù)責(zé)存儲消息，轉(zhuǎn)發(fā)消息。Broker是具體提供業(yè)務(wù)的服務(wù)器，單個(gè)Broker節(jié)點(diǎn)與所有的NameServer節(jié)點(diǎn)保持長連接及心跳，并會定時(shí)將Topic信息注冊到NameServer，順帶一提底層的通信和連接都是基于Netty實(shí)現(xiàn)的。Broker負(fù)責(zé)消息存儲，以Topic為緯度支持輕量級的隊(duì)列，單機(jī)可以支撐上萬隊(duì)列規(guī)模，支持消息推拉模型。官網(wǎng)上有數(shù)據(jù)顯示：具有上億級消息堆積能力，同時(shí)可嚴(yán)格保證消息的有序性。

Topic：主題！它是消息的第一級類型。比如一個(gè)電商系統(tǒng)可以分為：交易消息、物流消息等，一條消息必須有一個(gè) Topic 。Topic與生產(chǎn)者和消費(fèi)者的關(guān)系非常松散，一個(gè) Topic 可以有0個(gè)、1個(gè)、多個(gè)生產(chǎn)者向其發(fā)送消息，一個(gè)生產(chǎn)者也可以同時(shí)向不同的 Topic 發(fā)送消息。一個(gè) Topic 也可以被 0個(gè)、1個(gè)、多個(gè)消費(fèi)者訂閱。

Tag：標(biāo)簽！可以看作子主題，它是消息的第二級類型，用于為用戶提供額外的靈活性。使用標(biāo)簽，同一業(yè)務(wù)模塊不同目的的消息就可以用相同Topic而不同的Tag來標(biāo)識。比如交易消息又可以分為：交易創(chuàng)建消息、交易完成消息等，一條消息可以沒有Tag。標(biāo)簽有助于保持您的代碼干凈和連貫，并且還可以為RabbitMQ提供的查詢系統(tǒng)提供幫助。

MessageQueue：一個(gè)Topic下可以設(shè)置多個(gè)消息隊(duì)列，發(fā)送消息時(shí)執(zhí)行該消息的Topic，RocketMQ會輪詢該Topic下的所有隊(duì)列將消息發(fā)出去。消息的物理管理單位。一個(gè)Topic下可以有多個(gè)Queue，Queue的引入使得消息的存儲可以分布式集群化，具有了水平擴(kuò)展能力。

NameServer：類似Kafka中的ZooKeeper，但NameServer集群之間是沒有通信的，相對ZK來說更加輕量。它主要負(fù)責(zé)對于源數(shù)據(jù)的管理，包括了對于Topic和路由信息的管理。每個(gè)Broker在啟動(dòng)的時(shí)候會到NameServer注冊，Producer在發(fā)送消息前會根據(jù)Topic去NameServer獲取對應(yīng)Broker的路由信息，Consumer也會定時(shí)獲取 Topic 的路由信息。

Producer：生產(chǎn)者，支持三種方式發(fā)送消息：同步、異步和單向單向發(fā)送 ：消息發(fā)出去后，可以繼續(xù)發(fā)送下一條消息或執(zhí)行業(yè)務(wù)代碼，不等待服務(wù)器回應(yīng)，且沒有回調(diào)函數(shù)。異步發(fā)送 ：消息發(fā)出去后，可以繼續(xù)發(fā)送下一條消息或執(zhí)行業(yè)務(wù)代碼，不等待服務(wù)器回應(yīng)，有回調(diào)函數(shù)。同步發(fā)送 ：消息發(fā)出去后，等待服務(wù)器響應(yīng)成功或失敗，才能繼續(xù)后面的操作。

Consumer：消費(fèi)者，支持 PUSH 和 PULL 兩種消費(fèi)模式，支持集群消費(fèi)和廣播消費(fèi)集群消費(fèi) ：該模式下一個(gè)消費(fèi)者集群共同消費(fèi)一個(gè)主題的多個(gè)隊(duì)列，一個(gè)隊(duì)列只會被一個(gè)消費(fèi)者消費(fèi)，如果某個(gè)消費(fèi)者掛掉，分組內(nèi)其它消費(fèi)者會接替掛掉的消費(fèi)者繼續(xù)消費(fèi)。廣播消費(fèi) ：會發(fā)給消費(fèi)者組中的每一個(gè)消費(fèi)者進(jìn)行消費(fèi)。相當(dāng)于RabbitMQ的發(fā)布訂閱模式。

Group：分組，一個(gè)組可以訂閱多個(gè)Topic。分為ProducerGroup，ConsumerGroup，代表某一類的生產(chǎn)者和消費(fèi)者，一般來說同一個(gè)服務(wù)可以作為Group，同一個(gè)Group一般來說發(fā)送和消費(fèi)的消息都是一樣的

Offset：在RocketMQ中，所有消息隊(duì)列都是持久化，長度無限的數(shù)據(jù)結(jié)構(gòu)，所謂長度無限是指隊(duì)列中的每個(gè)存儲單元都是定長，訪問其中的存儲單元使用Offset來訪問，Offset為Java Long類型，64位，理論上在 100年內(nèi)不會溢出，所以認(rèn)為是長度無限。也可以認(rèn)為Message Queue是一個(gè)長度無限的數(shù)組，Offset就是下標(biāo)。

延時(shí)消息

開源版的RocketMQ不支持任意時(shí)間精度，僅支持特定的level，例如定時(shí)5s，10s，1min等。其中，level=0級表示不延時(shí)，level=1表示1級延時(shí)，level=2表示2級延時(shí)，以此類推。

延時(shí)等級如下：

messageDelayLevel=1s5s10s30s1m2m3m4m5m6m7m8m9m10m20m30m1h2h

順序消息

消息有序指的是可以按照消息的發(fā)送順序來消費(fèi)（FIFO）。RocketMQ可以嚴(yán)格的保證消息有序，可以分為 分區(qū)有序 或者 全局有序 。

事務(wù)消息

消息隊(duì)列MQ提供類似X/Open XA的分布式事務(wù)功能，通過消息隊(duì)列MQ事務(wù)消息能達(dá)到分布式事務(wù)的最終一致。上圖說明了事務(wù)消息的大致流程：正常事務(wù)消息的發(fā)送和提交、事務(wù)消息的補(bǔ)償流程。另外，搜索公眾號前端技術(shù)精選后臺回復(fù)“前端”，獲取一份驚喜禮包。

事務(wù)消息發(fā)送及提交：

發(fā)送half消息

服務(wù)端響應(yīng)消息寫入結(jié)果

根據(jù)發(fā)送結(jié)果執(zhí)行本地事務(wù)（如果寫入失敗，此時(shí)half消息對業(yè)務(wù)不可見，本地邏輯不執(zhí)行）；

根據(jù)本地事務(wù)狀態(tài)執(zhí)行Commit或Rollback（Commit操作生成消息索引，消息對消費(fèi)者可見）。

事務(wù)消息的補(bǔ)償流程：

對沒有Commit/Rollback的事務(wù)消息（pending狀態(tài)的消息），從服務(wù)端發(fā)起一次“回查”；

Producer收到回查消息，檢查回查消息對應(yīng)的本地事務(wù)的狀態(tài)。

根據(jù)本地事務(wù)狀態(tài)，重新Commit或RollBack

其中，補(bǔ)償階段用于解決消息Commit或Rollback發(fā)生超時(shí)或者失敗的情況。

事務(wù)消息狀態(tài)：

事務(wù)消息共有三種狀態(tài)：提交狀態(tài)、回滾狀態(tài)、中間狀態(tài)：

TransactionStatus.CommitTransaction：提交事務(wù)，它允許消費(fèi)者消費(fèi)此消息。

TransactionStatus.RollbackTransaction：回滾事務(wù)，它代表該消息將被刪除，不允許被消費(fèi)。

TransactionStatus.Unkonwn：中間狀態(tài)，它代表需要檢查消息隊(duì)列來確定消息狀態(tài)。

RocketMQ的高可用機(jī)制

RocketMQ是天生支持分布式的，可以配置主從以及水平擴(kuò)展。

Master角色的Broker支持讀和寫，Slave角色的Broker僅支持讀，也就是 Producer只能和Master角色的Broker連接寫入消息；Consumer可以連接 Master角色的Broker，也可以連接Slave角色的Broker來讀取消息。

消息消費(fèi)的高可用（主從）：

在Consumer的配置文件中，并不需要設(shè)置是從Master讀還是從Slave讀，當(dāng)Master不可用或者繁忙的時(shí)候，Consumer會被自動(dòng)切換到從Slave讀。有了自動(dòng)切換Consumer這種機(jī)制，當(dāng)一個(gè)Master角色的機(jī)器出現(xiàn)故障后，Consumer仍然可以從Slave讀取消息，不影響Consumer程序。這就達(dá)到了消費(fèi)端的高可用性。RocketMQ目前還不支持把Slave自動(dòng)轉(zhuǎn)成Master，如果機(jī)器資源不足，需要把Slave轉(zhuǎn)成Master，則要手動(dòng)停止Slave角色的Broker，更改配置文件，用新的配置文件啟動(dòng)Broker。

消息發(fā)送高可用（配置多個(gè)主節(jié)點(diǎn)）：

在創(chuàng)建Topic的時(shí)候，把Topic的多個(gè)Message Queue創(chuàng)建在多個(gè)Broker組上（相同Broker名稱，不同 brokerId的機(jī)器組成一個(gè)Broker組），這樣當(dāng)一個(gè)Broker組的Master不可用后，其他組的Master仍然可用，Producer仍然可以發(fā)送消息。

主從復(fù)制：

如果一個(gè)Broker組有Master和Slave，消息需要從Master復(fù)制到Slave 上，有同步和異步兩種復(fù)制方式。

同步復(fù)制：同步復(fù)制方式是等Master和Slave均寫成功后才反饋給客戶端寫成功狀態(tài)。如果Master出故障， Slave上有全部的備份數(shù)據(jù)，容易恢復(fù)同步復(fù)制會增大數(shù)據(jù)寫入延遲，降低系統(tǒng)吞吐量。

異步復(fù)制：異步復(fù)制方式是只要Master寫成功 即可反饋給客戶端寫成功狀態(tài)。在異步復(fù)制方式下，系統(tǒng)擁有較低的延遲和較高的吞吐量，但是如果Master出了故障，有些數(shù)據(jù)因?yàn)闆]有被寫 入Slave，有可能會丟失

通常情況下，應(yīng)該把Master和Save配置成同步刷盤方式，主從之間配置成異步的復(fù)制方式，這樣即使有一臺機(jī)器出故障，仍然能保證數(shù)據(jù)不丟，是個(gè)不錯(cuò)的選擇。

負(fù)載均衡

Producer負(fù)載均衡：

Producer端，每個(gè)實(shí)例在發(fā)消息的時(shí)候，默認(rèn)會輪詢所有的Message Queue發(fā)送，以達(dá)到讓消息平均落在不同的Queue上。而由于Queue可以散落在不同的Broker，所以消息就發(fā)送到不同的Broker下，如下圖：

Consumer負(fù)載均衡：

如果Consumer實(shí)例的數(shù)量比Message Queue的總數(shù)量還多的話，多出來的Consumer實(shí)例將無法分到Queue，也就無法消費(fèi)到消息，也就無法起到分?jǐn)傌?fù)載的作用了。所以需要控制讓Queue的總數(shù)量大于等于Consumer的數(shù)量。

消費(fèi)者的集群模式：啟動(dòng)多個(gè)消費(fèi)者就可以保證消費(fèi)者的負(fù)載均衡（均攤隊(duì)列）

默認(rèn)使用的是均攤隊(duì)列：會按照Queue的數(shù)量和實(shí)例的數(shù)量平均分配Queue給每個(gè)實(shí)例，這樣每個(gè)消費(fèi)者可以均攤消費(fèi)的隊(duì)列，如下圖所示6個(gè)隊(duì)列和三個(gè)生產(chǎn)者。

另外一種平均的算法環(huán)狀輪流分Queue的形式，每個(gè)消費(fèi)者，均攤不同主節(jié)點(diǎn)的一個(gè)消息隊(duì)列，如下圖所示：

對于廣播模式并不是負(fù)載均衡的，要求一條消息需要投遞到一個(gè)消費(fèi)組下面所有的消費(fèi)者實(shí)例，所以也就沒有消息被分?jǐn)傁M(fèi)的說法。

死信隊(duì)列

當(dāng)一條消息消費(fèi)失敗，RocketMQ就會自動(dòng)進(jìn)行消息重試。而如果消息超過最大重試次數(shù)，RocketMQ就會認(rèn)為這個(gè)消息有問題。但是此時(shí)，RocketMQ不會立刻將這個(gè)有問題的消息丟棄，而會將其發(fā)送到這個(gè)消費(fèi)者組對應(yīng)的一種特殊隊(duì)列：死信隊(duì)列。死信隊(duì)列的名稱是 %DLQ%+ConsumGroup 。

死信隊(duì)列具有以下特性：

一個(gè)死信隊(duì)列對應(yīng)一個(gè)Group ID， 而不是對應(yīng)單個(gè)消費(fèi)者實(shí)例。

如果一個(gè)Group ID未產(chǎn)生死信消息，消息隊(duì)列RocketMQ不會為其創(chuàng)建相應(yīng)的死信隊(duì)列。

一個(gè)死信隊(duì)列包含了對應(yīng)Group ID產(chǎn)生的所有死信消息，不論該消息屬于哪個(gè)Topic

Kafka

Kafka是一個(gè)分布式、支持分區(qū)的、多副本的，基于ZooKeeper協(xié)調(diào)的分布式消息系統(tǒng)。

它最大的特性就是可以實(shí)時(shí)的處理大量數(shù)據(jù)以滿足各種需求場景：比如基于Hadoop的批處理系統(tǒng)、低延遲的實(shí)時(shí)系統(tǒng)、Storm/Spark流式處理引擎，Web/Nginx日志、訪問日志，消息服務(wù)等等，用Scala語言編寫。屬于Apache基金會的頂級開源項(xiàng)目。

先看一下Kafka的架構(gòu)圖 ：

Kafka的核心概念

在Kafka中有幾個(gè)核心概念：

Broker：消息中間件處理節(jié)點(diǎn)，一個(gè)Kafka節(jié)點(diǎn)就是一個(gè)Broker，一個(gè)或者多個(gè)Broker可以組成一個(gè)Kafka集群

Topic：Kafka根據(jù)topic對消息進(jìn)行歸類，發(fā)布到Kafka集群的每條消息都需要指定一個(gè)topic

Producer：消息生產(chǎn)者，向Broker發(fā)送消息的客戶端

Consumer：消息消費(fèi)者，從Broker讀取消息的客戶端

ConsumerGroup：每個(gè)Consumer屬于一個(gè)特定的ConsumerGroup，一條消息可以被多個(gè)不同的ConsumerGroup消費(fèi)，但是一個(gè)ConsumerGroup中只能有一個(gè)Consumer能夠消費(fèi)該消息

Partition：物理上的概念，一個(gè)topic可以分為多個(gè)partition，每個(gè)partition內(nèi)部消息是有序的

Leader：每個(gè)Partition有多個(gè)副本，其中有且僅有一個(gè)作為Leader，Leader是負(fù)責(zé)數(shù)據(jù)讀寫的Partition。

Follower：Follower跟隨Leader，所有寫請求都通過Leader路由，數(shù)據(jù)變更會廣播給所有Follower，F(xiàn)ollower與Leader保持?jǐn)?shù)據(jù)同步。如果Leader失效，則從Follower中選舉出一個(gè)新的Leader。當(dāng)Follower與Leader掛掉、卡住或者同步太慢，Leader會把這個(gè)Follower從 ISR列表 中刪除，重新創(chuàng)建一個(gè)Follower。

Offset：偏移量。Kafka的存儲文件都是按照offset.kafka來命名，用Offset做名字的好處是方便查找。例如你想找位于2049的位置，只要找到2048.kafka的文件即可。

可以這么來理解Topic，Partition和Broker：

一個(gè)Topic，代表邏輯上的一個(gè)業(yè)務(wù)數(shù)據(jù)集，比如訂單相關(guān)操作消息放入訂單Topic，用戶相關(guān)操作消息放入用戶Topic，對于大型網(wǎng)站來說，后端數(shù)據(jù)都是海量的，訂單消息很可能是非常巨量的，比如有幾百個(gè)G甚至達(dá)到TB級別，如果把這么多數(shù)據(jù)都放在一臺機(jī)器上可定會有容量限制問題，那么就可以在Topic內(nèi)部劃分多個(gè)Partition來分片存儲數(shù)據(jù)，不同的Partition可以位于不同的機(jī)器上，相當(dāng)于分布式存儲。每臺機(jī)器上都運(yùn)行一個(gè)Kafka的進(jìn)程Broker。

Kafka核心總控制器Controller

在Kafka集群中會有一個(gè)或者多個(gè)Broker，其中有一個(gè)Broker會被選舉為控制器（Kafka Controller），可以理解為 Broker-Leader ，它負(fù)責(zé)管理整個(gè) 集群中所有分區(qū)和副本的狀態(tài)。

Partition-Leader

Controller選舉機(jī)制

在Kafka集群啟動(dòng)的時(shí)候，選舉的過程是集群中每個(gè)Broker都會嘗試在ZooKeeper上創(chuàng)建一個(gè) /controller臨時(shí)節(jié)點(diǎn)，ZooKeeper會保證有且僅有一個(gè)Broker能創(chuàng)建成功，這個(gè)Broker就會成為集群的總控器Controller。

當(dāng)這個(gè)Controller角色的Broker宕機(jī)了，此時(shí)ZooKeeper臨時(shí)節(jié)點(diǎn)會消失，集群里其他Broker會一直監(jiān)聽這個(gè)臨時(shí)節(jié) 點(diǎn)，發(fā)現(xiàn)臨時(shí)節(jié)點(diǎn)消失了，就競爭再次創(chuàng)建臨時(shí)節(jié)點(diǎn)，就是我們上面說的選舉機(jī)制，ZooKeeper又會保證有一個(gè)Broker成為新的Controller。具備控制器身份的Broker需要比其他普通的Broker多一份職責(zé)，具體細(xì)節(jié)如下：

監(jiān)聽Broker相關(guān)的變化。為ZooKeeper中的/brokers/ids/節(jié)點(diǎn)添加BrokerChangeListener，用來處理Broker增減的變化。

監(jiān)聽Topic相關(guān)的變化。為ZooKeeper中的/brokers/topics節(jié)點(diǎn)添加TopicChangeListener，用來處理Topic增減的變化；為ZooKeeper中的/admin/delete_topics節(jié)點(diǎn)添加TopicDeletionListener，用來處理刪除Topic的動(dòng)作。

從ZooKeeper中讀取獲取當(dāng)前所有與Topic、Partition以及Broker有關(guān)的信息并進(jìn)行相應(yīng)的管理 。對于所有Topic所對應(yīng)的ZooKeeper中的/brokers/topics/節(jié)點(diǎn)添加PartitionModificationsListener，用來監(jiān)聽Topic中的分區(qū)分配變化。

更新集群的元數(shù)據(jù)信息，同步到其他普通的Broker節(jié)點(diǎn)中

Partition副本選舉Leader機(jī)制

Controller感知到分區(qū)Leader所在的Broker掛了，Controller會從ISR列表（參數(shù) unclean.leader.election.enable=false的前提下）里挑第一個(gè)Broker作為Leader（第一個(gè)Broker最先放進(jìn)ISR列表，可能是同步數(shù)據(jù)最多的副本），如果參數(shù)unclean.leader.election.enable為true，代表在ISR列表里所有副本都掛了的時(shí)候可以在ISR列表以外的副本中選Leader，這種設(shè)置，可以提高可用性，但是選出的新Leader有可能數(shù)據(jù)少很多。副本進(jìn)入ISR列表有兩個(gè)條件：

副本節(jié)點(diǎn)不能產(chǎn)生分區(qū)，必須能與ZooKeeper保持會話以及跟Leader副本網(wǎng)絡(luò)連通

副本能復(fù)制Leader上的所有寫操作，并且不能落后太多。（與Leader副本同步滯后的副本，是由replica.lag.time.max.ms配置決定的，超過這個(gè)時(shí)間都沒有跟Leader同步過的一次的副本會被移出ISR列表）

消費(fèi)者消費(fèi)消息的Offset記錄機(jī)制

每個(gè)Consumer會定期將自己消費(fèi)分區(qū)的Offset提交給Kafka內(nèi)部Topic：consumer_offsets，提交過去的時(shí)候，key是consumerGroupId+topic+分區(qū)號，value就是當(dāng)前Offset的值，Kafka會定期清理Topic里的消息，最后就保留最新的那條數(shù)據(jù)。

因?yàn)開_consumer_offsets可能會接收高并發(fā)的請求，Kafka默認(rèn)給其分配50個(gè)分區(qū)（可以通過 offsets.topic.num.partitions設(shè)置），這樣可以通過加機(jī)器的方式抗大并發(fā)。

消費(fèi)者Rebalance機(jī)制

Rebalance就是說 如果消費(fèi)組里的消費(fèi)者數(shù)量有變化或消費(fèi)的分區(qū)數(shù)有變化，Kafka會重新分配消費(fèi)者與消費(fèi)分區(qū)的關(guān)系 。比如consumer group中某個(gè)消費(fèi)者掛了，此時(shí)會自動(dòng)把分配給他的分區(qū)交給其他的消費(fèi)者，如果他又重啟了，那么又會把一些分區(qū)重新交還給他。

注意：Rebalance只針對subscribe這種不指定分區(qū)消費(fèi)的情況，如果通過assign這種消費(fèi)方式指定了分區(qū)，Kafka不會進(jìn)行Rebalance。

如下情況可能會觸發(fā)消費(fèi)者Rebalance：

消費(fèi)組里的Consumer增加或減少了

動(dòng)態(tài)給Topic增加了分區(qū)

消費(fèi)組訂閱了更多的Topic

Rebalance過程中，消費(fèi)者無法從Kafka消費(fèi)消息，這對Kafka的TPS會有影響，如果Kafka集群內(nèi)節(jié)點(diǎn)較多，比如數(shù)百 個(gè)，那重平衡可能會耗時(shí)極多，所以應(yīng)盡量避免在系統(tǒng)高峰期的重平衡發(fā)生。

Rebalance過程如下

當(dāng)有消費(fèi)者加入消費(fèi)組時(shí)，消費(fèi)者、消費(fèi)組及組協(xié)調(diào)器之間會經(jīng)歷以下幾個(gè)階段：

第一階段：選擇組協(xié)調(diào)器

組協(xié)調(diào)器GroupCoordinator：每個(gè)consumer group都會選擇一個(gè)Broker作為自己的組協(xié)調(diào)器coordinator，負(fù)責(zé)監(jiān)控這個(gè)消費(fèi)組里的所有消費(fèi)者的心跳，以及判斷是否宕機(jī)，然后開啟消費(fèi)者Rebalance。consumer group中的每個(gè)consumer啟動(dòng)時(shí)會向Kafka集群中的某個(gè)節(jié)點(diǎn)發(fā)送FindCoordinatorRequest請求來查找對應(yīng)的組協(xié)調(diào)器GroupCoordinator，并跟其建立網(wǎng)絡(luò)連接。組協(xié)調(diào)器選擇方式：通過如下公式可以選出consumer消費(fèi)的Offset要提交到__consumer_offsets的哪個(gè)分區(qū)，這個(gè)分區(qū)Leader對應(yīng)的Broker就是這個(gè)consumer group的coordinator公式：

hash(consumer group id) % 對應(yīng)主題的分區(qū)數(shù)

第二階段：加入消費(fèi)組JOIN GROUP

在成功找到消費(fèi)組所對應(yīng)的GroupCoordinator之后就進(jìn)入加入消費(fèi)組的階段，在此階段的消費(fèi)者會向GroupCoordinator發(fā)送JoinGroupRequest請求，并處理響應(yīng)。然后GroupCoordinator從一個(gè)consumer group中選擇第一個(gè)加入group的consumer作為Leader（消費(fèi)組協(xié)調(diào)器），把consumer group情況發(fā)送給這個(gè)Leader，接著這個(gè)Leader會負(fù)責(zé)制定分區(qū)方案。另外，搜索公眾號Python人工智能技術(shù)后臺回復(fù)“名著”，獲取一份驚喜禮包。

第三階段（SYNC GROUP）

consumer leader通過給GroupCoordinator發(fā)送SyncGroupRequest，接著GroupCoordinator就把分區(qū)方案下發(fā)給各個(gè)consumer，他們會根據(jù)指定分區(qū)的Leader Broker進(jìn)行網(wǎng)絡(luò)連接以及消息消費(fèi)。

消費(fèi)者Rebalance分區(qū)分配策略

主要有三種Rebalance的策略：range 、 round-robin 、 sticky 。默認(rèn)情況為range分配策略。

假設(shè)一個(gè)主題有10個(gè)分區(qū)（0-9），現(xiàn)在有三個(gè)consumer消費(fèi)：

range策略：按照分區(qū)序號排序分配 ，假設(shè)n＝分區(qū)數(shù)／消費(fèi)者數(shù)量 = 3， m＝分區(qū)數(shù)%消費(fèi)者數(shù)量 = 1，那么前 m 個(gè)消 費(fèi)者每個(gè)分配 n+1 個(gè)分區(qū)，后面的（消費(fèi)者數(shù)量－m ）個(gè)消費(fèi)者每個(gè)分配 n 個(gè)分區(qū)。比如分區(qū)0~ 3給一個(gè)consumer，分區(qū)4~ 6給一個(gè)consumer，分區(qū)7~9給一個(gè)consumer。

round-robin策略：輪詢分配 ，比如分區(qū)0、3、6、9給一個(gè)consumer，分區(qū)1、4、7給一個(gè)consumer，分區(qū)2、5、 8給一個(gè)consumer

sticky策略：初始時(shí)分配策略與round-robin類似，但是在rebalance的時(shí)候，需要保證如下兩個(gè)原則：

分區(qū)的分配要盡可能均勻 。

分區(qū)的分配盡可能與上次分配的保持相同。

當(dāng)兩者發(fā)生沖突時(shí)，第一個(gè)目標(biāo)優(yōu)先于第二個(gè)目標(biāo) 。這樣可以最大程度維持原來的分區(qū)分配的策略。比如對于第一種range情況的分配，如果第三個(gè)consumer掛了，那么重新用sticky策略分配的結(jié)果如下：consumer1除了原有的0~ 3，會再分配一個(gè)7 consumer2除了原有的4~ 6，會再分配8和9。

Producer發(fā)布消息機(jī)制剖析

1、寫入方式

producer采用push模式將消息發(fā)布到broker，每條消息都被append到patition中，屬于順序?qū)懘疟P（順序?qū)懘疟P 比 隨機(jī)寫 效率要高，保障 kafka 吞吐率）。

牛逼啊！接私活必備的 N 個(gè)開源項(xiàng)目！趕快收藏吧

2、消息路由

producer發(fā)送消息到broker時(shí)，會根據(jù)分區(qū)算法選擇將其存儲到哪一個(gè)partition。其路由機(jī)制為：

hash(key)%分區(qū)數(shù)

3、寫入流程

producer先從ZooKeeper的 "/brokers/…/state" 節(jié)點(diǎn)找到該partition的leader

producer將消息發(fā)送給該leader

leader將消息寫入本地log

followers從leader pull消息，寫入本地log后向leader發(fā)送ACK

leader收到所有ISR中的replica的ACK后，增加HW（high watermark，最后commit的offset）并向producer發(fā)送ACK

HW與LEO

HW俗稱高水位 ，HighWatermark的縮寫，取一個(gè)partition對應(yīng)的ISR中最小的LEO（log-end-offset）作為HW， consumer最多只能消費(fèi)到HW所在的位置。另外每個(gè)replica都有HW,leader和follower各自負(fù)責(zé)更新自己的HW的狀 態(tài)。對于leader新寫入的消息，consumer不能立刻消費(fèi)，leader會等待該消息被所有ISR中的replicas同步后更新HW， 此時(shí)消息才能被consumer消費(fèi)。這樣就保證了如果leader所在的broker失效，該消息仍然可以從新選舉的leader中獲取。對于來自內(nèi)部broker的讀取請求，沒有HW的限制。

日志分段存儲

Kafka一個(gè)分區(qū)的消息數(shù)據(jù)對應(yīng)存儲在一個(gè)文件夾下，以topic名稱+分區(qū)號命名，消息在分區(qū)內(nèi)是分段存儲的， 每個(gè)段的消息都存儲在不一樣的log文件里，Kafka規(guī)定了一個(gè)段位的log文件最大為1G，做這個(gè)限制目的是為了方便把log文件加載到內(nèi)存去操作：

1###部分消息的offset索引文件，kafka每次往分區(qū)發(fā)4K(可配置)消息就會記錄一條當(dāng)前消息的offset到index文件， 2###如果要定位消息的offset會先在這個(gè)文件里快速定位，再去log文件里找具體消息 300000000000000000000.index 4###消息存儲文件，主要存offset和消息體 500000000000000000000.log 6###消息的發(fā)送時(shí)間索引文件，kafka每次往分區(qū)發(fā)4K(可配置)消息就會記錄一條當(dāng)前消息的發(fā)送時(shí)間戳與對應(yīng)的offset到timeindex文件， 7###如果需要按照時(shí)間來定位消息的offset，會先在這個(gè)文件里查找 800000000000000000000.timeindex 9 1000000000000005367851.index 1100000000000005367851.log 1200000000000005367851.timeindex 13 1400000000000009936472.index 1500000000000009936472.log 1600000000000009936472.timeindex

這個(gè)9936472之類的數(shù)字，就是代表了這個(gè)日志段文件里包含的起始 Offset，也就說明這個(gè)分區(qū)里至少都寫入了接近1000萬條數(shù)據(jù)了。Kafka Broker有一個(gè)參數(shù)，log.segment.bytes，限定了每個(gè)日志段文件的大小，最大就是1GB。一個(gè)日志段文件滿了，就自動(dòng)開一個(gè)新的日志段文件來寫入，避免單個(gè)文件過大，影響文件的讀寫性能，這個(gè)過程叫做log rolling，正在被寫入的那個(gè)日志段文件，叫做active log segment。

最后附一張ZooKeeper節(jié)點(diǎn)數(shù)據(jù)圖

MQ帶來的一些問題、及解決方案

如何保證順序消費(fèi)？

RabbitMQ：一個(gè)Queue對應(yīng)一個(gè)Consumer即可解決。

RocketMQhash(key)%隊(duì)列數(shù)

Kafka：hash(key)%分區(qū)數(shù)

如何實(shí)現(xiàn)延遲消費(fèi)？

RabbitMQ：兩種方案 死信隊(duì)列 + TTL引入RabbitMQ的延遲插件

RocketMQ：天生支持延時(shí)消息。

Kafka：步驟如下 專門為要延遲的消息創(chuàng)建一個(gè)Topic新建一個(gè)消費(fèi)者去消費(fèi)這個(gè)Topic消息持久化再開一個(gè)線程定時(shí)去拉取持久化的消息，放入實(shí)際要消費(fèi)的Topic實(shí)際消費(fèi)的消費(fèi)者從實(shí)際要消費(fèi)的Topic拉取消息。

如何保證消息的可靠性投遞

RabbitMQ：

Broker-->消費(fèi)者：手動(dòng)ACK

生產(chǎn)者-->Broker：兩種方案

數(shù)據(jù)庫持久化：

1.將業(yè)務(wù)訂單數(shù)據(jù)和生成的Message進(jìn)行持久化操作（一般情況下插入數(shù)據(jù)庫，這里如果分庫的話可能涉及到分布式事務(wù)） 2.將Message發(fā)送到Broker服務(wù)器中 3.通過RabbitMQ的Confirm機(jī)制，在producer端，監(jiān)聽服務(wù)器是否ACK。 4.如果ACK了，就將Message這條數(shù)據(jù)狀態(tài)更新為已發(fā)送。如果失敗，修改為失敗狀態(tài)。 5.分布式定時(shí)任務(wù)查詢數(shù)據(jù)庫3分鐘（這個(gè)具體時(shí)間應(yīng)該根據(jù)的時(shí)效性來定）之前的發(fā)送失敗的消息 6.重新發(fā)送消息，記錄發(fā)送次數(shù) 7.如果發(fā)送次數(shù)過多仍然失敗，那么就需要人工排查之類的操作。

優(yōu)點(diǎn)：能夠保證消息百分百不丟失。

缺點(diǎn)：第一步會涉及到分布式事務(wù)問題。

消息的延遲投遞：

流程圖中，顏色不同的代表不同的message 1.將業(yè)務(wù)訂單持久化 2.發(fā)送一條Message到broker(稱之為主Message)，再發(fā)送相同的一條到不同的隊(duì)列或者交換機(jī)(這條稱為確認(rèn)Message)中。 3.主Message由實(shí)際業(yè)務(wù)處理端消費(fèi)后，生成一條響應(yīng)Message。之前的確認(rèn)Message由MessageService應(yīng)用處理入庫。 4~6.實(shí)際業(yè)務(wù)處理端發(fā)送的確認(rèn)Message由MessageService接收后，將原Message狀態(tài)修改。 7.如果該條Message沒有被確認(rèn)，則通過rpc調(diào)用重新由producer進(jìn)行全過程。

優(yōu)點(diǎn)：相對于持久化方案來說響應(yīng)速度有所提升

缺點(diǎn)：系統(tǒng)復(fù)雜性有點(diǎn)高，萬一消息都失敗了，消息存在丟失情況，仍需Confirm機(jī)制做補(bǔ)償。擴(kuò)展：接私活兒

RocketMQ

生產(chǎn)者弄丟數(shù)據(jù)：

Producer在把Message發(fā)送Broker的過程中，因?yàn)榫W(wǎng)絡(luò)問題等發(fā)生丟失，或者M(jìn)essage到了Broker，但是出了問題，沒有保存下來。針對這個(gè)問題，RocketMQ對Producer發(fā)送消息設(shè)置了3種方式：

同步發(fā)送 異步發(fā)送 單向發(fā)送

Broker弄丟數(shù)據(jù)：

Broker接收到Message暫存到內(nèi)存，Consumer還沒來得及消費(fèi)，Broker掛掉了。

可以通過 持久化 設(shè)置去解決：

創(chuàng)建Queue的時(shí)候設(shè)置持久化，保證Broker持久化Queue的元數(shù)據(jù)，但是不會持久化Queue里面的消息

將Message的deliveryMode設(shè)置為2，可以將消息持久化到磁盤，這樣只有Message支持化到磁盤之后才會發(fā)送通知Producer ack

這兩步過后，即使Broker掛了，Producer肯定收不到ack的，就可以進(jìn)行重發(fā)。

消費(fèi)者弄丟數(shù)據(jù)：

Consumer有消費(fèi)到Message，但是內(nèi)部出現(xiàn)問題，Message還沒處理，Broker以為Consumer處理完了，只會把后續(xù)的消息發(fā)送。這時(shí)候，就要 關(guān)閉autoack，消息處理過后，進(jìn)行手動(dòng)ack , 多次消費(fèi)失敗的消息，會進(jìn)入 死信隊(duì)列 ，這時(shí)候需要人工干預(yù)。

Kafka

生產(chǎn)者弄丟數(shù)據(jù)

設(shè)置了 acks=all ，一定不會丟，要求是，你的 leader 接收到消息，所有的 follower 都同步到了消息之后，才認(rèn)為本次寫成功了。如果沒滿足這個(gè)條件，生產(chǎn)者會自動(dòng)不斷的重試，重試無限次。

Broker弄丟數(shù)據(jù)

Kafka 某個(gè) broker 宕機(jī)，然后重新選舉 partition 的 leader。大家想想，要是此時(shí)其他的 follower 剛好還有些數(shù)據(jù)沒有同步，結(jié)果此時(shí) leader 掛了，然后選舉某個(gè) follower 成 leader 之后，不就少了一些數(shù)據(jù)？這就丟了一些數(shù)據(jù)啊。

此時(shí)一般是要求起碼設(shè)置如下 4 個(gè)參數(shù)：

replication.factor min.insync.replicas acks=all retries=MAX

我們生產(chǎn)環(huán)境就是按照上述要求配置的，這樣配置之后，至少在 Kafka broker 端就可以保證在 leader 所在 broker 發(fā)生故障，進(jìn)行 leader 切換時(shí)，數(shù)據(jù)不會丟失。

消費(fèi)者弄丟數(shù)據(jù)

你消費(fèi)到了這個(gè)消息，然后消費(fèi)者那邊自動(dòng)提交了 offset，讓 Kafka 以為你已經(jīng)消費(fèi)好了這個(gè)消息，但其實(shí)你才剛準(zhǔn)備處理這個(gè)消息，你還沒處理，你自己就掛了，此時(shí)這條消息就丟咯。

這不是跟 RabbitMQ 差不多嗎，大家都知道 Kafka 會自動(dòng)提交 offset，那么只要 關(guān)閉自動(dòng)提交 offset，在處理完之后自己手動(dòng)提交 offset，就可以保證數(shù)據(jù)不會丟。但是此時(shí)確實(shí)還是可能會有重復(fù)消費(fèi)，比如你剛處理完，還沒提交 offset，結(jié)果自己掛了，此時(shí)肯定會重復(fù)消費(fèi)一次，自己保證冪等性就好了。

如何保證消息的冪等？

以 RocketMQ 為例，下面列出了消息重復(fù)的場景：

發(fā)送時(shí)消息重復(fù)

當(dāng)一條消息已被成功發(fā)送到服務(wù)端并完成持久化，此時(shí)出現(xiàn)了網(wǎng)絡(luò)閃斷或者客戶端宕機(jī)，導(dǎo)致服務(wù)端對客戶端應(yīng)答失敗。如果此時(shí)生產(chǎn)者意識到消息發(fā)送失敗并嘗試再次發(fā)送消息，消費(fèi)者后續(xù)會收到兩條內(nèi)容相同并且Message ID也相同的消息。

投遞時(shí)消息重復(fù)

消息消費(fèi)的場景下，消息已投遞到消費(fèi)者并完成業(yè)務(wù)處理，當(dāng)客戶端給服務(wù)端反饋應(yīng)答的時(shí)候網(wǎng)絡(luò)閃斷。為了保證消息至少被消費(fèi)一次，消息隊(duì)列RocketMQ版的服務(wù)端將在網(wǎng)絡(luò)恢復(fù)后再次嘗試投遞之前已被處理過的消息，消費(fèi)者后續(xù)會收到兩條內(nèi)容相同并且Message ID也相同的消息。

負(fù)載均衡時(shí)消息重復(fù)（包括但不限于網(wǎng)絡(luò)抖動(dòng)、Broker重啟以及消費(fèi)者應(yīng)用重啟）

當(dāng)消息隊(duì)列RocketMQ版的Broker或客戶端重啟、擴(kuò)容或縮容時(shí)，會觸發(fā)Rebalance，此時(shí)消費(fèi)者可能會收到重復(fù)消息。

那么，有什么解決方案呢？直接上圖。

如何解決消息積壓的問題？

關(guān)于這個(gè)問題，有幾個(gè)點(diǎn)需要考慮：

如何快速讓積壓的消息被消費(fèi)掉？

臨時(shí)寫一個(gè)消息分發(fā)的消費(fèi)者，把積壓隊(duì)列里的消息均勻分發(fā)到N個(gè)隊(duì)列中，同時(shí)一個(gè)隊(duì)列對應(yīng)一個(gè)消費(fèi)者，相當(dāng)于消費(fèi)速度提高了N倍。

修改前：

修改后：

積壓時(shí)間太久，導(dǎo)致部分消息過期，怎么處理？

批量重導(dǎo)。在業(yè)務(wù)不繁忙的時(shí)候，比如凌晨，提前準(zhǔn)備好程序，把丟失的那批消息查出來，重新導(dǎo)入到MQ中。

消息大量積壓，MQ磁盤被寫滿了，導(dǎo)致新消息進(jìn)不來了，丟掉了大量消息，怎么處理？

這個(gè)沒辦法。誰讓【消息分發(fā)的消費(fèi)者】寫的太慢了，你臨時(shí)寫程序，接入數(shù)據(jù)來消費(fèi)，消費(fèi)一個(gè)丟棄一個(gè)，都不要了，快速消費(fèi)掉所有的消息。然后走第二個(gè)方案，到了晚上再補(bǔ)數(shù)據(jù)吧。