周末無聊刷著手機(jī)，某寶網(wǎng) App 突然蹦出來一條消息“為了回饋老客戶，女朋友買一送一，活動(dòng)僅限今天!”。

買一送一還有這種好事，那我可不能錯(cuò)過!忍不住立馬點(diǎn)了去。于是選了兩個(gè)***款，下單、支付一氣呵成!滿足的躺在床上，想著馬上有女朋友了，竟然幸福的失眠了……

第二天正常上著班，突然接到快遞小哥的電話：

小哥：“你是 xx 嗎?你的女朋友到了，我現(xiàn)在在你樓下，你來拿一下吧!”。

我：“這……我在上班呢，可以晚上送過來嗎?“。

小哥：“晚上可不行哦，晚上我也下班了呢!”。

于是兩個(gè)人僵持了很久……

***小哥說，要不我?guī)湍惴诺綐窍滦》急憷臧?，你晚上下班了過來拿，尷尬的局面這才得以緩解!

為什么需要消息隊(duì)列

回到正題，如果沒有小芳便利店，那快遞小哥和我的交互圖就應(yīng)：

會(huì)出現(xiàn)什么情況呢?

• 為了這個(gè)女朋友，我請(qǐng)假回去拿(老板不批)。
• 小哥一直在你樓下等(小哥還有其他的快遞要送)。
• 周末再送(顯然等不及)。
• 這個(gè)女朋友我不要了(絕對(duì)不可能)!

小芳便利店出現(xiàn)后，交互圖如下：

在上面例子中，“快遞小哥”和“買女朋友的我”就是需要交互的兩個(gè)系統(tǒng)，小芳便利店就是我們本文要講的消息中間件。

總結(jié)下來小芳便利店(消息中間件)出現(xiàn)后有如下好處：

解耦

快遞小哥手上有很多快遞需要送，他每次都需要先電話一一確認(rèn)收貨人是否有空、哪個(gè)時(shí)間段有空，然后再確定好送貨的方案。這樣完全依賴收貨人了!如果快遞一多，快遞小哥估計(jì)得忙瘋了……

如果有了便利店，快遞小哥只需要將同一個(gè)小區(qū)的快遞放在同一個(gè)便利店，然后通知收貨人來取貨就可以了，這時(shí)候快遞小哥和收貨人就實(shí)現(xiàn)了解耦!

異步

快遞小哥打電話給我后需要一直在你樓下等著，直到我拿走你的快遞他才能去送其他人的。

快遞小哥將快遞放在小芳便利店后，又可以干其他的活兒去了，不需要等待你到來而一直處于等待狀態(tài)，提高了工作的效率。

削峰

假設(shè)雙十一我買了不同店里的各種商品，而恰巧這些店發(fā)貨的快遞都不一樣，有中通、圓通、申通、各種通等……更巧的是他們都同時(shí)到貨了!

中通的小哥打來電話叫我去北門取快遞、圓通小哥叫我去南門、申通小哥叫我去東門。我一時(shí)手忙腳亂……

我們能看到在系統(tǒng)需要交互的場(chǎng)景中，使用消息隊(duì)列中間件真的是好處多多，基于這種思路，就有了豐巢、菜鳥驛站等比小芳便利店更專業(yè)的“中間件”了。

***，上面的故事純屬虛構(gòu)……

消息隊(duì)列通信的模式

通過上面的例子我們引出了消息中間件，并且介紹了消息隊(duì)列出現(xiàn)后的好處，這里就需要介紹消息隊(duì)列通信的兩種模式了：

點(diǎn)對(duì)點(diǎn)模式

如上圖所示，點(diǎn)對(duì)點(diǎn)模式通常是基于拉取或者輪詢的消息傳送模型，這個(gè)模型的特點(diǎn)是發(fā)送到隊(duì)列的消息被一個(gè)且只有一個(gè)消費(fèi)者進(jìn)行處理。

生產(chǎn)者將消息放入消息隊(duì)列后，由消費(fèi)者主動(dòng)的去拉取消息進(jìn)行消費(fèi)。點(diǎn)對(duì)點(diǎn)模型的優(yōu)點(diǎn)是消費(fèi)者拉取消息的頻率可以由自己控制。

但是消息隊(duì)列是否有消息需要消費(fèi)，在消費(fèi)者端無法感知，所以在消費(fèi)者端需要額外的線程去監(jiān)控。

