從面試角度詳解Kafka

Kafka 是一個(gè)優(yōu)秀的分布式消息中間件，許多系統(tǒng)中都會(huì)使用到 Kafka 來做消息通信。對(duì)分布式消息系統(tǒng)的了解和使用幾乎成為一個(gè)開發(fā)人員必備的技能。

講一講分布式消息中間件

問題

• 什么是分布式消息中間件？
• 消息中間件的作用是什么？
• 消息中間件的使用場(chǎng)景是什么？
• 消息中間件選型？

分布式消息是一種通信機(jī)制，和 RPC、HTTP、RMI 等不一樣，消息中間件采用分布式中間代理的方式進(jìn)行通信。如圖所示，采用了消息中間件之后，上游業(yè)務(wù)系統(tǒng)發(fā)送消息，先存儲(chǔ)在消息中間件，然后由消息中間件將消息分發(fā)到對(duì)應(yīng)的業(yè)務(wù)模塊應(yīng)用（分布式生產(chǎn)者 - 消費(fèi)者模式）。這種異步的方式，減少了服務(wù)之間的耦合程度。

定義消息中間件：

• 利用高效可靠的消息傳遞機(jī)制進(jìn)行平臺(tái)無關(guān)的數(shù)據(jù)交流
• 基于數(shù)據(jù)通信，來進(jìn)行分布式系統(tǒng)的集成
• 通過提供消息傳遞和消息排隊(duì)模型，可以在分布式環(huán)境下擴(kuò)展進(jìn)程間的通信

在系統(tǒng)架構(gòu)中引用額外的組件，必然提高系統(tǒng)的架構(gòu)復(fù)雜度和運(yùn)維的難度，那么在系統(tǒng)中使用分布式消息中間件有什么優(yōu)勢(shì)呢？消息中間件在系統(tǒng)中起的作用又是什么呢？

• 解耦
• 冗余（存儲(chǔ)）
• 擴(kuò)展性
• 削峰
• 可恢復(fù)性
• 順序保證
• 緩沖
• 異步通信

下面是常見的幾種分布式消息系統(tǒng)的對(duì)比：

答案關(guān)鍵字

• 什么是分布式消息中間件？通信，隊(duì)列，分布式，生產(chǎn)消費(fèi)者模式。
• 消息中間件的作用是什么？解耦、峰值處理、異步通信、緩沖。
• 消息中間件的使用場(chǎng)景是什么？異步通信，消息存儲(chǔ)處理。
• 消息中間件選型？語言，協(xié)議、HA、數(shù)據(jù)可靠性、性能、事務(wù)、生態(tài)、簡(jiǎn)易、推拉模式。

Kafka 基本概念和架構(gòu)

問題

• 簡(jiǎn)單講下 Kafka 的架構(gòu)？
• Kafka 是推模式還是拉模式，推拉的區(qū)別是什么？
• Kafka 如何廣播消息？
• Kafka 的消息是否是有序的？
• Kafka 是否支持讀寫分離？
• Kafka 如何保證數(shù)據(jù)高可用？
• Kafka 中 zookeeper 的作用？
• 是否支持事務(wù)？
• 分區(qū)數(shù)是否可以減少？

Kafka 架構(gòu)中的一般概念：

• Producer：生產(chǎn)者，也就是發(fā)送消息的一方。生產(chǎn)者負(fù)責(zé)創(chuàng)建消息，然后將其發(fā)送到 Kafka。
• Consumer：消費(fèi)者，也就是接受消息的一方。消費(fèi)者連接到 Kafka 上并接收消息，進(jìn)而進(jìn)行相應(yīng)的業(yè)務(wù)邏輯處理。
• Consumer Group：一個(gè)消費(fèi)者組可以包含一個(gè)或多個(gè)消費(fèi)者。使用多分區(qū) + 多消費(fèi)者方式可以極大提高數(shù)據(jù)下游的處理速度，同一消費(fèi)組中的消費(fèi)者不會(huì)重復(fù)消費(fèi)消息，同樣的，不同消費(fèi)組中的消費(fèi)者消息消息時(shí)互不影響。Kafka 就是通過消費(fèi)組的方式來實(shí)現(xiàn)消息 P2P 模式和廣播模式。
• Broker：服務(wù)代理節(jié)點(diǎn)。Broker 是 Kafka 的服務(wù)節(jié)點(diǎn)，即 Kafka 的服務(wù)器。
• Topic：Kafka 中的消息以 Topic 為單位進(jìn)行劃分，生產(chǎn)者將消息發(fā)送到特定的 Topic，而消費(fèi)者負(fù)責(zé)訂閱 Topic 的消息并進(jìn)行消費(fèi)。
• Partition：Topic 是一個(gè)邏輯的概念，它可以細(xì)分為多個(gè)分區(qū)，每個(gè)分區(qū)只屬于單個(gè)主題。同一個(gè)主題下不同分區(qū)包含的消息是不同的，分區(qū)在存儲(chǔ)層面可以看作一個(gè)可追加的日志（Log）文件，消息在被追加到分區(qū)日志文件的時(shí)候都會(huì)分配一個(gè)特定的偏移量（offset）。
• Offset：offset 是消息在分區(qū)中的唯一標(biāo)識(shí)，Kafka 通過它來保證消息在分區(qū)內(nèi)的順序性，不過 offset 并不跨越分區(qū)，也就是說，Kafka 保證的是分區(qū)有序性而不是主題有序性。
• Replication：副本，是 Kafka 保證數(shù)據(jù)高可用的方式，Kafka 同一 Partition 的數(shù)據(jù)可以在多 Broker 上存在多個(gè)副本，通常只有主副本對(duì)外提供讀寫服務(wù)，當(dāng)主副本所在 broker 崩潰或發(fā)生網(wǎng)絡(luò)一場(chǎng)，Kafka 會(huì)在 Controller 的管理下會(huì)重新選擇新的 Leader 副本對(duì)外提供讀寫服務(wù)。
• Record：實(shí)際寫入 Kafka 中并可以被讀取的消息記錄。每個(gè) record 包含了 key、value 和 timestamp。

Kafka Topic Partitions Layout

Kafka 將 Topic 進(jìn)行分區(qū)，分區(qū)可以并發(fā)讀寫。

Kafka Consumer Offset

zookeeper

• Broker 注冊(cè)：Broker 是分布式部署并且之間相互獨(dú)立，Zookeeper 用來管理注冊(cè)到集群的所有 Broker 節(jié)點(diǎn)。
• Topic 注冊(cè)：在 Kafka 中，同一個(gè) Topic 的消息會(huì)被分成多個(gè)分區(qū)并將其分布在多個(gè) Broker 上，這些分區(qū)信息及與 Broker 的對(duì)應(yīng)關(guān)系也都是由 Zookeeper 在維護(hù)
• 生產(chǎn)者負(fù)載均衡：由于同一個(gè) Topic 消息會(huì)被分區(qū)并將其分布在多個(gè) Broker 上，因此，生產(chǎn)者需要將消息合理地發(fā)送到這些分布式的 Broker 上。
• 消費(fèi)者負(fù)載均衡：與生產(chǎn)者類似，Kafka 中的消費(fèi)者同樣需要進(jìn)行負(fù)載均衡來實(shí)現(xiàn)多個(gè)消費(fèi)者合理地從對(duì)應(yīng)的 Broker 服務(wù)器上接收消息，每個(gè)消費(fèi)者分組包含若干消費(fèi)者，每條消息都只會(huì)發(fā)送給分組中的一個(gè)消費(fèi)者，不同的消費(fèi)者分組消費(fèi)自己特定的 Topic 下面的消息，互不干擾。

答案關(guān)鍵字

• 簡(jiǎn)單講下 Kafka 的架構(gòu)？

  Producer、Consumer、Consumer Group、Topic、Partition

• Kafka 是推模式還是拉模式，推拉的區(qū)別是什么？

  Kafka Producer 向 Broker 發(fā)送消息使用 Push 模式，Consumer 消費(fèi)采用的 Pull 模式。拉取模式，讓 consumer 自己管理 offset，可以提供讀取性能

• Kafka 如何廣播消息？

  Consumer group

• Kafka 的消息是否是有序的？

  Topic 級(jí)別無序，Partition 有序

• Kafka 是否支持讀寫分離？

  不支持，只有 Leader 對(duì)外提供讀寫服務(wù)

• Kafka 如何保證數(shù)據(jù)高可用？

  副本，ack，HW

• Kafka 中 zookeeper 的作用？

  集群管理，元數(shù)據(jù)管理

• 是否支持事務(wù)？

  0.11 后支持事務(wù)，可以實(shí)現(xiàn)”exactly once“

• 分區(qū)數(shù)是否可以減少？

  不可以，會(huì)丟失數(shù)據(jù)

Kafka 使用

問題

• Kafka 有哪些命令行工具？你用過哪些？
• Kafka Producer 的執(zhí)行過程？
• Kafka Producer 有哪些常見配置？
• 如何讓 Kafka 的消息有序？
• Producer 如何保證數(shù)據(jù)發(fā)送不丟失？
• 如何提升 Producer 的性能？
• 如果同一 group 下 consumer 的數(shù)量大于 part 的數(shù)量，kafka 如何處理？
• Kafka Consumer 是否是線程安全的？
• 講一下你使用 Kafka Consumer 消費(fèi)消息時(shí)的線程模型，為何如此設(shè)計(jì)？
• Kafka Consumer 的常見配置？
• Consumer 什么時(shí)候會(huì)被踢出集群？
• 當(dāng)有 Consumer 加入或退出時(shí)，Kafka 會(huì)作何反應(yīng)？
• 什么是 Rebalance，何時(shí)會(huì)發(fā)生 Rebalance？

