讀文有感：Kafka 官方設(shè)計(jì)文檔

-? ? ?數(shù)據(jù)持久化：無(wú)懼文件系統(tǒng)? ? -

磁盤(pán)的讀寫(xiě)速度，取決于如何讀寫(xiě)。對(duì)于線(xiàn)性讀寫(xiě)方式，操作系統(tǒng)做了充分的優(yōu)化：提前讀 - 預(yù)取若干數(shù)據(jù)塊，滯后寫(xiě) - 將小的邏輯寫(xiě)操作合并成一個(gè)大的物理寫(xiě)操作。

研究表明：順序讀寫(xiě)磁盤(pán)（sequential disk access）的速度有些時(shí)候比隨機(jī)訪問(wèn)內(nèi)存還要快。

現(xiàn)代操作系統(tǒng)激進(jìn)地盡可能將空閑內(nèi)存用作磁盤(pán)緩存。所有磁盤(pán)讀寫(xiě)都經(jīng)過(guò)操作系統(tǒng)提供的統(tǒng)一緩存。這個(gè)特性沒(méi)法輕易關(guān)閉，除非直接 I/O （direct I/O），因此，如果程序在用戶(hù)進(jìn)程中進(jìn)行數(shù)據(jù)緩存，緩存的數(shù)據(jù)通常也是和操作系統(tǒng)頁(yè)緩存重復(fù)的，緩存兩遍，沒(méi)啥意義，也浪費(fèi)內(nèi)存。

而且，Kafka 是構(gòu)建在 JVM 之上的，了解 Java 內(nèi)存使用方式的人應(yīng)該都知道：

1. 對(duì)象的內(nèi)存開(kāi)銷(xiāo)非常高，通常是實(shí)際數(shù)據(jù)大小的2倍（甚至更多）

2. 隨著堆上數(shù)據(jù)量增大，Java 的 GC 表現(xiàn)也會(huì)更糟糕

因此，使用文件系統(tǒng)并依賴(lài)于操作系統(tǒng)內(nèi)存頁(yè)緩存，優(yōu)于在程序中維護(hù)一塊內(nèi)存緩存或其它結(jié)構(gòu)。至少操作系統(tǒng)內(nèi)存頁(yè)緩存的可用內(nèi)存翻倍了。另外，如果使用緊湊的字節(jié)結(jié)構(gòu)來(lái)緩存數(shù)據(jù)，相比使用對(duì)象，可用內(nèi)存可能還會(huì)翻倍。在 32GB 內(nèi)存的機(jī)器上這么搞，緩存可用到 20-30GB，還不會(huì)對(duì) GC 造成了什么壞影響。并且，即使服務(wù)重啟，這塊緩存空間也是熱的（除非機(jī)器重啟），用戶(hù)進(jìn)程內(nèi)的內(nèi)存緩存在服務(wù)重啟后得重建（10GB的數(shù)據(jù)緩存可能需要10分鐘左右）。

這樣也可以簡(jiǎn)化代碼邏輯，因?yàn)榫彺婧臀募到y(tǒng)之間的一致性由操作系統(tǒng)來(lái)保證了。

這樣一分析，設(shè)計(jì)就簡(jiǎn)單了：我們反其道而行之，所有數(shù)據(jù)都直接寫(xiě)到文件系統(tǒng)上持久化日志文件中，不需要在程序中使用內(nèi)存緩存，也不必確保將數(shù)據(jù)刷到磁盤(pán)。這實(shí)際意味著數(shù)據(jù)轉(zhuǎn)移到了內(nèi)核的內(nèi)存頁(yè)緩存。

-? ? ?常亮?xí)r間就能搞定? ? -

B 樹(shù)的 O(log N) 時(shí)間復(fù)雜度，對(duì)于磁盤(pán)操作來(lái)說(shuō)，并不能等同于常量時(shí)間復(fù)雜度。

Kafka 采用日志文件方式，確保讀寫(xiě)操作的時(shí)間復(fù)雜度是 O(1)。

Kafka 不會(huì)在消息一被消費(fèi)就立即刪除，而是保留一段時(shí)間，這樣對(duì)于消費(fèi)者來(lái)說(shuō)也更靈活一些。