發(fā)布訂閱模式

如上圖所示，發(fā)布訂閱模式是一個(gè)基于消息送的消息傳送模型，該模型可以有多種不同的訂閱者。

生產(chǎn)者將消息放入消息隊(duì)列后，隊(duì)列會(huì)將消息推送給訂閱過該類消息的消費(fèi)者(類似微信公眾號(hào))。

由于是消費(fèi)者被動(dòng)接收推送，所以無需感知消息隊(duì)列是否有待消費(fèi)的消息!但是 Consumer1、Consumer2、Consumer3 由于機(jī)器性能不一樣，所以處理消息的能力也會(huì)不一樣，但消息隊(duì)列卻無法感知消費(fèi)者消費(fèi)的速度!

所以推送的速度成了發(fā)布訂閱模式的一個(gè)問題!假設(shè)三個(gè)消費(fèi)者處理速度分別是 8M/s、5M/s、2M/s，如果隊(duì)列推送的速度為 5M/s，則 Consumer3 無法承受!

如果隊(duì)列推送的速度為 2M/s，則 Consumer1、Consumer2 會(huì)出現(xiàn)資源的極大浪費(fèi)!

Kafka

上面簡(jiǎn)單的介紹了為什么需要消息隊(duì)列以及消息隊(duì)列通信的兩種模式，接下來就到了我們本文的主角 Kafka 閃亮登場(chǎng)的時(shí)候了!

Kafka 是一種高吞吐量的分布式發(fā)布訂閱消息系統(tǒng)，它可以處理消費(fèi)者規(guī)模的網(wǎng)站中的所有動(dòng)作流數(shù)據(jù)，具有高性能、持久化、多副本備份、橫向擴(kuò)展能力………

基礎(chǔ)架構(gòu)及術(shù)語

話不多說，先看圖，通過這張圖我們來捋一捋相關(guān)的概念及之間的關(guān)系：

如果看到這張圖你很懵逼，木有關(guān)系!我們先來分析相關(guān)概念：

• Producer：Producer 即生產(chǎn)者，消息的產(chǎn)生者，是消息的入口。
• Kafka Cluster：

Broker：Broker 是 Kafka 實(shí)例，每個(gè)服務(wù)器上有一個(gè)或多個(gè) Kafka 的實(shí)例，我們姑且認(rèn)為每個(gè) Broker 對(duì)應(yīng)一臺(tái)服務(wù)器。

每個(gè) Kafka 集群內(nèi)的 Broker 都有一個(gè)不重復(fù)的編號(hào)，如圖中的 Broker-0、Broker-1 等……

Topic：消息的主題，可以理解為消息的分類，Kafka 的數(shù)據(jù)就保存在 Topic。在每個(gè) Broker 上都可以創(chuàng)建多個(gè) Topic。

Partition：Topic 的分區(qū)，每個(gè) Topic 可以有多個(gè)分區(qū)，分區(qū)的作用是做負(fù)載，提高 Kafka 的吞吐量。

同一個(gè) Topic 在不同的分區(qū)的數(shù)據(jù)是不重復(fù)的，Partition 的表現(xiàn)形式就是一個(gè)一個(gè)的文件夾!

Replication：每一個(gè)分區(qū)都有多個(gè)副本，副本的作用是做備胎。當(dāng)主分區(qū)(Leader)故障的時(shí)候會(huì)選擇一個(gè)備胎(Follower)上位，成為 Leader。

在 Kafka 中默認(rèn)副本的***數(shù)量是 10 個(gè)，且副本的數(shù)量不能大于 Broker 的數(shù)量，F(xiàn)ollower 和 Leader 絕對(duì)是在不同的機(jī)器，同一機(jī)器對(duì)同一個(gè)分區(qū)也只可能存放一個(gè)副本(包括自己)。

• Message：每一條發(fā)送的消息主體。
• Consumer：消費(fèi)者，即消息的消費(fèi)方，是消息的出口。
• Consumer Group：我們可以將多個(gè)消費(fèi)組組成一個(gè)消費(fèi)者組，在 Kafka 的設(shè)計(jì)中同一個(gè)分區(qū)的數(shù)據(jù)只能被消費(fèi)者組中的某一個(gè)消費(fèi)者消費(fèi)。

同一個(gè)消費(fèi)者組的消費(fèi)者可以消費(fèi)同一個(gè) Topic 的不同分區(qū)的數(shù)據(jù)，這也是為了提高 Kafka 的吞吐量!