命令行工具

Kafka 的命令行工具在 Kafka 包的/bin目錄下，主要包括服務(wù)和集群管理腳本，配置腳本，信息查看腳本，Topic 腳本，客戶端腳本等。

• kafka-configs.sh：配置管理腳本
• kafka-console-consumer.sh：kafka 消費(fèi)者控制臺(tái)
• kafka-console-producer.sh：kafka 生產(chǎn)者控制臺(tái)
• kafka-consumer-groups.sh：kafka 消費(fèi)者組相關(guān)信息
• kafka-delete-records.sh：刪除低水位的日志文件
• kafka-log-dirs.sh：kafka 消息日志目錄信息
• kafka-mirror-maker.sh：不同數(shù)據(jù)中心 kafka 集群復(fù)制工具
• kafka-preferred-replica-election.sh：觸發(fā) preferred replica 選舉
• kafka-producer-perf-test.sh：kafka 生產(chǎn)者性能測(cè)試腳本
• kafka-reassign-partitions.sh：分區(qū)重分配腳本
• kafka-replica-verification.sh：復(fù)制進(jìn)度驗(yàn)證腳本
• kafka-server-start.sh：?jiǎn)?dòng) kafka 服務(wù)
• kafka-server-stop.sh：停止 kafka 服務(wù)
• kafka-topics.sh：topic 管理腳本
• kafka-verifiable-consumer.sh：可檢驗(yàn)的 kafka 消費(fèi)者
• kafka-verifiable-producer.sh：可檢驗(yàn)的 kafka 生產(chǎn)者
• zookeeper-server-start.sh：?jiǎn)?dòng) zk 服務(wù)
• zookeeper-server-stop.sh：停止 zk 服務(wù)
• zookeeper-shell.sh：zk 客戶端

我們通?？梢允褂?code style="">kafka-console-consumer.sh和kafka-console-producer.sh腳本來測(cè)試 Kafka 生產(chǎn)和消費(fèi)，kafka-consumer-groups.sh可以查看和管理集群中的 Topic，kafka-topics.sh通常用于查看 Kafka 的消費(fèi)組情況。

Kafka Producer

Kafka producer 的正常生產(chǎn)邏輯包含以下幾個(gè)步驟：

1. 配置生產(chǎn)者客戶端參數(shù)常見生產(chǎn)者實(shí)例。
2. 構(gòu)建待發(fā)送的消息。
3. 發(fā)送消息。
4. 關(guān)閉生產(chǎn)者實(shí)例。

Producer 發(fā)送消息的過程如下圖所示，需要經(jīng)過攔截器，序列化器和分區(qū)器，最終由累加器批量發(fā)送至 Broker。

Kafka Producer 需要以下必要參數(shù)：

• bootstrap.server：指定 Kafka 的 Broker 的地址
• key.serializer：key 序列化器
• value.serializer：value 序列化器

常見參數(shù)：

• batch.num.messages

  默認(rèn)值：200，每次批量消息的數(shù)量，只對(duì) asyc 起作用。

• request.required.acks

  默認(rèn)值：0，0 表示 producer 毋須等待 leader 的確認(rèn)，1 代表需要 leader 確認(rèn)寫入它的本地 log 并立即確認(rèn)，-1 代表所有的備份都完成后確認(rèn)。只對(duì) async 模式起作用，這個(gè)參數(shù)的調(diào)整是數(shù)據(jù)不丟失和發(fā)送效率的 tradeoff，如果對(duì)數(shù)據(jù)丟失不敏感而在乎效率的場(chǎng)景可以考慮設(shè)置為 0，這樣可以大大提高 producer 發(fā)送數(shù)據(jù)的效率。

• request.timeout.ms

  默認(rèn)值：10000，確認(rèn)超時(shí)時(shí)間。

• partitioner.class

  默認(rèn)值：kafka.producer.DefaultPartitioner，必須實(shí)現(xiàn) kafka.producer.Partitioner，根據(jù) Key 提供一個(gè)分區(qū)策略。有時(shí)候我們需要相同類型的消息必須順序處理，這樣我們就必須自定義分配策略，從而將相同類型的數(shù)據(jù)分配到同一個(gè)分區(qū)中。

• producer.type

  默認(rèn)值：sync，指定消息發(fā)送是同步還是異步。異步 asyc 成批發(fā)送用 kafka.producer.AyncProducer， 同步 sync 用 kafka.producer.SyncProducer。同步和異步發(fā)送也會(huì)影響消息生產(chǎn)的效率。

• compression.topic

  默認(rèn)值：none，消息壓縮，默認(rèn)不壓縮。其余壓縮方式還有，"gzip"、"snappy"和"lz4"。對(duì)消息的壓縮可以極大地減少網(wǎng)絡(luò)傳輸量、降低網(wǎng)絡(luò) IO，從而提高整體性能。

• compressed.topics

  默認(rèn)值：null，在設(shè)置了壓縮的情況下，可以指定特定的 topic 壓縮，未指定則全部壓縮。

• message.send.max.retries

  默認(rèn)值：3，消息發(fā)送最大嘗試次數(shù)。

• retry.backoff.ms

  默認(rèn)值：300，每次嘗試增加的額外的間隔時(shí)間。

• topic.metadata.refresh.interval.ms

  默認(rèn)值：600000，定期的獲取元數(shù)據(jù)的時(shí)間。當(dāng)分區(qū)丟失，leader 不可用時(shí) producer 也會(huì)主動(dòng)獲取元數(shù)據(jù)，如果為 0，則每次發(fā)送完消息就獲取元數(shù)據(jù)，不推薦。如果為負(fù)值，則只有在失敗的情況下獲取元數(shù)據(jù)。

• queue.buffering.max.ms

  默認(rèn)值：5000，在 producer queue 的緩存的數(shù)據(jù)最大時(shí)間，僅僅 for asyc。

• queue.buffering.max.message

  默認(rèn)值：10000，producer 緩存的消息的最大數(shù)量，僅僅 for asyc。

• queue.enqueue.timeout.ms

  默認(rèn)值：-1，0 當(dāng) queue 滿時(shí)丟掉，負(fù)值是 queue 滿時(shí) block, 正值是 queue 滿時(shí) block 相應(yīng)的時(shí)間，僅僅 for asyc。

Kafka Consumer

Kafka 有消費(fèi)組的概念，每個(gè)消費(fèi)者只能消費(fèi)所分配到的分區(qū)的消息，每一個(gè)分區(qū)只能被一個(gè)消費(fèi)組中的一個(gè)消費(fèi)者所消費(fèi)，所以同一個(gè)消費(fèi)組中消費(fèi)者的數(shù)量如果超過了分區(qū)的數(shù)量，將會(huì)出現(xiàn)有些消費(fèi)者分配不到消費(fèi)的分區(qū)。消費(fèi)組與消費(fèi)者關(guān)系如下圖所示：

Kafka Consumer Client 消費(fèi)消息通常包含以下步驟：

1. 配置客戶端，創(chuàng)建消費(fèi)者
2. 訂閱主題
3. 拉去消息并消費(fèi)
4. 提交消費(fèi)位移
5. 關(guān)閉消費(fèi)者實(shí)例

因?yàn)?Kafka 的 Consumer 客戶端是線程不安全的，為了保證線程安全，并提升消費(fèi)性能，可以在 Consumer 端采用類似 Reactor 的線程模型來消費(fèi)數(shù)據(jù)。

Kafka consumer 參數(shù)

• bootstrap.servers：連接 broker 地址，host：port 格式。
• group.id：消費(fèi)者隸屬的消費(fèi)組。
• key.deserializer：與生產(chǎn)者的key.serializer對(duì)應(yīng)，key 的反序列化方式。
• value.deserializer：與生產(chǎn)者的value.serializer對(duì)應(yīng)，value 的反序列化方式。
• session.timeout.ms：coordinator 檢測(cè)失敗的時(shí)間。默認(rèn) 10s 該參數(shù)是 Consumer Group 主動(dòng)檢測(cè) （組內(nèi)成員 comsummer) 崩潰的時(shí)間間隔，類似于心跳過期時(shí)間。
• auto.offset.reset：該屬性指定了消費(fèi)者在讀取一個(gè)沒有偏移量后者偏移量無效（消費(fèi)者長(zhǎng)時(shí)間失效當(dāng)前的偏移量已經(jīng)過時(shí)并且被刪除了）的分區(qū)的情況下，應(yīng)該作何處理，默認(rèn)值是 latest，也就是從最新記錄讀取數(shù)據(jù)（消費(fèi)者啟動(dòng)之后生成的記錄），另一個(gè)值是 earliest，意思是在偏移量無效的情況下，消費(fèi)者從起始位置開始讀取數(shù)據(jù)。
• enable.auto.commit：否自動(dòng)提交位移，如果為false，則需要在程序中手動(dòng)提交位移。對(duì)于精確到一次的語義，最好手動(dòng)提交位移
• fetch.max.bytes：?jiǎn)未卫?shù)據(jù)的最大字節(jié)數(shù)量
• max.poll.records：?jiǎn)未?poll 調(diào)用返回的最大消息數(shù)，如果處理邏輯很輕量，可以適當(dāng)提高該值。但是max.poll.records條數(shù)據(jù)需要在在 session.timeout.ms 這個(gè)時(shí)間內(nèi)處理完 。默認(rèn)值為 500
• request.timeout.ms：一次請(qǐng)求響應(yīng)的最長(zhǎng)等待時(shí)間。如果在超時(shí)時(shí)間內(nèi)未得到響應(yīng)，kafka 要么重發(fā)這條消息，要么超過重試次數(shù)的情況下直接置為失敗。