-? ? ?效率? ? -

對(duì)于 Kafka 這類(lèi)系統(tǒng)而言，即使像前述那樣消除了糟糕的磁盤(pán)訪問(wèn)模式，也會(huì)遇到兩個(gè)導(dǎo)致數(shù)據(jù)效率低的問(wèn)題：過(guò)多的小 I/O 操作，以及過(guò)多的字節(jié)拷貝。

小 I/O 問(wèn)題在客戶(hù)端與服務(wù)端之間，以及服務(wù)端內(nèi)部的數(shù)據(jù)持久化操作中都會(huì)發(fā)生。對(duì)此，Kafka 協(xié)議建立在 “消息集” （即一批消息）的抽象之上，這樣網(wǎng)絡(luò)請(qǐng)求讀寫(xiě)的是一批一批的消息，減少了網(wǎng)絡(luò)往返的時(shí)間開(kāi)銷(xiāo)（注：消息處理的實(shí)時(shí)性會(huì)相對(duì)差一點(diǎn)）。服務(wù)端也是一次將一批消息寫(xiě)到日志文件中，消費(fèi)者也按序一次獲取一批消息。這一簡(jiǎn)單的優(yōu)化可以將吞吐能力提升幾個(gè)數(shù)量級(jí)。

對(duì)于過(guò)多的字節(jié)拷貝問(wèn)題，在消息量大的時(shí)候，影響比較明顯。Kafka 采用了一種標(biāo)準(zhǔn)化的二進(jìn)制消息格式，producer、broker、consumer 都使用這種格式，這樣數(shù)據(jù)塊在傳輸期間不需要變動(dòng)。

broker 維護(hù)的消息日志只是一個(gè)目錄下的一堆文件，文件內(nèi)容是按序?qū)懭氲南⒓?，消息集的?shù)據(jù)格式同于 producer、consumer 使用的。共用一種數(shù)據(jù)格式方便了一個(gè)重要的操作優(yōu)化：持久化日志塊的網(wǎng)絡(luò)傳輸。對(duì)于從內(nèi)存頁(yè)緩存（pagecache）到網(wǎng)絡(luò)套接字（socket）的數(shù)據(jù)傳輸操作，現(xiàn)代 UNIX 操作系統(tǒng)提供了一種高度優(yōu)化的代碼執(zhí)行路徑。Linux 中使用 sendfile 系統(tǒng)調(diào)用?可以利用這個(gè)優(yōu)化。

要理解 sendfile 的收益，需要先理解從文件到套接字傳輸數(shù)據(jù)的常規(guī)代碼執(zhí)行路徑：

1. 操作系統(tǒng)從磁盤(pán)將數(shù)據(jù)讀到內(nèi)核空間的內(nèi)存頁(yè)緩存（pagecache）

2. 應(yīng)用程序從內(nèi)核空間將數(shù)據(jù)讀到用戶(hù)空間緩沖區(qū)

3. 應(yīng)用程序?qū)?shù)據(jù)從用戶(hù)空間緩沖區(qū)讀到內(nèi)核空間的套接字緩沖區(qū)

4. 操作系統(tǒng)將數(shù)據(jù)從套接字緩沖區(qū)讀到 NIC 緩沖區(qū)，網(wǎng)卡從 NIC 緩沖區(qū)讀取數(shù)據(jù)通過(guò)網(wǎng)絡(luò)發(fā)出去

這一代碼執(zhí)行路徑，涉及 4 次數(shù)據(jù)拷貝和 2 次系統(tǒng)調(diào)用，很顯然是低效的。使用 sendfile，可以避免內(nèi)核空間和用戶(hù)空間之間一些不必要的數(shù)據(jù)拷貝，操作系統(tǒng)可以直接將數(shù)據(jù)從內(nèi)存頁(yè)緩存發(fā)送到網(wǎng)絡(luò)。

進(jìn)一步了解 sendfile 以及 Java 平臺(tái)如何支持零拷貝，可以閱讀這篇文章（https://developer.ibm.com/articles/j-zerocopy/）。