• Zookeeper：Kafka 集群依賴 Zookeeper 來保存集群的的元信息，來保證系統(tǒng)的可用性。

工作流程分析

上面介紹了 Kafka 的基礎(chǔ)架構(gòu)及基本概念，不知道大家看完有沒有對(duì) Kafka 有個(gè)大致印象，如果還比較懵也沒關(guān)系!

我們接下來再結(jié)合上面的結(jié)構(gòu)圖分析 Kafka 的工作流程，***再回來整個(gè)梳理一遍我相信你會(huì)更有收獲!

發(fā)送數(shù)據(jù)

我們看上面的架構(gòu)圖中，Producer 就是生產(chǎn)者，是數(shù)據(jù)的入口。注意看圖中的紅色箭頭，Producer 在寫入數(shù)據(jù)的時(shí)候永遠(yuǎn)在找 Leader，不會(huì)直接將數(shù)據(jù)寫入 Follower!

那 Leader 怎么找呢?寫入的流程又是什么樣的呢?我們看下圖：

發(fā)送的流程就在圖中已經(jīng)說明了，就不單獨(dú)在文字列出來了!需要注意的一點(diǎn)是，消息寫入 Leader 后，F(xiàn)ollower 是主動(dòng)的去 Leader 進(jìn)行同步的!

Producer 采用 Push 模式將數(shù)據(jù)發(fā)布到 Broker，每條消息追加到分區(qū)中，順序?qū)懭氪疟P，所以保證同一分區(qū)內(nèi)的數(shù)據(jù)是有序的!

寫入示意圖如下：

上面說到數(shù)據(jù)會(huì)寫入到不同的分區(qū)，那 Kafka 為什么要做分區(qū)呢?相信大家應(yīng)該也能猜到，分區(qū)的主要目的是：

• 方便擴(kuò)展。因?yàn)橐粋€(gè) Topic 可以有多個(gè) Partition，所以我們可以通過擴(kuò)展機(jī)器去輕松的應(yīng)對(duì)日益增長(zhǎng)的數(shù)據(jù)量。
• 提高并發(fā)。以 Partition 為讀寫單位，可以多個(gè)消費(fèi)者同時(shí)消費(fèi)數(shù)據(jù)，提高了消息的處理效率。

熟悉負(fù)載均衡的朋友應(yīng)該知道，當(dāng)我們向某個(gè)服務(wù)器發(fā)送請(qǐng)求的時(shí)候，服務(wù)端可能會(huì)對(duì)請(qǐng)求做一個(gè)負(fù)載，將流量分發(fā)到不同的服務(wù)器。

那在 Kafka 中，如果某個(gè) Topic 有多個(gè) Partition，Producer 又怎么知道該將數(shù)據(jù)發(fā)往哪個(gè) Partition 呢?

Kafka 中有幾個(gè)原則：

• Partition 在寫入的時(shí)候可以指定需要寫入的 Partition，如果有指定，則寫入對(duì)應(yīng)的 Partition。
• 如果沒有指定 Partition，但是設(shè)置了數(shù)據(jù)的 Key，則會(huì)根據(jù) Key 的值 Hash 出一個(gè) Partition。
• 如果既沒指定 Partition，又沒有設(shè)置 Key，則會(huì)輪詢選出一個(gè) Partition。

保證消息不丟失是一個(gè)消息隊(duì)列中間件的基本保證，那 Producer 在向 Kafka 寫入消息的時(shí)候，怎么保證消息不丟失呢?

其實(shí)上面的寫入流程圖中有描述出來，那就是通過 ACK 應(yīng)答機(jī)制!在生產(chǎn)者向隊(duì)列寫入數(shù)據(jù)的時(shí)候可以設(shè)置參數(shù)來確定是否確認(rèn) Kafka 接收到數(shù)據(jù)，這個(gè)參數(shù)可設(shè)置的值為 0、1、all：

• 0 代表 Producer 往集群發(fā)送數(shù)據(jù)不需要等到集群的返回，不確保消息發(fā)送成功。安全性***但是效率***。
• 1 代表 Producer 往集群發(fā)送數(shù)據(jù)只要 Leader 應(yīng)答就可以發(fā)送下一條，只確保 Leader 發(fā)送成功。
• all 代表 Producer 往集群發(fā)送數(shù)據(jù)需要所有的 Follower 都完成從 Leader 的同步才會(huì)發(fā)送下一條，確保 Leader 發(fā)送成功和所有的副本都完成備份。安全性***，但是效率***。