Kafka Rebalance

rebalance 本質(zhì)上是一種協(xié)議，規(guī)定了一個(gè) consumer group 下的所有 consumer 如何達(dá)成一致來分配訂閱 topic 的每個(gè)分區(qū)。比如某個(gè) group 下有 20 個(gè) consumer，它訂閱了一個(gè)具有 100 個(gè)分區(qū)的 topic。正常情況下，Kafka 平均會(huì)為每個(gè) consumer 分配 5 個(gè)分區(qū)。這個(gè)分配的過程就叫 rebalance。

什么時(shí)候 rebalance？

這也是經(jīng)常被提及的一個(gè)問題。rebalance 的觸發(fā)條件有三種：

• 組成員發(fā)生變更（新 consumer 加入組、已有 consumer 主動(dòng)離開組或已有 consumer 崩潰了——這兩者的區(qū)別后面會(huì)談到）
• 訂閱主題數(shù)發(fā)生變更
• 訂閱主題的分區(qū)數(shù)發(fā)生變更

如何進(jìn)行組內(nèi)分區(qū)分配？

Kafka 默認(rèn)提供了兩種分配策略：Range 和 Round-Robin。當(dāng)然 Kafka 采用了可插拔式的分配策略，你可以創(chuàng)建自己的分配器以實(shí)現(xiàn)不同的分配策略。

答案關(guān)鍵字

• Kafka 有哪些命令行工具？你用過哪些？/bin目錄，管理 kafka 集群、管理 topic、生產(chǎn)和消費(fèi) kafka
• Kafka Producer 的執(zhí)行過程？攔截器，序列化器，分區(qū)器和累加器
• Kafka Producer 有哪些常見配置？broker 配置，ack 配置，網(wǎng)絡(luò)和發(fā)送參數(shù)，壓縮參數(shù)，ack 參數(shù)
• 如何讓 Kafka 的消息有序？Kafka 在 Topic 級(jí)別本身是無序的，只有 partition 上才有序，所以為了保證處理順序，可以自定義分區(qū)器，將需順序處理的數(shù)據(jù)發(fā)送到同一個(gè) partition
• Producer 如何保證數(shù)據(jù)發(fā)送不丟失？ack 機(jī)制，重試機(jī)制
• 如何提升 Producer 的性能？批量，異步，壓縮
• 如果同一 group 下 consumer 的數(shù)量大于 part 的數(shù)量，kafka 如何處理？多余的 Part 將處于無用狀態(tài)，不消費(fèi)數(shù)據(jù)
• Kafka Consumer 是否是線程安全的？不安全，單線程消費(fèi)，多線程處理
• 講一下你使用 Kafka Consumer 消費(fèi)消息時(shí)的線程模型，為何如此設(shè)計(jì)？拉取和處理分離
• Kafka Consumer 的常見配置？broker, 網(wǎng)絡(luò)和拉取參數(shù)，心跳參數(shù)
• Consumer 什么時(shí)候會(huì)被踢出集群？奔潰，網(wǎng)絡(luò)異常，處理時(shí)間過長(zhǎng)提交位移超時(shí)
• 當(dāng)有 Consumer 加入或退出時(shí)，Kafka 會(huì)作何反應(yīng)？進(jìn)行 Rebalance
• 什么是 Rebalance，何時(shí)會(huì)發(fā)生 Rebalance？topic 變化，consumer 變化

高可用和性能

問題

• Kafka 如何保證高可用？
• Kafka 的交付語義？
• Replic 的作用？
• 什么事 AR，ISR？
• Leader 和 Flower 是什么？
• Kafka 中的 HW、LEO、LSO、LW 等分別代表什么？
• Kafka 為保證優(yōu)越的性能做了哪些處理？

分區(qū)與副本

在分布式數(shù)據(jù)系統(tǒng)中，通常使用分區(qū)來提高系統(tǒng)的處理能力，通過副本來保證數(shù)據(jù)的高可用性。多分區(qū)意味著并發(fā)處理的能力，這多個(gè)副本中，只有一個(gè)是 leader，而其他的都是 follower 副本。僅有 leader 副本可以對(duì)外提供服務(wù)。多個(gè) follower 副本通常存放在和 leader 副本不同的 broker 中。通過這樣的機(jī)制實(shí)現(xiàn)了高可用，當(dāng)某臺(tái)機(jī)器掛掉后，其他 follower 副本也能迅速”轉(zhuǎn)正“，開始對(duì)外提供服務(wù)。

為什么 follower 副本不提供讀服務(wù)？

這個(gè)問題本質(zhì)上是對(duì)性能和一致性的取舍。試想一下，如果 follower 副本也對(duì)外提供服務(wù)那會(huì)怎么樣呢？首先，性能是肯定會(huì)有所提升的。但同時(shí)，會(huì)出現(xiàn)一系列問題。類似數(shù)據(jù)庫事務(wù)中的幻讀，臟讀。比如你現(xiàn)在寫入一條數(shù)據(jù)到 kafka 主題 a，消費(fèi)者 b 從主題 a 消費(fèi)數(shù)據(jù)，卻發(fā)現(xiàn)消費(fèi)不到，因?yàn)橄M(fèi)者 b 去讀取的那個(gè)分區(qū)副本中，最新消息還沒寫入。而這個(gè)時(shí)候，另一個(gè)消費(fèi)者 c 卻可以消費(fèi)到最新那條數(shù)據(jù)，因?yàn)樗M(fèi)了 leader 副本。Kafka 通過 WH 和 Offset 的管理來決定 Consumer 可以消費(fèi)哪些數(shù)據(jù)，已經(jīng)當(dāng)前寫入的數(shù)據(jù)。

只有 Leader 可以對(duì)外提供讀服務(wù)，那如何選舉 Leader

kafka 會(huì)將與 leader 副本保持同步的副本放到 ISR 副本集合中。當(dāng)然，leader 副本是一直存在于 ISR 副本集合中的，在某些特殊情況下，ISR 副本中甚至只有 leader 一個(gè)副本。當(dāng) leader 掛掉時(shí)，kakfa 通過 zookeeper 感知到這一情況，在 ISR 副本中選取新的副本成為 leader，對(duì)外提供服務(wù)。但這樣還有一個(gè)問題，前面提到過，有可能 ISR 副本集合中，只有 leader，當(dāng) leader 副本掛掉后，ISR 集合就為空，這時(shí)候怎么辦呢？這時(shí)候如果設(shè)置 unclean.leader.election.enable 參數(shù)為 true，那么 kafka 會(huì)在非同步，也就是不在 ISR 副本集合中的副本中，選取出副本成為 leader。

副本的存在就會(huì)出現(xiàn)副本同步問題

Kafka 在所有分配的副本 (AR) 中維護(hù)一個(gè)可用的副本列表 (ISR)，Producer 向 Broker 發(fā)送消息時(shí)會(huì)根據(jù)ack配置來確定需要等待幾個(gè)副本已經(jīng)同步了消息才相應(yīng)成功，Broker 內(nèi)部會(huì)ReplicaManager服務(wù)來管理 flower 與 leader 之間的數(shù)據(jù)同步。

性能優(yōu)化

• partition 并發(fā)
• 順序讀寫磁盤
• page cache：按頁讀寫
• 預(yù)讀：Kafka 會(huì)將將要消費(fèi)的消息提前讀入內(nèi)存
• 高性能序列化（二進(jìn)制）
• 內(nèi)存映射
• 無鎖 offset 管理：提高并發(fā)能力
• Java NIO 模型
• 批量：批量讀寫
• 壓縮：消息壓縮，存儲(chǔ)壓縮，減小網(wǎng)絡(luò)和 IO 開銷

Partition 并發(fā)

一方面，由于不同 Partition 可位于不同機(jī)器，因此可以充分利用集群優(yōu)勢(shì)，實(shí)現(xiàn)機(jī)器間的并行處理。另一方面，由于 Partition 在物理上對(duì)應(yīng)一個(gè)文件夾，即使多個(gè) Partition 位于同一個(gè)節(jié)點(diǎn)，也可通過配置讓同一節(jié)點(diǎn)上的不同 Partition 置于不同的 disk drive 上，從而實(shí)現(xiàn)磁盤間的并行處理，充分發(fā)揮多磁盤的優(yōu)勢(shì)。

順序讀寫