-? ? ?生產(chǎn)者（The Producer）? ? -

負(fù)載均衡

消息應(yīng)該發(fā)到哪個(gè)分區(qū)（partition）由客戶(hù)端根據(jù)哈希算法（或者隨機(jī)）決定，并且消息是直接由 producer 發(fā)到目標(biāo)分區(qū)的 leader broker，沒(méi)有任何中間路由層。

所有 Kafka 節(jié)點(diǎn)都可以響應(yīng)元數(shù)據(jù)請(qǐng)求 - 告知客戶(hù)端（producer 或 consumer）哪些服務(wù)節(jié)點(diǎn)還存活以及某個(gè) topic 的各個(gè)分區(qū) leader 分別是哪個(gè)節(jié)點(diǎn)（疑惑：如果某個(gè)分區(qū) leader 節(jié)點(diǎn)掛掉之后，客戶(hù)端如何獲知？何時(shí)可以獲知？）

消息交付語(yǔ)義

producer 和 consumer 之間的消息交付語(yǔ)義，分 3 種：

1. 最多消費(fèi)一次 - 消息可能會(huì)丟失，但不會(huì)被重復(fù)消費(fèi)
2. 最少消費(fèi)一次 - 消息不會(huì)丟，但可能被重復(fù)消費(fèi)
3. 僅消費(fèi)一次 - 每個(gè)消息都會(huì)被消費(fèi)且僅消費(fèi)一次

這個(gè)問(wèn)題可以分成兩個(gè)階段的問(wèn)題：producer 向 broker 發(fā)布一個(gè)消息時(shí)的持久性保證?以及?consumer 消費(fèi)一個(gè)消息時(shí)的語(yǔ)義保證?（the durability guarantees for publishing a message and the guarantees when consuming a message）。

producer 向 Kafka 集群發(fā)消息時(shí)，會(huì)提供一個(gè)請(qǐng)求參數(shù)?acks：

1. acks=0：表示 producer 不需要等分區(qū) leader broker 返回任何響應(yīng)，將消息存入套接字緩沖區(qū)（socket buffer）就當(dāng)做消息已經(jīng)發(fā)送成功。所以可靠性是沒(méi)有保證的。
2. acks=1：表示 分區(qū) leader broker 將消息寫(xiě)入自己的本地日志文件，就向 producer 響應(yīng)成功，不必等待分區(qū)副本 broker 同步好消息。
3. acks=-1 或 acks=all：表示 分區(qū) leader broker 需要等待所有同步副本 broker 同步好消息并響應(yīng)成功，才向 producer 響應(yīng)成功

第 2 種情況，如果分區(qū) leader broker 掛掉/不存活，則副本未來(lái)得及同步的消息會(huì)丟失。

第 3 種情況，只要有同步副本正常同步消息，那么即使 leader 掛了也不會(huì)丟數(shù)據(jù)。

如果 leader 被系統(tǒng)判定為不存活，則會(huì)從（同步）副本中選舉一個(gè)新的 leader，那么 Kafka 如何判定一個(gè)節(jié)點(diǎn)是否存活？存活判定依賴(lài) 2 個(gè)條件：

1. 節(jié)點(diǎn)必須維持與 Zookeeper 的 session 連接（通過(guò) Zookeeper 的心跳機(jī)制）
2. 如果是一個(gè)從節(jié)點(diǎn)（follower），則必須不斷從 leader 節(jié)點(diǎn)同步消息數(shù)據(jù)，且同步進(jìn)度沒(méi)有落后太多

如果 producer 在發(fā)送消息的過(guò)程中發(fā)生網(wǎng)絡(luò)問(wèn)題，它沒(méi)法判定分區(qū) leader 是否收到消息。0.11.0.0 版本之前，producer 只能重發(fā)消息，別無(wú)他法，因此只能提供“最少消費(fèi)一次的”交付語(yǔ)義。0.11.0.0 版本之后，Kafka producer 支持一個(gè)冪等交付功能選項(xiàng)，可以確保消息重發(fā)不會(huì)導(dǎo)致 Kafka 的消息日志中出現(xiàn)重復(fù)的條目：broker 為每個(gè) producer 分配一個(gè) ID，然后基于消息序號(hào)來(lái)去重。