***要注意的是，如果往不存在的 Topic 寫數(shù)據(jù)，能不能寫入成功呢?Kafka 會(huì)自動(dòng)創(chuàng)建 Topic，分區(qū)和副本的數(shù)量根據(jù)默認(rèn)配置都是 1。

保存數(shù)據(jù)

Producer 將數(shù)據(jù)寫入 Kafka 后，集群就需要對(duì)數(shù)據(jù)進(jìn)行保存了!Kafka 將數(shù)據(jù)保存在磁盤，可能在我們的一般的認(rèn)知里，寫入磁盤是比較耗時(shí)的操作，不適合這種高并發(fā)的組件。

Kafka 初始會(huì)單獨(dú)開辟一塊磁盤空間，順序?qū)懭霐?shù)據(jù)(效率比隨機(jī)寫入高)。

①Partition 結(jié)構(gòu)

前面說過了每個(gè) Topic 都可以分為一個(gè)或多個(gè) Partition，如果你覺得 Topic 比較抽象，那 Partition 就是比較具體的東西了!

Partition 在服務(wù)器上的表現(xiàn)形式就是一個(gè)一個(gè)的文件夾，每個(gè) Partition 的文件夾下面會(huì)有多組 Segment 文件。

每組 Segment 文件又包含 .index 文件、.log 文件、.timeindex 文件(早期版本中沒有)三個(gè)文件。

Log 文件就是實(shí)際存儲(chǔ) Message 的地方，而 Index 和 Timeindex 文件為索引文件，用于檢索消息。

如上圖，這個(gè) Partition 有三組 Segment 文件，每個(gè) Log 文件的大小是一樣的，但是存儲(chǔ)的 Message 數(shù)量是不一定相等的(每條的 Message 大小不一致)。

文件的命名是以該 Segment 最小 Offset 來命名的，如 000.index 存儲(chǔ) Offset 為 0~368795 的消息，Kafka 就是利用分段+索引的方式來解決查找效率的問題。

②Message 結(jié)構(gòu)

上面說到 Log 文件就實(shí)際是存儲(chǔ) Message 的地方，我們?cè)?Producer 往 Kafka 寫入的也是一條一條的 Message。

那存儲(chǔ)在 Log 中的 Message 是什么樣子的呢?消息主要包含消息體、消息大小、Offset、壓縮類型……等等!

我們重點(diǎn)需要知道的是下面三個(gè)：

• Offset：Offset 是一個(gè)占 8byte 的有序 id 號(hào)，它可以唯一確定每條消息在 Parition 內(nèi)的位置!
• 消息大?。合⒋笮≌加?4byte，用于描述消息的大小。
• 消息體：消息體存放的是實(shí)際的消息數(shù)據(jù)(被壓縮過)，占用的空間根據(jù)具體的消息而不一樣。

③存儲(chǔ)策略

無論消息是否被消費(fèi)，Kafka 都會(huì)保存所有的消息。那對(duì)于舊數(shù)據(jù)有什么刪除策略呢?

• 基于時(shí)間，默認(rèn)配置是 168 小時(shí)(7 天)。
• 基于大小，默認(rèn)配置是 1073741824。

需要注意的是，Kafka 讀取特定消息的時(shí)間復(fù)雜度是 O(1)，所以這里刪除過期的文件并不會(huì)提高 Kafka 的性能!

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

消息存儲(chǔ)在 Log 文件后，消費(fèi)者就可以進(jìn)行消費(fèi)了。在講消息隊(duì)列通信的兩種模式的時(shí)候講到過點(diǎn)對(duì)點(diǎn)模式和發(fā)布訂閱模式。

Kafka 采用的是點(diǎn)對(duì)點(diǎn)的模式，消費(fèi)者主動(dòng)的去 Kafka 集群拉取消息，與 Producer 相同的是，消費(fèi)者在拉取消息的時(shí)候也是找 Leader 去拉取。

多個(gè)消費(fèi)者可以組成一個(gè)消費(fèi)者組(Consumer Group)，每個(gè)消費(fèi)者組都有一個(gè)組 id!