Kafka 每一個(gè) partition 目錄下的文件被平均切割成大小相等（默認(rèn)一個(gè)文件是 500 兆，可以手動(dòng)去設(shè)置）的數(shù)據(jù)文件， 每一個(gè)數(shù)據(jù)文件都被稱為一個(gè)段（segment file）, 每個(gè) segment 都采用 append 的方式追加數(shù)據(jù)。

答案關(guān)鍵字

• Kafka 如何保證高可用？

  通過副本來保證數(shù)據(jù)的高可用，producer ack、重試、自動(dòng) Leader 選舉，Consumer 自平衡

• Kafka 的交付語義？

  交付語義一般有at least once、at most once和exactly once。kafka 通過 ack 的配置來實(shí)現(xiàn)前兩種。

• Replic 的作用？

  實(shí)現(xiàn)數(shù)據(jù)的高可用

• 什么是 AR，ISR？

  AR：Assigned Replicas。AR 是主題被創(chuàng)建后，分區(qū)創(chuàng)建時(shí)被分配的副本集合，副本個(gè) 數(shù)由副本因子決定。ISR：In-Sync Replicas。Kafka 中特別重要的概念，指代的是 AR 中那些與 Leader 保 持同步的副本集合。在 AR 中的副本可能不在 ISR 中，但 Leader 副本天然就包含在 ISR 中。關(guān)于 ISR，還有一個(gè)常見的面試題目是如何判斷副本是否應(yīng)該屬于 ISR。目前的判斷 依據(jù)是：Follower 副本的 LEO 落后 Leader LEO 的時(shí)間，是否超過了 Broker 端參數(shù) replica.lag.time.max.ms 值。如果超過了，副本就會(huì)被從 ISR 中移除。

• Leader 和 Flower 是什么？

• Kafka 中的 HW 代表什么？

  高水位值 (High watermark)。這是控制消費(fèi)者可讀取消息范圍的重要字段。一 個(gè)普通消費(fèi)者只能“看到”Leader 副本上介于 Log Start Offset 和 HW（不含）之間的 所有消息。水位以上的消息是對(duì)消費(fèi)者不可見的。

• Kafka 為保證優(yōu)越的性能做了哪些處理？

  partition 并發(fā)、順序讀寫磁盤、page cache 壓縮、高性能序列化（二進(jìn)制）、內(nèi)存映射 無鎖 offset 管理、Java NIO 模型

建議讀者同學(xué)結(jié)合 Kafka 的配置去了解 Kafka 的實(shí)現(xiàn)原理，Kafka 有大量的配置，這也是 Kafka 高度擴(kuò)展的一個(gè)表現(xiàn)，很多同學(xué)對(duì) Kafka 的配置也不敢輕易改動(dòng)。所以理解這些配置背后的實(shí)現(xiàn)原理，可以讓我們?cè)趯?shí)踐中懂得如何使用和優(yōu)化 Kafka。既可面試造火箭，也可以實(shí)戰(zhàn)造火箭。

Kafka 配置說明鏈接：https://kafka.apache.org/documentation

Architecture

理解 Kafka 架構(gòu)，就是理解 Kafka 的各種組件的概念，以及這些組件的關(guān)系。先簡(jiǎn)單看一下各組件及其簡(jiǎn)單說明。

不要去嘗試記憶他們

Producer： 生產(chǎn)者，發(fā)送消息的一方。生產(chǎn)者負(fù)責(zé)創(chuàng)建消息，然后將其發(fā)送到 Kafka。
Consumer： 消費(fèi)者，接受消息的一方。消費(fèi)者連接到 Kafka 上并接收消息，進(jìn)而進(jìn)行相應(yīng)的業(yè)務(wù)邏輯處理。
Consumer Group： 一個(gè)消費(fèi)者組可以包含一個(gè)或多個(gè)消費(fèi)者。使用多分區(qū) + 多消費(fèi)者方式可以極大提高數(shù)據(jù)下游的處理速度，同一消費(fèi)組中的消費(fèi)者不會(huì)重復(fù)消費(fèi)消息，同樣的，不同消費(fèi)組中的消費(fèi)者消息消息時(shí)互不影響。Kafka 就是通過消費(fèi)組的方式來實(shí)現(xiàn)消息 P2P 模式和廣播模式。
Broker： 服務(wù)代理節(jié)點(diǎn)。Broker 是 Kafka 的服務(wù)節(jié)點(diǎn)，即 Kafka 的服務(wù)器。
Topic： Kafka 中的消息以 Topic 為單位進(jìn)行劃分，生產(chǎn)者將消息發(fā)送到特定的 Topic，而消費(fèi)者負(fù)責(zé)訂閱 Topic 的消息并進(jìn)行消費(fèi)。
Partition： Topic 是一個(gè)邏輯的概念，它可以細(xì)分為多個(gè)分區(qū)，每個(gè)分區(qū)只屬于單個(gè)主題。同一個(gè)主題下不同分區(qū)包含的消息是不同的，分區(qū)在存儲(chǔ)層面可以看作一個(gè)可追加的日志（Log）文件，消息在被追加到分區(qū)日志文件的時(shí)候都會(huì)分配一個(gè)特定的偏移量（offset）。
Offset： offset 是消息在分區(qū)中的唯一標(biāo)識(shí)，Kafka 通過它來保證消息在分區(qū)內(nèi)的順序性，不過 offset 并不跨越分區(qū)，也就是說，Kafka 保證的是分區(qū)有序性而不是主題有序性。
Replication： 副本，是 Kafka 保證數(shù)據(jù)高可用的方式，Kafka 同一 Partition 的數(shù)據(jù)可以在多 Broker 上存在多個(gè)副本，通常只有主副本對(duì)外提供讀寫服務(wù)，當(dāng)主副本所在 broker 崩潰或發(fā)生網(wǎng)絡(luò)異常，Kafka 會(huì)在 Controller 的管理下會(huì)重新選擇新的 Leader 副本對(duì)外提供讀寫服務(wù)。
Record： 實(shí)際寫入 Kafka 中并可以被讀取的消息記錄。每個(gè) record 包含了 key、value 和 timestamp。

我們理解了也就自然記住了

我們應(yīng)該通過理解的方式去記憶它們。

生產(chǎn)者-消費(fèi)者

生產(chǎn)者-消費(fèi)者是一種設(shè)計(jì)模式，生產(chǎn)者和消費(fèi)者之間通過添加一個(gè)中間組件來達(dá)到解耦。生產(chǎn)者向中間組件生成數(shù)據(jù)，消費(fèi)者消費(fèi)數(shù)據(jù)。

就像讀書時(shí)65 哥給小芳寫情書，這里寫情書就是生產(chǎn)者，情書就是消息，小芳就是消費(fèi)者。但有時(shí)候小芳不在，或者比較忙，65 哥也比較害羞，不敢直接將情書塞小芳手里，于是將情書塞在小芳抽屜中。所以抽屜就是這個(gè)中間組件。

在程序中我們通常使用Queue來作為這個(gè)中間組件?？梢允褂枚嗑€程向隊(duì)列中寫入數(shù)據(jù)，另外的消費(fèi)者線程依次讀取隊(duì)列中的數(shù)據(jù)進(jìn)行消費(fèi)。模型如下圖所示：

生產(chǎn)者-消費(fèi)者模式通過添加一個(gè)中間層，不僅可以解耦生產(chǎn)者和消費(fèi)者，使其易于擴(kuò)展，還可以異步化調(diào)用、緩沖消息等。

分布式隊(duì)列

后來 65 哥和小芳異地了，65 哥在卷都奮斗，小芳在魔都逛街。于是只能通過郵局寄曖昧信了。這樣 65 哥、郵局和小芳就成了分布式的了。65 哥將信件發(fā)給郵局，小芳從郵局拿到 65 哥寫的信，再回去慢慢看。

Kafka 的消息生產(chǎn)者就是Producer，上游消費(fèi)者進(jìn)程添加 Kafka Client 創(chuàng)建 Kafka Producer，向 Broker 發(fā)送消息，Broker 是集群部署在遠(yuǎn)程服務(wù)器上的 Kafka Server 進(jìn)程，下游消費(fèi)者進(jìn)程引入 Kafka Consumer API 持續(xù)消費(fèi)隊(duì)列中消息。

因?yàn)?Kafka Consumer 使用 Poll 的模式，需要 Consumer 主動(dòng)拉去消息。所有小芳只能定期去郵局拿信件了(呃，果然主動(dòng)權(quán)都在小芳手上啊)。

主題

郵局不能只為 65 哥服務(wù)，雖然 65 哥一天寫好幾封信。但也無法挽回郵局的損失。所以郵局是可以供任何人寄信。只需要寄信人寫好地址(主題)，郵局建有兩地的通道就可以發(fā)收信件了。

Kafka 的 Topic 才相當(dāng)于一個(gè)隊(duì)列，Broker 是所有隊(duì)列部署的機(jī)器?？梢园礃I(yè)務(wù)創(chuàng)建不同的 Topic，Producer 向所屬業(yè)務(wù)的 Topic 發(fā)送消息，相應(yīng)的 Consumer 可以消費(fèi)并處理消息。

分區(qū)

由于 65 哥寫的信太多，一個(gè)郵局已經(jīng)無法滿足 65 哥的需求，郵政公司只能多建幾個(gè)郵局了，65 哥將信件按私密度分類(分區(qū)策略)，從不同的郵局寄送。