也是從 0.11.0.0 版本開(kāi)始，Producer 支持以類(lèi)事務(wù)的語(yǔ)義向多個(gè) topic 分區(qū)發(fā)送消息：要么所有消息都發(fā)送成功，要么都不成功。這個(gè)能力主要用于實(shí)現(xiàn) Kafka topic 之間的僅處理一次語(yǔ)義。

從 consumer 角度來(lái)看，同一個(gè)分區(qū)的所有副本，日志數(shù)據(jù)相同，消費(fèi)進(jìn)度也一樣。consumer 可以控制自己對(duì)分區(qū)日志數(shù)據(jù)的消費(fèi)位置。

1. 如果 consumer 讀取消息后，先向 kafka 提交消費(fèi)位置，再處理消息；如果該 consumer 掛掉或重啟，會(huì)可能導(dǎo)致丟消息，從而只能滿(mǎn)足“最多處理一次”交付語(yǔ)義。
2. 如果 consumer 讀取消息后，是先處理，再提交消費(fèi)位置；如果該 consumer 掛掉或重啟，則可能導(dǎo)致重復(fù)消費(fèi)消息，從而只能滿(mǎn)足“最少處理一次”交付語(yǔ)義。

如何實(shí)現(xiàn)“僅處理一次”語(yǔ)義？借助 Producer 的事務(wù)能力。

-? ? ?復(fù)制? ? -

復(fù)制的粒度/單元是 topic 分區(qū)。Kafka 集群中，每個(gè)分區(qū)都有一個(gè) leader broker 節(jié)點(diǎn)，0個(gè)或多個(gè)從節(jié)點(diǎn)（follower）。分區(qū)讀寫(xiě)都是由 leader broker 處理。

如同一個(gè)普通的 consumer，從節(jié)點(diǎn)從 leader broker 拉?。╬ull）消息，然后寫(xiě)到自己的消息日志文件中。讓從節(jié)點(diǎn)以 pull 的方式獲取 leader 的消息數(shù)據(jù)，好處在于批量讀寫(xiě)。

對(duì)于 follower 節(jié)點(diǎn)而言，“是否存活”的實(shí)際含義是“是否順利地從 leader 同步消息”，leader 節(jié)點(diǎn)會(huì)追蹤“同步中”節(jié)點(diǎn)集（ISRs）。如果一個(gè) follower 掛掉了/卡住了/同步落后太多了，則將其從這個(gè) ISRs 中移除。follow 是否卡住或者同步落后太多，依據(jù)?replica.lag.time.max.ms?配置參數(shù)判定。

將某消息寫(xiě)到某個(gè)分區(qū)，如果該分區(qū)所有同步中副本都已經(jīng)將該消息寫(xiě)到自己的消息日志文件中，則可以認(rèn)為該消息的寫(xiě)操作已提交（committed），也就是真正的寫(xiě)成功。

只有寫(xiě)提交的消息才會(huì)分發(fā)給 consumer。

producer 可以選擇是否等待消息寫(xiě)操作提交，在延遲（latency）和持久性（durability）之間權(quán)衡。

Kafka 集群在某分區(qū)的 leader 節(jié)點(diǎn)掛掉之后，會(huì)快速進(jìn)行失敗轉(zhuǎn)移（a short fail-over period），選舉出新的分區(qū) leader 節(jié)點(diǎn)，可用性不會(huì)受到影響。但如果發(fā)生網(wǎng)絡(luò)分區(qū)（network partitions）問(wèn)題，則無(wú)法保證可用性。CAP - C（Consistency）：一致性，A（Availability）：可用性，P（Partition Tolerance）：分區(qū)容錯(cuò)性 - 放棄了 分區(qū)容錯(cuò)性。

日志數(shù)據(jù)復(fù)制：仲裁成員集（Quorums）、同步中副本集（ISRs）和狀態(tài)機(jī)

（備注：這一節(jié)我理解得還不太透徹。）