同一個(gè)消費(fèi)組者的消費(fèi)者可以消費(fèi)同一 Topic 下不同分區(qū)的數(shù)據(jù)，但是不會(huì)組內(nèi)多個(gè)消費(fèi)者消費(fèi)同一分區(qū)的數(shù)據(jù)!

是不是有點(diǎn)繞?我們看下圖：

圖示是消費(fèi)者組內(nèi)的消費(fèi)者小于 Partition 數(shù)量的情況，所以會(huì)出現(xiàn)某個(gè)消費(fèi)者消費(fèi)多個(gè) Partition 數(shù)據(jù)的情況，消費(fèi)的速度也就不及只處理一個(gè) Partition 的消費(fèi)者的處理速度!

如果是消費(fèi)者組的消費(fèi)者多于 Partition 的數(shù)量，那會(huì)不會(huì)出現(xiàn)多個(gè)消費(fèi)者消費(fèi)同一個(gè) Partition 的數(shù)據(jù)呢?

上面已經(jīng)提到過不會(huì)出現(xiàn)這種情況!多出來的消費(fèi)者不消費(fèi)任何 Partition 的數(shù)據(jù)。

所以在實(shí)際的應(yīng)用中，建議消費(fèi)者組的 Consumer 的數(shù)量與 Partition 的數(shù)量一致!

在保存數(shù)據(jù)的小節(jié)里面，我們聊到了 Partition 劃分為多組 Segment，每個(gè) Segment 又包含 .log、.index、.timeindex 文件，存放的每條 Message 包含 Offset、消息大小、消息體……

我們多次提到 Segment 和 Offset，查找消息的時(shí)候是怎么利用 Segment+Offset 配合查找的呢?

假如現(xiàn)在需要查找一個(gè) Offset 為 368801 的 Message 是什么樣的過程呢?我們先看看下面的圖：

①先找到 Offset 的 368801message 所在的 Segment 文件(利用二分法查找)，這里找到的就是在第二個(gè) Segment 文件。

②打開找到的 Segment 中的 .index 文件(也就是 368796.index 文件，該文件起始偏移量為 368796+1。

我們要查找的 Offset 為 368801 的 Message 在該 Index 內(nèi)的偏移量為 368796+5=368801，所以這里要查找的相對(duì) Offset 為 5)。

由于該文件采用的是稀疏索引的方式存儲(chǔ)著相對(duì) Offset 及對(duì)應(yīng) Message 物理偏移量的關(guān)系，所以直接找相對(duì) Offset 為 5 的索引找不到。

這里同樣利用二分法查找相對(duì) Offset 小于或者等于指定的相對(duì) Offset 的索引條目中***的那個(gè)相對(duì) Offset，所以找到的是相對(duì) Offset 為 4 的這個(gè)索引。

③根據(jù)找到的相對(duì) Offset 為 4 的索引確定 Message 存儲(chǔ)的物理偏移位置為 256。

打開數(shù)據(jù)文件，從位置為 256 的那個(gè)地方開始順序掃描直到找到 Offset 為 368801 的那條 Message。

這套機(jī)制是建立在 Offset 為有序的基礎(chǔ)上，利用 Segment+有序 Offset+稀疏索引+二分查找+順序查找等多種手段來高效的查找數(shù)據(jù)!

至此，消費(fèi)者就能拿到需要處理的數(shù)據(jù)進(jìn)行處理了。那每個(gè)消費(fèi)者又是怎么記錄自己消費(fèi)的位置呢?

在早期的版本中，消費(fèi)者將消費(fèi)到的 Offset 維護(hù)在 Zookeeper 中，Consumer 每間隔一段時(shí)間上報(bào)一次，這里容易導(dǎo)致重復(fù)消費(fèi)，且性能不好!

在新的版本中消費(fèi)者消費(fèi)到的 Offset 已經(jīng)直接維護(hù)在 Kafka 集群的 __consumer_offsets 這個(gè) Topic 中!

作者：蘇靜

簡(jiǎn)介：有過多年大型互聯(lián)網(wǎng)項(xiàng)目的開發(fā)經(jīng)驗(yàn)，對(duì)高并發(fā)、分布式、以及微服務(wù)技術(shù)有深入的研究及相關(guān)實(shí)踐經(jīng)驗(yàn)。經(jīng)歷過自學(xué)，熱衷于技術(shù)研究與分享!格言：始終保持虛心學(xué)習(xí)的態(tài)度!