同一個(gè) Topic 可以創(chuàng)建多個(gè)分區(qū)。理論上分區(qū)越多并發(fā)度越高，Kafka 會(huì)根據(jù)分區(qū)策略將分區(qū)盡可能均衡的分布在不同的 Broker 節(jié)點(diǎn)上，以避免消息傾斜，不同的 Broker 負(fù)載差異太大。分區(qū)也不是越多越好哦，畢竟太多郵政公司也管理不過來。

副本

為防止由于郵局的問題，比如交通斷啦，郵車沒油啦。導(dǎo)致 65 哥的曖昧信無法寄到小芳手上，使得 65 哥晚上遠(yuǎn)程跪鍵盤。郵局決定將 65 哥的信件復(fù)制幾份發(fā)到多個(gè)正常的郵局，這樣只要有一個(gè)郵局還在，小芳就可以收到 65 哥的信了。

Kafka 采用分區(qū)副本的方式來保證數(shù)據(jù)的高可用，每個(gè)分區(qū)都將建立指定數(shù)量的副本數(shù)，kakfa 保證同一分區(qū)副本盡量分布在不同的 Broker 節(jié)點(diǎn)上，以防止 Broker 宕機(jī)導(dǎo)致所有副本不可用。Kafka 會(huì)為分區(qū)的多個(gè)副本選舉一個(gè)作為主副本(Leader)，主副本對(duì)外提供讀寫服務(wù)，從副本(Follower)實(shí)時(shí)同步 Leader 的數(shù)據(jù)。

多消費(fèi)者

哎，65 哥的信件滿天飛，小芳天天跑郵局，還要一一拆開看，65 哥寫的信又臭又長(zhǎng)，讓小芳忙得滿身大漢大汗。于是小芳啪的一下，很快啊，變出多個(gè)分身去不同的郵局取信，這樣小芳終于可以擠出額外的時(shí)間逛街了。

廣播消息

郵局最近提供了定制明信片業(yè)務(wù)，每個(gè)人都可以設(shè)計(jì)明信片，同一個(gè)身份只能領(lǐng)取一種明信片。65 哥設(shè)計(jì)了一堆，廣播給所有漂亮的小妹妹都可以來領(lǐng)取，美女啪變出的分身也可以來領(lǐng)取，但是同一個(gè)身份的多個(gè)分身只能取一種明信片。

Kafka 通過 Consumer Group 來實(shí)現(xiàn)廣播模式消息訂閱，即不同 group 下的 consumer 可以重復(fù)消費(fèi)消息，相互不影響，同一個(gè) group 下的 consumer 構(gòu)成一個(gè)整體。

最后我們完成了 Kafka 的整體架構(gòu)，如下：

Zookeeper

Zookeeper 是一個(gè)成熟的分布式協(xié)調(diào)服務(wù)，它可以為分布式服務(wù)提供分布式配置服、同步服務(wù)和命名注冊(cè)等能力.。對(duì)于任何分布式系統(tǒng)，都需要一種協(xié)調(diào)任務(wù)的方法。Kafka 是使用 ZooKeeper 而構(gòu)建的分布式系統(tǒng)。但是也有一些其他技術(shù)（例如 Elasticsearch 和 MongoDB）具有其自己的內(nèi)置任務(wù)協(xié)調(diào)機(jī)制。

Kafka 將 Broker、Topic 和 Partition 的元數(shù)據(jù)信息存儲(chǔ)在 Zookeeper 上。通過在 Zookeeper 上建立相應(yīng)的數(shù)據(jù)節(jié)點(diǎn)，并監(jiān)聽節(jié)點(diǎn)的變化，Kafka 使用 Zookeeper 完成以下功能：

• Kafka Controller 的 Leader 選舉
• Kafka 集群成員管理
• Topic 配置管理
• 分區(qū)副本管理

我們看一看 Zookeeper 下 Kafka 創(chuàng)建的節(jié)點(diǎn)，即可一目了然的看出這些相關(guān)的功能。

Controller

Controller 是從 Broker 中選舉出來的，負(fù)責(zé)分區(qū) Leader 和 Follower 的管理。當(dāng)某個(gè)分區(qū)的 leader 副本發(fā)生故障時(shí)，由 Controller 負(fù)責(zé)為該分區(qū)選舉新的 leader 副本。當(dāng)檢測(cè)到某個(gè)分區(qū)的 ISR(In-Sync Replica)集合發(fā)生變化時(shí)，由控制器負(fù)責(zé)通知所有 broker 更新其元數(shù)據(jù)信息。當(dāng)使用kafka-topics.sh腳本為某個(gè) topic 增加分區(qū)數(shù)量時(shí)，同樣還是由控制器負(fù)責(zé)分區(qū)的重新分配。

Kafka 中 Contorller 的選舉的工作依賴于 Zookeeper，成功競(jìng)選為控制器的 broker 會(huì)在 Zookeeper 中創(chuàng)建/controller這個(gè)臨時(shí)（EPHEMERAL）節(jié)點(diǎn)。

選舉過程

Broker 啟動(dòng)的時(shí)候嘗試去讀取/controller節(jié)點(diǎn)的brokerid的值，如果brokerid的值不等于-1，則表明已經(jīng)有其他的 Broker 成功成為 Controller 節(jié)點(diǎn)，當(dāng)前 Broker 主動(dòng)放棄競(jìng)選；如果不存在/controller節(jié)點(diǎn)，或者 brokerid 數(shù)值異常，當(dāng)前 Broker 嘗試去創(chuàng)建/controller這個(gè)節(jié)點(diǎn)，此時(shí)也有可能其他 broker 同時(shí)去嘗試創(chuàng)建這個(gè)節(jié)點(diǎn)，只有創(chuàng)建成功的那個(gè) broker 才會(huì)成為控制器，而創(chuàng)建失敗的 broker 則表示競(jìng)選失敗。每個(gè) broker 都會(huì)在內(nèi)存中保存當(dāng)前控制器的 brokerid 值，這個(gè)值可以標(biāo)識(shí)為 activeControllerId。

實(shí)現(xiàn)

Controller 讀取 Zookeeper 中的節(jié)點(diǎn)數(shù)據(jù)，初始化上下文(Controller Context)，并管理節(jié)點(diǎn)變化，變更上下文，同時(shí)也需要將這些變更信息同步到其他普通的 broker 節(jié)點(diǎn)中。Controller 通過定時(shí)任務(wù)，或者監(jiān)聽器模式獲取 zookeeper 信息，事件監(jiān)聽會(huì)更新更新上下文信息，如圖所示，Controller 內(nèi)部也采用生產(chǎn)者-消費(fèi)者實(shí)現(xiàn)模式，Controller 將 zookeeper 的變動(dòng)通過事件的方式發(fā)送給事件隊(duì)列，隊(duì)列就是一個(gè)LinkedBlockingQueue，事件消費(fèi)者線程組通過消費(fèi)消費(fèi)事件，將相應(yīng)的事件同步到各 Broker 節(jié)點(diǎn)。這種隊(duì)列 FIFO 的模式保證了消息的有序性。

職責(zé)

Controller 被選舉出來，作為整個(gè) Broker 集群的管理者，管理所有的集群信息和元數(shù)據(jù)信息。它的職責(zé)包括下面幾部分：

1. 處理 Broker 節(jié)點(diǎn)的上線和下線，包括自然下線、宕機(jī)和網(wǎng)絡(luò)不可達(dá)導(dǎo)致的集群變動(dòng)，Controller 需要及時(shí)更新集群元數(shù)據(jù)，并將集群變化通知到所有的 Broker 集群節(jié)點(diǎn)；
2. 創(chuàng)建 Topic 或者 Topic 擴(kuò)容分區(qū)，Controller 需要負(fù)責(zé)分區(qū)副本的分配工作，并主導(dǎo) Topic 分區(qū)副本的 Leader 選舉。
3. 管理集群中所有的副本和分區(qū)的狀態(tài)機(jī)，監(jiān)聽狀態(tài)機(jī)變化事件，并作出相應(yīng)的處理。Kafka 分區(qū)和副本數(shù)據(jù)采用狀態(tài)機(jī)的方式管理，分區(qū)和副本的變化都在狀態(tài)機(jī)內(nèi)會(huì)引起狀態(tài)機(jī)狀態(tài)的變更，從而觸發(fā)相應(yīng)的變化事件。

“

狀態(tài)機(jī)啊，聽起來好復(fù)雜。

”

Controller 管理著集群中所有副本和分區(qū)的狀態(tài)機(jī)。大家不要被狀態(tài)機(jī)這個(gè)詞唬住了。理解狀態(tài)機(jī)很簡(jiǎn)單。先理解模型，即這是什么關(guān)于什么模型，然后就是模型的狀態(tài)有哪些，模型狀態(tài)之間如何轉(zhuǎn)換，轉(zhuǎn)換時(shí)發(fā)送相應(yīng)的變化事件。

Kafka 的分區(qū)和副本狀態(tài)機(jī)很簡(jiǎn)單。我們先理解，這分別是管理 Kafka Topic 的分區(qū)和副本的。它們的狀態(tài)也很簡(jiǎn)單，就是 CRUD，具體說來如下：

分區(qū)狀態(tài)機(jī)