一類(lèi)常見(jiàn)的分布式系統(tǒng)是主從模式的，由主節(jié)點(diǎn)決定狀態(tài)變化的順序（the order of a series of values）。從節(jié)點(diǎn)通過(guò)日志復(fù)制（replicated log）方式同步狀態(tài)數(shù)據(jù)。對(duì)于提交決策（commit decision）和選主（leader election），通常是基于多數(shù)人投票的機(jī)制。假設(shè)副本個(gè)數(shù)（注：個(gè)人理解包含主節(jié)點(diǎn)）為 2f+1，那么只有當(dāng) f+1 個(gè)副本寫(xiě)入成功，主節(jié)點(diǎn)才會(huì)將這個(gè)寫(xiě)操作標(biāo)記為已提交（committed）。當(dāng)主節(jié)點(diǎn)掛掉之后，基于 f 個(gè)狀態(tài)最新的副本節(jié)點(diǎn)，可以選舉出新的主節(jié)點(diǎn)，且狀態(tài)不會(huì)有任何丟失。

多數(shù)人投票方式，有一個(gè)優(yōu)點(diǎn)：延遲取決于速度快的節(jié)點(diǎn)，而不是慢的。缺點(diǎn)是：對(duì)于實(shí)際的生產(chǎn)系統(tǒng)，抗風(fēng)險(xiǎn)能力還不夠，而且不夠靈活，不能讓使用者做權(quán)衡。

Kafka 選擇仲裁成員集（quorum set）的方式與此不同，而不是基于多數(shù)人投票，而是動(dòng)態(tài)維護(hù)一組同步中副本（ISR），這些副本與主節(jié)點(diǎn)保持同步。只有這組副本中的成員才有資格當(dāng)選為主節(jié)點(diǎn)。ISR 集發(fā)生變化時(shí)會(huì)持久化到 Zookeeper 上。

基于 ISR 模型，如果 topic 分區(qū)有 f+1 個(gè)副本，則可以容忍 f 個(gè)節(jié)點(diǎn)掛掉，也不會(huì)丟失任何已提交的消息。

與 Kafka ISR 模型實(shí)際實(shí)現(xiàn)最相近的學(xué)術(shù)論文是微軟的 PacificA（http://research.microsoft.com/apps/pubs/default.aspx?id=66814）。

-? ? ?可用性和持久性保證? ? -

注意：producer 發(fā)送消息時(shí)設(shè)定?acks=all?并不是要求所有的副本都確認(rèn)寫(xiě)入成功，而是在當(dāng)前同步中副本（ISR）都確認(rèn)寫(xiě)入成功時(shí)，分區(qū) leader 就向 producer 響應(yīng)成功。例如：某個(gè) topic 被設(shè)置為 2 個(gè)副本，然后其中一個(gè)副本節(jié)點(diǎn)掛掉，此時(shí)要求?acks=all?的寫(xiě)操作也會(huì)成功。如果剩下的副本節(jié)點(diǎn)也掛了，那么就會(huì)丟消息啦。

為了方便用戶(hù)在 可用性 和 持久性 之間權(quán)衡，Kafka 提供兩個(gè) topic 級(jí)別的配置，用于 持久性 比 可用性 重要的情況：

1. 禁用臟 leader 選舉
2. 指定一個(gè)最小 ISR 集大?。?/span>min.insync.replicas?參數(shù)設(shè)置）：只有當(dāng) ISR 集大小大于設(shè)定的最小值，分區(qū) [leader] 才會(huì)接受消息寫(xiě)入。這個(gè)設(shè)置只有當(dāng) producer 使用?acks=all?時(shí)才會(huì)生效。（注：在我們生產(chǎn)環(huán)境中，分區(qū)副本數(shù)通常申請(qǐng)為 3（包含 leader），那么?min.insync.replicas?應(yīng)該設(shè)定為 2，但默認(rèn)是 1。使用 1，那么當(dāng)分區(qū)只有一個(gè)副本（即 leader），producer 也能寫(xiě)入成功，但如果這個(gè)副本又掛了，就會(huì)丟數(shù)據(jù)。）