PartitionStateChange，管理 Topic 的分區(qū)，它有以下 4 種狀態(tài)：

1. NonExistentPartition：該狀態(tài)表示分區(qū)沒有被創(chuàng)建過或創(chuàng)建后被刪除了。
2. NewPartition：分區(qū)剛創(chuàng)建后，處于這個(gè)狀態(tài)。此狀態(tài)下分區(qū)已經(jīng)分配了副本，但是還沒有選舉 leader，也沒有 ISR 列表。
3. OnlinePartition：一旦這個(gè)分區(qū)的 leader 被選舉出來，將處于這個(gè)狀態(tài)。
4. OfflinePartition：當(dāng)分區(qū)的 leader 宕機(jī)，轉(zhuǎn)移到這個(gè)狀態(tài)。

我們用一張圖來直觀的看看這些狀態(tài)是如何變化的，以及在狀態(tài)發(fā)生變化時(shí) Controller 都有哪些操作：

副本狀態(tài)機(jī)

ReplicaStateChange，副本狀態(tài)，管理分區(qū)副本信息，它也有 4 種狀態(tài)：

1. NewReplica: 創(chuàng)建 topic 和分區(qū)分配后創(chuàng)建 replicas，此時(shí)，replica 只能獲取到成為 follower 狀態(tài)變化請(qǐng)求。
2. OnlineReplica: 當(dāng) replica 成為 parition 的 assingned replicas 時(shí)，其狀態(tài)變?yōu)?OnlineReplica, 即一個(gè)有效的 OnlineReplica。
3. OfflineReplica: 當(dāng)一個(gè) replica 下線，進(jìn)入此狀態(tài)，這一般發(fā)生在 broker 宕機(jī)的情況下；
4. NonExistentReplica: Replica 成功刪除后，replica 進(jìn)入 NonExistentReplica 狀態(tài)。

副本狀態(tài)間的變化如下圖所示，Controller 在狀態(tài)變化時(shí)會(huì)做出相應(yīng)的操作：

Network

Kafka 的網(wǎng)絡(luò)通信模型是基于 NIO 的 Reactor 多線程模型來設(shè)計(jì)的。其中包含了一個(gè)Acceptor線程，用于處理新的連接，Acceptor 有 N 個(gè) Processor 線程 select 和 read socket 請(qǐng)求，N 個(gè) Handler 線程處理請(qǐng)求并相應(yīng)，即處理業(yè)務(wù)邏輯。下面就是 KafkaServer 的模型圖：

Kafka 性能全景

從高度抽象的角度來看，性能問題逃不出下面三個(gè)方面：

• 網(wǎng)絡(luò)
• 磁盤
• 復(fù)雜度

對(duì)于 Kafka 這種網(wǎng)絡(luò)分布式隊(duì)列來說，網(wǎng)絡(luò)和磁盤更是優(yōu)化的重中之重。針對(duì)于上面提出的抽象問題，解決方案高度抽象出來也很簡(jiǎn)單：

• 并發(fā)
• 壓縮
• 批量
• 緩存
• 算法

知道了問題和思路，我們?cè)賮砜纯?，?Kafka 中，有哪些角色，而這些角色就是可以優(yōu)化的點(diǎn)：

• Producer
• Broker
• Consumer

是的，所有的問題，思路，優(yōu)化點(diǎn)都已經(jīng)列出來了，我們可以盡可能的細(xì)化，三個(gè)方向都可以細(xì)化，如此，所有的實(shí)現(xiàn)便一目了然，即使不看 Kafka 的實(shí)現(xiàn)，我們自己也可以想到一二點(diǎn)可以優(yōu)化的地方。

這就是思考方式。提出問題 > 列出問題點(diǎn) > 列出優(yōu)化方法 > 列出具體可切入的點(diǎn) > tradeoff和細(xì)化實(shí)現(xiàn)。

現(xiàn)在，你也可以嘗試自己想一想優(yōu)化的點(diǎn)和方法，不用盡善盡美，不用管好不好實(shí)現(xiàn)，想一點(diǎn)是一點(diǎn)。

“

不行啊，我很笨，也很懶，你還是直接和我說吧，我白嫖比較行。

”

順序?qū)?/span>

“

人家 Redis 是基于純內(nèi)存的系統(tǒng)，你 kafka 還要讀寫磁盤，能比？

”

為什么說寫磁盤慢？

我們不能只知道結(jié)論，而不知其所以然。要回答這個(gè)問題，就得回到在校時(shí)我們學(xué)的操作系統(tǒng)課程了。來，翻到講磁盤的章節(jié)，讓我們回顧一下磁盤的運(yùn)行原理。

“

鬼還留著哦，課程還沒上到一半書就沒了。要不是考試俺眼神好，估計(jì)現(xiàn)在還沒畢業(yè)。

”

看經(jīng)典大圖：

完成一次磁盤 IO，需要經(jīng)過尋道、旋轉(zhuǎn)和數(shù)據(jù)傳輸三個(gè)步驟。

影響磁盤 IO 性能的因素也就發(fā)生在上面三個(gè)步驟上，因此主要花費(fèi)的時(shí)間就是：

1. 尋道時(shí)間：Tseek 是指將讀寫磁頭移動(dòng)至正確的磁道上所需要的時(shí)間。尋道時(shí)間越短，I/O 操作越快，目前磁盤的平均尋道時(shí)間一般在 3-15ms。
2. 旋轉(zhuǎn)延遲：Trotation 是指盤片旋轉(zhuǎn)將請(qǐng)求數(shù)據(jù)所在的扇區(qū)移動(dòng)到讀寫磁盤下方所需要的時(shí)間。旋轉(zhuǎn)延遲取決于磁盤轉(zhuǎn)速，通常用磁盤旋轉(zhuǎn)一周所需時(shí)間的 1/2 表示。比如：7200rpm 的磁盤平均旋轉(zhuǎn)延遲大約為 60*1000/7200/2 = 4.17ms，而轉(zhuǎn)速為 15000rpm 的磁盤其平均旋轉(zhuǎn)延遲為 2ms。
3. 數(shù)據(jù)傳輸時(shí)間：Ttransfer 是指完成傳輸所請(qǐng)求的數(shù)據(jù)所需要的時(shí)間，它取決于數(shù)據(jù)傳輸率，其值等于數(shù)據(jù)大小除以數(shù)據(jù)傳輸率。目前 IDE/ATA 能達(dá)到 133MB/s，SATA II 可達(dá)到 300MB/s 的接口數(shù)據(jù)傳輸率，數(shù)據(jù)傳輸時(shí)間通常遠(yuǎn)小于前兩部分消耗時(shí)間。簡(jiǎn)單計(jì)算時(shí)可忽略。

因此，如果在寫磁盤的時(shí)候省去尋道、旋轉(zhuǎn)可以極大地提高磁盤讀寫的性能。

Kafka 采用順序?qū)?/code>文件的方式來提高磁盤寫入性能。順序?qū)?/code>文件，基本減少了磁盤尋道和旋轉(zhuǎn)的次數(shù)。磁頭再也不用在磁道上亂舞了，而是一路向前飛速前行。

Kafka 中每個(gè)分區(qū)是一個(gè)有序的，不可變的消息序列，新的消息不斷追加到 Partition 的末尾，在 Kafka 中 Partition 只是一個(gè)邏輯概念，Kafka 將 Partition 劃分為多個(gè) Segment，每個(gè) Segment 對(duì)應(yīng)一個(gè)物理文件，Kafka 對(duì) segment 文件追加寫，這就是順序?qū)懳募?/p>

“

為什么 Kafka 可以使用追加寫的方式呢？

”

這和 Kafka 的性質(zhì)有關(guān)，我們來看看 Kafka 和 Redis，說白了，Kafka 就是一個(gè)Queue，而 Redis 就是一個(gè)HashMap。Queue和Map的區(qū)別是什么？

Queue 是 FIFO 的，數(shù)據(jù)是有序的；HashMap數(shù)據(jù)是無序的，是隨機(jī)讀寫的。Kafka 的不可變性，有序性使得 Kafka 可以使用追加寫的方式寫文件。

其實(shí)很多符合以上特性的數(shù)據(jù)系統(tǒng)，都可以采用追加寫的方式來優(yōu)化磁盤性能。典型的有Redis的 AOF 文件，各種數(shù)據(jù)庫的WAL(Write ahead log)機(jī)制等等。

“

所以清楚明白自身業(yè)務(wù)的特點(diǎn)，就可以針對(duì)性地做出優(yōu)化。

”

零拷貝

“

哈哈，這個(gè)我面試被問到過。可惜答得一般般，唉。

”

什么是零拷貝？

我們從 Kafka 的場(chǎng)景來看，Kafka Consumer 消費(fèi)存儲(chǔ)在 Broker 磁盤的數(shù)據(jù)，從讀取 Broker 磁盤到網(wǎng)絡(luò)傳輸給 Consumer，期間涉及哪些系統(tǒng)交互。Kafka Consumer 從 Broker 消費(fèi)數(shù)據(jù)，Broker 讀取 Log，就使用了 sendfile。如果使用傳統(tǒng)的 IO 模型，偽代碼邏輯就如下所示：

readFile(buffer)
send(buffer)