-? ? ?副本管理? ? -

一個(gè) Kafka 集群上一般會(huì)有多個(gè) topic，每個(gè) topic 又有多個(gè) partition，為了節(jié)點(diǎn)之間負(fù)載均衡，通常以循環(huán)（round-robin）方式在所有節(jié)點(diǎn)上分布 partition 和 分區(qū) leader 角色。

另外，在分區(qū) leader 節(jié)點(diǎn)之后重新選出 leader 之前，存在一段不可用的時(shí)間窗口，為了縮短這個(gè)時(shí)間窗口，Kafka 會(huì)從所有 broker 中選擇一個(gè)作為“控制器（controller）”，這個(gè)控制器會(huì)檢測(cè) broker 級(jí)別的問(wèn)題（failures），在發(fā)現(xiàn)某個(gè) broker 掛掉之后，負(fù)責(zé)為受影響的分區(qū)指定新的 leader，而不是每個(gè)分區(qū)自己負(fù)責(zé)重新選主，這樣的選主過(guò)程更輕量更快。如果控制器節(jié)點(diǎn)掛了，還存活的 broker 中的一個(gè)會(huì)成為新的控制器。

-? ? ?消費(fèi)者消費(fèi)進(jìn)度跟蹤? ? -

Kafka 為每個(gè)消費(fèi)組（consumer group）指定一個(gè) broker 來(lái)存儲(chǔ)目標(biāo) topic 各個(gè)分區(qū)的消費(fèi)進(jìn)度（offsets），這個(gè) broker 稱(chēng)為?組協(xié)調(diào)器（group coordinator）。這個(gè)消費(fèi)組中的任一消費(fèi)者實(shí)例都應(yīng)該將消費(fèi)進(jìn)度提交到這個(gè)組協(xié)調(diào)器，或者從這個(gè)組協(xié)調(diào)器獲取啟動(dòng)之前上次的消費(fèi)進(jìn)度。Kafka 基于消費(fèi)組的名稱(chēng)為消費(fèi)組分配協(xié)調(diào)器。消費(fèi)者可以向任一 broker 發(fā)送 FindCoordinatorRequest 請(qǐng)求來(lái)查找自己的協(xié)調(diào)器，并從 FindCoordinatorResponse 響應(yīng)中獲取協(xié)調(diào)器的詳細(xì)信息。

在組協(xié)調(diào)器接收到一個(gè) OffsetCommitRequest 請(qǐng)求后，會(huì)將請(qǐng)求數(shù)據(jù)寫(xiě)到一個(gè)特殊的經(jīng)壓實(shí)的（compacted）（http://kafka.apache.org/documentation/#compaction）?Kafka topic -?__consumer_offsets。在目標(biāo)分區(qū)的所有副本都確認(rèn)收到了，協(xié)調(diào)器才會(huì)向消費(fèi)者發(fā)送進(jìn)度提交成功的響應(yīng)。這個(gè) topic 的消息日志數(shù)據(jù)會(huì)定期進(jìn)行壓實(shí)（compact），因?yàn)橹恍枰獮槊總€(gè)分區(qū)維護(hù)最新的消費(fèi)進(jìn)度。協(xié)調(diào)器也會(huì)在內(nèi)存中緩存消費(fèi)進(jìn)度，方便快速響應(yīng)消費(fèi)進(jìn)度查詢(xún)請(qǐng)求。

注：如果消費(fèi)者/消費(fèi)組特別多（例如：我們廣告引擎服務(wù)，讀取正排消息 topic，一個(gè)機(jī)器實(shí)例就是一個(gè) consumer group，數(shù)量在幾百到幾千不等），那么組協(xié)調(diào)器的壓力會(huì)比較大，那么確保組協(xié)調(diào)器的角色均勻分配到集群的所有 broker，比較關(guān)鍵。另外，__consumer_offsets?這個(gè) topic 的分區(qū)數(shù)量不能太少，最好和 broker 數(shù)量相同或者整數(shù)倍數(shù)量。

作者：xiayf

來(lái)源：

http://blog.xiayf.cn/2019/10/13/reading-kafka-design/