如圖，如果采用傳統(tǒng)的 IO 流程，先讀取網(wǎng)絡(luò) IO，再寫入磁盤 IO，實(shí)際需要將數(shù)據(jù) Copy 四次。

1. 第一次：讀取磁盤文件到操作系統(tǒng)內(nèi)核緩沖區(qū)；
2. 第二次：將內(nèi)核緩沖區(qū)的數(shù)據(jù)，copy 到應(yīng)用程序的 buffer；
3. 第三步：將應(yīng)用程序 buffer 中的數(shù)據(jù)，copy 到 socket 網(wǎng)絡(luò)發(fā)送緩沖區(qū)；
4. 第四次：將 socket buffer 的數(shù)據(jù)，copy 到網(wǎng)卡，由網(wǎng)卡進(jìn)行網(wǎng)絡(luò)傳輸。

“

啊，操作系統(tǒng)這么傻嗎？copy 來 copy 去的。

”

并不是操作系統(tǒng)傻，操作系統(tǒng)的設(shè)計(jì)就是每個(gè)應(yīng)用程序都有自己的用戶內(nèi)存，用戶內(nèi)存和內(nèi)核內(nèi)存隔離，這是為了程序和系統(tǒng)安全考慮，否則的話每個(gè)應(yīng)用程序內(nèi)存滿天飛，隨意讀寫那還得了。

不過，還有零拷貝技術(shù)，英文——Zero-Copy。零拷貝就是盡量去減少上面數(shù)據(jù)的拷貝次數(shù)，從而減少拷貝的 CPU 開銷，減少用戶態(tài)內(nèi)核態(tài)的上下文切換次數(shù)，從而優(yōu)化數(shù)據(jù)傳輸?shù)男阅堋?/p>

常見的零拷貝思路主要有三種：

• 直接 I/O：數(shù)據(jù)直接跨過內(nèi)核，在用戶地址空間與 I/O 設(shè)備之間傳遞，內(nèi)核只是進(jìn)行必要的虛擬存儲(chǔ)配置等輔助工作；
• 避免內(nèi)核和用戶空間之間的數(shù)據(jù)拷貝：當(dāng)應(yīng)用程序不需要對(duì)數(shù)據(jù)進(jìn)行訪問時(shí)，則可以避免將數(shù)據(jù)從內(nèi)核空間拷貝到用戶空間；
• 寫時(shí)復(fù)制：數(shù)據(jù)不需要提前拷貝，而是當(dāng)需要修改的時(shí)候再進(jìn)行部分拷貝。

Kafka 使用到了 mmap 和 sendfile 的方式來實(shí)現(xiàn)零拷貝。分別對(duì)應(yīng) Java 的 MappedByteBuffer 和 FileChannel.transferTo。

使用 Java NIO 實(shí)現(xiàn)零拷貝，如下：

FileChannel.transferTo()

在此模型下，上下文切換的數(shù)量減少到一個(gè)。具體而言，transferTo()方法指示塊設(shè)備通過 DMA 引擎將數(shù)據(jù)讀取到讀取緩沖區(qū)中。然后，將該緩沖區(qū)復(fù)制到另一個(gè)內(nèi)核緩沖區(qū)以暫存到套接字。最后，套接字緩沖區(qū)通過 DMA 復(fù)制到 NIC 緩沖區(qū)。

我們將副本數(shù)從四減少到三，并且這些副本中只有一個(gè)涉及 CPU。我們還將上下文切換的數(shù)量從四個(gè)減少到了兩個(gè)。這是一個(gè)很大的改進(jìn)，但是還沒有查詢零副本。當(dāng)運(yùn)行 Linux 內(nèi)核 2.4 及更高版本以及支持收集操作的網(wǎng)絡(luò)接口卡時(shí)，后者可以作為進(jìn)一步的優(yōu)化來實(shí)現(xiàn)。如下所示。

根據(jù)前面的示例，調(diào)用transferTo()方法會(huì)使設(shè)備通過 DMA 引擎將數(shù)據(jù)讀取到內(nèi)核讀取緩沖區(qū)中。但是，使用gather操作時(shí)，讀取緩沖區(qū)和套接字緩沖區(qū)之間沒有復(fù)制。取而代之的是，給 NIC 一個(gè)指向讀取緩沖區(qū)的指針以及偏移量和長(zhǎng)度，該偏移量和長(zhǎng)度由 DMA 清除。CPU 絕對(duì)不參與復(fù)制緩沖區(qū)。

關(guān)于零拷貝詳情，可以詳讀這篇文章零拷貝 (Zero-copy) 淺析及其應(yīng)用。

PageCache

producer 生產(chǎn)消息到 Broker 時(shí)，Broker 會(huì)使用 pwrite() 系統(tǒng)調(diào)用【對(duì)應(yīng)到 Java NIO 的 FileChannel.write() API】按偏移量寫入數(shù)據(jù)，此時(shí)數(shù)據(jù)都會(huì)先寫入page cache。consumer 消費(fèi)消息時(shí)，Broker 使用 sendfile() 系統(tǒng)調(diào)用【對(duì)應(yīng) FileChannel.transferTo() API】，零拷貝地將數(shù)據(jù)從 page cache 傳輸?shù)?broker 的 Socket buffer，再通過網(wǎng)絡(luò)傳輸。

leader 與 follower 之間的同步，與上面 consumer 消費(fèi)數(shù)據(jù)的過程是同理的。

page cache中的數(shù)據(jù)會(huì)隨著內(nèi)核中 flusher 線程的調(diào)度以及對(duì) sync()/fsync() 的調(diào)用寫回到磁盤，就算進(jìn)程崩潰，也不用擔(dān)心數(shù)據(jù)丟失。另外，如果 consumer 要消費(fèi)的消息不在page cache里，才會(huì)去磁盤讀取，并且會(huì)順便預(yù)讀出一些相鄰的塊放入 page cache，以方便下一次讀取。

因此如果 Kafka producer 的生產(chǎn)速率與 consumer 的消費(fèi)速率相差不大，那么就能幾乎只靠對(duì) broker page cache 的讀寫完成整個(gè)生產(chǎn) - 消費(fèi)過程，磁盤訪問非常少。

網(wǎng)絡(luò)模型

“

網(wǎng)絡(luò)嘛，作為 Java 程序員，自然是 Netty

”

是的，Netty 是 JVM 領(lǐng)域一個(gè)優(yōu)秀的網(wǎng)絡(luò)框架，提供了高性能的網(wǎng)絡(luò)服務(wù)。大多數(shù) Java 程序員提到網(wǎng)絡(luò)框架，首先想到的就是 Netty。Dubbo、Avro-RPC 等等優(yōu)秀的框架都使用 Netty 作為底層的網(wǎng)絡(luò)通信框架。

Kafka 自己實(shí)現(xiàn)了網(wǎng)絡(luò)模型做 RPC。底層基于 Java NIO，采用和 Netty 一樣的 Reactor 線程模型。

Reacotr 模型主要分為三個(gè)角色

• Reactor：把 IO 事件分配給對(duì)應(yīng)的 handler 處理
• Acceptor：處理客戶端連接事件
• Handler：處理非阻塞的任務(wù)

在傳統(tǒng)阻塞 IO 模型中，每個(gè)連接都需要獨(dú)立線程處理，當(dāng)并發(fā)數(shù)大時(shí)，創(chuàng)建線程數(shù)多，占用資源；采用阻塞 IO 模型，連接建立后，若當(dāng)前線程沒有數(shù)據(jù)可讀，線程會(huì)阻塞在讀操作上，造成資源浪費(fèi)

針對(duì)傳統(tǒng)阻塞 IO 模型的兩個(gè)問題，Reactor 模型基于池化思想，避免為每個(gè)連接創(chuàng)建線程，連接完成后將業(yè)務(wù)處理交給線程池處理；基于 IO 復(fù)用模型，多個(gè)連接共用同一個(gè)阻塞對(duì)象，不用等待所有的連接。遍歷到有新數(shù)據(jù)可以處理時(shí)，操作系統(tǒng)會(huì)通知程序，線程跳出阻塞狀態(tài)，進(jìn)行業(yè)務(wù)邏輯處理

Kafka 即基于 Reactor 模型實(shí)現(xiàn)了多路復(fù)用和處理線程池。其設(shè)計(jì)如下：

其中包含了一個(gè)Acceptor線程，用于處理新的連接，Acceptor 有 N 個(gè) Processor 線程 select 和 read socket 請(qǐng)求，N 個(gè) Handler 線程處理請(qǐng)求并相應(yīng)，即處理業(yè)務(wù)邏輯。

I/O 多路復(fù)用可以通過把多個(gè) I/O 的阻塞復(fù)用到同一個(gè) select 的阻塞上，從而使得系統(tǒng)在單線程的情況下可以同時(shí)處理多個(gè)客戶端請(qǐng)求。它的最大優(yōu)勢(shì)是系統(tǒng)開銷小，并且不需要?jiǎng)?chuàng)建新的進(jìn)程或者線程，降低了系統(tǒng)的資源開銷。

總結(jié)： Kafka Broker 的 KafkaServer 設(shè)計(jì)是一個(gè)優(yōu)秀的網(wǎng)絡(luò)架構(gòu)，有想了解 Java 網(wǎng)絡(luò)編程，或需要使用到這方面技術(shù)的同學(xué)不妨去讀一讀源碼。

批量與壓縮

Kafka Producer 向 Broker 發(fā)送消息不是一條消息一條消息的發(fā)送。使用過 Kafka 的同學(xué)應(yīng)該知道，Producer 有兩個(gè)重要的參數(shù)：batch.size和linger.ms。這兩個(gè)參數(shù)就和 Producer 的批量發(fā)送有關(guān)。

Kafka Producer 的執(zhí)行流程如下圖所示：

發(fā)送消息依次經(jīng)過以下處理器：

• Serialize：鍵和值都根據(jù)傳遞的序列化器進(jìn)行序列化。優(yōu)秀的序列化方式可以提高網(wǎng)絡(luò)傳輸?shù)男省?/section>
• Partition：決定將消息寫入主題的哪個(gè)分區(qū)，默認(rèn)情況下遵循 murmur2 算法。自定義分區(qū)程序也可以傳遞給生產(chǎn)者，以控制應(yīng)將消息寫入哪個(gè)分區(qū)。
• Compress：默認(rèn)情況下，在 Kafka 生產(chǎn)者中不啟用壓縮.Compression 不僅可以更快地從生產(chǎn)者傳輸?shù)酱?，還可以在復(fù)制過程中進(jìn)行更快的傳輸。壓縮有助于提高吞吐量，降低延遲并提高磁盤利用率。
• Accumulate：Accumulate顧名思義，就是一個(gè)消息累計(jì)器。其內(nèi)部為每個(gè) Partition 維護(hù)一個(gè)Deque雙端隊(duì)列，隊(duì)列保存將要發(fā)送的批次數(shù)據(jù)，Accumulate將數(shù)據(jù)累計(jì)到一定數(shù)量，或者在一定過期時(shí)間內(nèi)，便將數(shù)據(jù)以批次的方式發(fā)送出去。記錄被累積在主題每個(gè)分區(qū)的緩沖區(qū)中。根據(jù)生產(chǎn)者批次大小屬性將記錄分組。主題中的每個(gè)分區(qū)都有一個(gè)單獨(dú)的累加器 / 緩沖區(qū)。
• Group Send：記錄累積器中分區(qū)的批次按將它們發(fā)送到的代理分組。批處理中的記錄基于 batch.size 和 linger.ms 屬性發(fā)送到代理。記錄由生產(chǎn)者根據(jù)兩個(gè)條件發(fā)送。當(dāng)達(dá)到定義的批次大小或達(dá)到定義的延遲時(shí)間時(shí)。

Kafka 支持多種壓縮算法：lz4、snappy、gzip。Kafka 2.1.0 正式支持 ZStandard —— ZStandard 是 Facebook 開源的壓縮算法，旨在提供超高的壓縮比 (compression ratio)，具體細(xì)節(jié)參見 zstd。

Producer、Broker 和 Consumer 使用相同的壓縮算法，在 producer 向 Broker 寫入數(shù)據(jù)，Consumer 向 Broker 讀取數(shù)據(jù)時(shí)甚至可以不用解壓縮，最終在 Consumer Poll 到消息時(shí)才解壓，這樣節(jié)省了大量的網(wǎng)絡(luò)和磁盤開銷。

分區(qū)并發(fā)

Kafka 的 Topic 可以分成多個(gè) Partition，每個(gè) Paritition 類似于一個(gè)隊(duì)列，保證數(shù)據(jù)有序。同一個(gè) Group 下的不同 Consumer 并發(fā)消費(fèi) Paritition，分區(qū)實(shí)際上是調(diào)優(yōu) Kafka 并行度的最小單元，因此，可以說，每增加一個(gè) Paritition 就增加了一個(gè)消費(fèi)并發(fā)。

Kafka 具有優(yōu)秀的分區(qū)分配算法——StickyAssignor，可以保證分區(qū)的分配盡量地均衡，且每一次重分配的結(jié)果盡量與上一次分配結(jié)果保持一致。這樣，整個(gè)集群的分區(qū)盡量地均衡，各個(gè) Broker 和 Consumer 的處理不至于出現(xiàn)太大的傾斜。

“

那是不是分區(qū)數(shù)越多越好呢？

”

當(dāng)然不是。

越多的分區(qū)需要打開更多的文件句柄

在 kafka 的 broker 中，每個(gè)分區(qū)都會(huì)對(duì)照著文件系統(tǒng)的一個(gè)目錄。在 kafka 的數(shù)據(jù)日志文件目錄中，每個(gè)日志數(shù)據(jù)段都會(huì)分配兩個(gè)文件，一個(gè)索引文件和一個(gè)數(shù)據(jù)文件。因此，隨著 partition 的增多，需要的文件句柄數(shù)急劇增加，必要時(shí)需要調(diào)整操作系統(tǒng)允許打開的文件句柄數(shù)。

客戶端 / 服務(wù)器端需要使用的內(nèi)存就越多

客戶端 producer 有個(gè)參數(shù) batch.size，默認(rèn)是 16KB。它會(huì)為每個(gè)分區(qū)緩存消息，一旦滿了就打包將消息批量發(fā)出?？瓷先ミ@是個(gè)能夠提升性能的設(shè)計(jì)。不過很顯然，因?yàn)檫@個(gè)參數(shù)是分區(qū)級(jí)別的，如果分區(qū)數(shù)越多，這部分緩存所需的內(nèi)存占用也會(huì)更多。

降低高可用性

分區(qū)越多，每個(gè) Broker 上分配的分區(qū)也就越多，當(dāng)一個(gè)發(fā)生 Broker 宕機(jī)，那么恢復(fù)時(shí)間將很長(zhǎng)。

文件結(jié)構(gòu)

Kafka 消息是以 Topic 為單位進(jìn)行歸類，各個(gè) Topic 之間是彼此獨(dú)立的，互不影響。每個(gè) Topic 又可以分為一個(gè)或多個(gè)分區(qū)。每個(gè)分區(qū)各自存在一個(gè)記錄消息數(shù)據(jù)的日志文件。

Kafka 每個(gè)分區(qū)日志在物理上實(shí)際按大小被分成多個(gè) Segment。

• segment file 組成：由 2 大部分組成，分別為 index file 和 data file，此 2 個(gè)文件一一對(duì)應(yīng)，成對(duì)出現(xiàn)，后綴”.index”和“.log”分別表示為 segment 索引文件、數(shù)據(jù)文件。
• segment 文件命名規(guī)則：partion 全局的第一個(gè) segment 從 0 開始，后續(xù)每個(gè) segment 文件名為上一個(gè) segment 文件最后一條消息的 offset 值。數(shù)值最大為 64 位 long 大小，19 位數(shù)字字符長(zhǎng)度，沒有數(shù)字用 0 填充。

index 采用稀疏索引，這樣每個(gè) index 文件大小有限，Kafka 采用mmap的方式，直接將 index 文件映射到內(nèi)存，這樣對(duì) index 的操作就不需要操作磁盤 IO。mmap的 Java 實(shí)現(xiàn)對(duì)應(yīng) MappedByteBuffer 。

“

mmap 是一種內(nèi)存映射文件的方法。即將一個(gè)文件或者其它對(duì)象映射到進(jìn)程的地址空間，實(shí)現(xiàn)文件磁盤地址和進(jìn)程虛擬地址空間中一段虛擬地址的一一對(duì)映關(guān)系。實(shí)現(xiàn)這樣的映射關(guān)系后，進(jìn)程就可以采用指針的方式讀寫操作這一段內(nèi)存，而系統(tǒng)會(huì)自動(dòng)回寫臟頁面到對(duì)應(yīng)的文件磁盤上，即完成了對(duì)文件的操作而不必再調(diào)用 read,write 等系統(tǒng)調(diào)用函數(shù)。相反，內(nèi)核空間對(duì)這段區(qū)域的修改也直接反映用戶空間，從而可以實(shí)現(xiàn)不同進(jìn)程間的文件共享。

”

Kafka 充分利用二分法來查找對(duì)應(yīng) offset 的消息位置：

1. 按照二分法找到小于 offset 的 segment 的.log 和.index
2. 用目標(biāo) offset 減去文件名中的 offset 得到消息在這個(gè) segment 中的偏移量。
3. 再次用二分法在 index 文件中找到對(duì)應(yīng)的索引。
4. 到 log 文件中，順序查找，直到找到 offset 對(duì)應(yīng)的消息。

總結(jié)

Kafka 是一個(gè)優(yōu)秀的開源項(xiàng)目。其在性能上面的優(yōu)化做的淋漓盡致，是很值得我們深入學(xué)習(xí)的一個(gè)項(xiàng)目。無論是思想還是實(shí)現(xiàn)，我們都應(yīng)該認(rèn)真的去看一看，想一想。

Kafka 性能優(yōu)化：

1. 零拷貝網(wǎng)絡(luò)和磁盤
2. 優(yōu)秀的網(wǎng)絡(luò)模型，基于 Java NIO
3. 高效的文件數(shù)據(jù)結(jié)構(gòu)設(shè)計(jì)
4. Parition 并行和可擴(kuò)展
5. 數(shù)據(jù)批量傳輸
6. 數(shù)據(jù)壓縮
7. 順序讀寫磁盤
8. 無鎖輕量級(jí) offset