Kafka是如何保證高性能和高吞吐量的？

• 2019 年 11 月 14 日
• 筆記

Kafka是一款分散式消息發布和訂閱系統，它的特點是高性能、高吞吐量。

最早設計的目的是作為LinkedIn的活動流和運營數據的處理管道。這些數據主要是用來對用戶做用戶畫像分析以及伺服器性能數據的一些監控。

所以kafka一開始設計的目標就是作為一個分散式、高吞吐量的消息系統，所以適合運用在大數據傳輸場景。

Kafka的應用場景

由於kafka具有更好的吞吐量、內置分區、冗餘及容錯性的優點(kafka每秒可以處理幾十萬消息)，讓kafka成為了一個很好的大規模消息處理應用的解決方案。所以在企業級應用長，主要會應用於如下幾個方面

• 行為跟蹤：kafka可以用於跟蹤用戶瀏覽頁面、搜索及其他行為。通過發布-訂閱模式實時記錄到對應的topic中，通過後端大數據平台接入處理分析，並做更進一步的實時處理和監控
• 日誌收集：日誌收集方面，有很多比較優秀的產品，比如Apache Flume，很多公司使用kafka代理日誌聚合。日誌聚合表示從伺服器上收集日誌文件，然後放到一個集中的平台（文件伺服器）進行處理。在實際應用開發中，我們應用程式的log都會輸出到本地的磁碟上，排查問題的話通過linux命令來搞定，如果應用程式組成了負載均衡集群，並且集群的機器有幾十台以上，那麼想通過日誌快速定位到問題，就是很麻煩的事情了。所以一般都會做一個日誌統一收集平台管理log日誌用來快速查詢重要應用的問題。所以很多公司的套路都是把應用日誌集中到kafka上，然後分別導入到es和hdfs上，用來做實時檢索分析和離線統計數據備份等。而另一方面，kafka本身又提供了很好的api來集成日誌並且做日誌收集。

Kafka的架構

一個典型的kafka集群包含若干Producer（可以是應用節點產生的消息，也可以是通過Flume收集日誌產生的事件），若干個Broker（kafka支援水平擴展）、若干個Consumer Group，以及一個zookeeper集群。kafka通過zookeeper管理集群配置及服務協同。Producer使用push模式將消息發布到broker，consumer通過監聽使用pull模式從broker訂閱並消費消息。

多個broker協同工作，producer和consumer部署在各個業務邏輯中。三者通過zookeeper管理協調請求和轉發。這樣就組成了一個高性能的分散式消息發布和訂閱系統。

圖上有一個細節是和其他mq中間件不同的點，producer 發送消息到broker的過程是push，而consumer從broker消費消息的過程是pull，主動去拉數據。而不是broker把數據主動發送給consumer。

名詞解釋

1）Broker Kafka集群包含一個或多個伺服器，這種伺服器被稱為broker。broker端不維護數據的消費狀態，提升了性能。直接使用磁碟進行存儲，線性讀寫，速度快：避免了數據在JVM記憶體和系統記憶體之間的複製，減少耗性能的創建對象和垃圾回收。 2）Producer 負責發布消息到Kafka broker 3）Consumer 消息消費者，向Kafka broker讀取消息的客戶端，consumer從broker拉取(pull)數據並進行處理。 4）Topic 每條發布到Kafka集群的消息都有一個類別，這個類別被稱為Topic。（物理上不同Topic的消息分開存儲，邏輯上一個Topic的消息雖然保存於一個或多個broker上但用戶只需指定消息的Topic即可生產或消費數據而不必關心數據存於何處） 5）Partition Parition是物理上的概念，每個Topic包含一個或多個Partition. 6）Consumer Group 每個Consumer屬於一個特定的Consumer Group（可為每個Consumer指定group name，若不指定group name則屬於默認的group） 7）Topic & Partition Topic在邏輯上可以被認為是一個queue，每條消費都必須指定它的Topic，可以簡單理解為必須指明把這條消息放進哪個queue里。為了使得Kafka的吞吐率可以線性提高，物理上把Topic分成一個或多個Partition，每個Partition在物理上對應一個文件夾，該文件夾下存儲這個Partition的所有消息和索引文件。若創建topic1和topic2兩個topic，且分別有13個和19個分區，則整個集群上會相應會生成共32個文件夾（本文所用集群共8個節點，此處topic1和topic2 replication-factor均為1）。

Java中使用kafka進行通訊

依賴

org.apache.kafka

kafka-clients

2.0.0

發送端程式碼

消費端程式碼

非同步發送

batch.size

生產者發送多個消息到broker上的同一個分區時，為了減少網路請求帶來的性能開銷，通過批量的方式來提交消息，可以通過這個參數來控制批量提交的位元組數大小，默認大小是16384byte,也就是16kb，意味著當一批消息大小達到指定的batch.size的時候會統一發送

linger.ms

Producer默認會把兩次發送時間間隔內收集到的所有Requests進行一次聚合然後再發送，以此提高吞吐量，而linger.ms就是為每次發送到broker的請求增加一些delay，以此來聚合更多的Message請求。這個有點想TCP裡面的Nagle演算法，在TCP協議的傳輸中，為了減少大量小數據包的發送，採用了Nagle演算法，也就是基於小包的等-停協議。 batch.size和linger.ms這兩個參數是kafka性能優化的關鍵參數，batch.size和linger.ms這兩者的作用是一樣的，如果兩個都配置了，那麼怎麼工作的呢？實際上，當二者都配置的時候，只要滿足其中一個要求，就會發送請求到broker上

一些基礎配置分析

group.id

consumer group是kafka提供的可擴展且具有容錯性的消費者機制。既然是一個組，那麼組內必然可以有多個消費者或消費者實例(consumer instance)，它們共享一個公共的ID，即group ID。組內的所有消費者協調在一起來消費訂閱主題(subscribed topics)的所有分區(partition)。當然，每個分區只能由同一個消費組內的一個consumer來消費.如下圖所示，分別有三個消費者，屬於兩個不同的group，那麼對於firstTopic這個topic來說，這兩個組的消費者都能同時消費這個topic中的消息，對於此時的架構來說，這個firstTopic就類似於ActiveMQ中的topic概念。如右圖所示，如果3個消費者都屬於同一個group，那麼此時firstTopic就是一個Queue的概念

enable.auto.commit

消費者消費消息以後自動提交，只有當消息提交以後，該消息才不會被再次接收到，還可以配合auto.commit.interval.ms控制自動提交的頻率。 當然，我們也可以通過consumer.commitSync()的方式實現手動提交

auto.offset.reset

這個參數是針對新的groupid中的消費者而言的，當有新groupid的消費者來消費指定的topic時，對於該參數的配置，會有不同的語義。 auto.offset.reset=latest情況下，新的消費者將會從其他消費者最後消費的offset處開始消費Topic下的消息。 auto.offset.reset= earliest情況下，新的消費者會從該topic最早的消息開始消費。 auto.offset.reset=none情況下，新的消費者加入以後，由於之前不存在offset，則會直接拋出異常。

max.poll.records

此設置限制每次調用poll返回的消息數，這樣可以更容易的預測每次poll間隔要處理的最大值。通過調整此值，可以減少poll間隔

原理分析

從前面的整個演示過程來看，只要不是超大規模的使用kafka，那麼基本上沒什麼大問題，否則，對於kafka本身的運維的挑戰會很大，同時，針對每一個參數的調優也顯得很重要。 據我了解，快手在使用kafka集群規模是挺大的，他們在19年的開發者大會上有提到

關於Topic和Partition

Topic

在kafka中，topic是一個存儲消息的邏輯概念，可以認為是一個消息集合。每條消息發送到kafka集群的消息都有一個類別。物理上來說，不同的topic的消息是分開存儲的， 每個topic可以有多個生產者向它發送消息，也可以有多個消費者去消費其中的消息。

Partition

每個topic可以劃分多個分區（每個Topic至少有一個分區），同一topic下的不同分區包含的消息是不同的。每個消息在被添加到分區時，都會被分配一個offset（稱之為偏移量），它是消息在此分區中的唯一編號，kafka通過offset保證消息在分區內的順序，offset的順序不跨分區，即kafka只保證在同一個分區內的消息是有序的。

下圖中，對於名字為test的topic，做了3個分區，分別是p0、p1、p2. 每一條消息發送到broker時，會根據partition的規則選擇存儲到哪一個partition。如果partition規則設置合理，那麼所有的消息會均勻的分布在不同的partition中，這樣就有點類似資料庫的分庫分表的概念，把數據做了分片處理。

Topic&Partition的存儲

Partition是以文件的形式存儲在文件系統中，比如創建一個名為firstTopic的topic，其中有3個partition，那麼在kafka的數據目錄（/tmp/kafka-log）中就有3個目錄，firstTopic-0~3， 命名規則是-

關於消息分發

kafka消息分發策略

消息是kafka中最基本的數據單元，在kafka中，一條消息由key、value兩部分構成，在發送一條消息時，我們可以指定這個key，那麼producer會根據key和partition機制來判斷當前這條消息應該發送並存儲到哪個partition中。我們可以根據需要進行擴展producer的partition機制。

自定義Partitioner

發送端程式碼添加自定義分區

消息默認的分發機制

默認情況下，kafka採用的是hash取模的分區演算法。如果Key為null，則會隨機分配一個分區。這個隨機是在這個參數」metadata.max.age.ms」的時間範圍內隨機選擇一個。對於這個時間段內，如果key為null，則只會發送到唯一的分區。這個值值哦默認情況下是10分鐘更新一次。

關於Metadata，這個之前沒講過，簡單理解就是Topic/Partition和broker的映射關係，每一個topic的每一個partition，需要知道對應的broker列表是什麼，leader是誰、follower是誰。這些資訊都是存儲在Metadata這個類裡面。

消費端如何消費指定的分區

通過下面的程式碼，就可以消費指定該topic下的0號分區。其他分區的數據就無法接收

消息的消費原理

在實際生產過程中，每個topic都會有多個partitions，多個partitions的好處在於，一方面能夠對broker上的數據進行分片有效減少了消息的容量從而提升io性能。另外一方面，為了提高消費端的消費能力，一般會通過多個consumer去消費同一個topic ，也就是消費端的負載均衡機制，也就是我們接下來要了解的，在多個partition以及多個consumer的情況下，消費者是如何消費消息的。 kafka存在consumer group的概念，也就是group.id一樣的consumer，這些consumer屬於一個consumer group，組內的所有消費者協調在一起來消費訂閱主題的所有分區。當然每一個分區只能由同一個消費組內的consumer來消費，那麼同一個consumergroup裡面的consumer是怎麼去分配該消費哪個分區里的數據的呢？如下圖所示，3個分區，3個消費者，那麼哪個消費者消分哪個分區？

對於上面這個圖來說，這3個消費者會分別消費test這個topic 的3個分區，也就是每個consumer消費一個partition。

• 演示1（3個partiton對應3個consumer） Ø 創建一個帶3個分區的topic Ø 啟動3個消費者消費同一個topic，並且這3個consumer屬於同一個組 Ø 啟動發送者進行消息發送

演示結果：consumer1會消費partition0分區、consumer2會消費partition1分區、consumer3會消費partition2分區 如果是2個consumer消費3個partition呢？會是怎麼樣的結果？

• 演示2（3個partiton對應2個consumer） Ø 基於上面演示的案例的topic不變 Ø 啟動2個消費這消費該topic Ø 啟動發送者進行消息發送 演示結果：consumer1會消費partition0/partition1分區、consumer2會消費partition2分區
• 演示3（3個partition對應4個或以上consumer） 演示結果：仍然只有3個consumer對應3個partition，其他的consumer無法消費消息 通過這個演示的過程，引出接下來需要了解的kafka的分區分配策略(Partition Assignment Strategy)

consumer和partition的數量建議

1. 如果consumer比partition多，是浪費，因為kafka的設計是在一個partition上是不允許並發的，所以consumer數不要大於partition數
2. 如果consumer比partition少，一個consumer會對應於多個partitions，這裡主要合理分配consumer數和partition數，否則會導致partition裡面的數據被取的不均勻。最好partiton數目是consumer數目的整數倍，所以partition數目很重要，比如取24，就很容易設定consumer數目
3. 如果consumer從多個partition讀到數據，不保證數據間的順序性，kafka只保證在一個partition上數據是有序的，但多個partition，根據你讀的順序會有不同
4. 增減consumer，broker，partition會導致rebalance，所以rebalance後consumer對應的partition會發生變化

什麼是分區分配策略

通過前面的案例演示，我們應該能猜到，同一個group中的消費者對於一個topic中的多個partition，存在一定的分區分配策略。 在kafka中，存在三種分區分配策略，一種是Range(默認)、 另一種是RoundRobin（輪詢）、StickyAssignor(粘性)。在消費端中的ConsumerConfig中，通過這個屬性來指定分區分配策略

RangeAssignor（範圍分區）

Range策略是對每個主題而言的，首先對同一個主題裡面的分區按照序號進行排序，並對消費者按照字母順序進行排序。

假設我們有10個分區，3個消費者，排完序的分區將會是0, 1, 2, 3, 4, 5, 6, 7, 8, 9；消費者執行緒排完序將會是C1-0, C2-0, C3-0。然後將partitions的個數除於消費者執行緒的總數來決定每個消費者執行緒消費幾個分區。如果除不盡，那麼前面幾個消費者執行緒將會多消費一個分區。在我們的例子裡面，我們有10個分區，3個消費者執行緒， 10 / 3 = 3，而且除不盡，那麼消費者執行緒 C1-0 將會多消費一個分區. 結果看起來是這樣的： C1-0 將消費 0, 1, 2, 3 分區 C2-0 將消費 4, 5, 6 分區 C3-0 將消費 7, 8, 9 分區

假如我們有11個分區，那麼最後分區分配的結果看起來是這樣的： C1-0 將消費 0, 1, 2, 3 分區 C2-0 將消費 4, 5, 6, 7 分區 C3-0 將消費 8, 9, 10 分區

假如我們有2個主題(T1和T2)，分別有10個分區，那麼最後分區分配的結果看起來是這樣的： C1-0 將消費 T1主題的 0, 1, 2, 3 分區以及 T2主題的 0, 1, 2, 3分區 C2-0 將消費 T1主題的 4, 5, 6 分區以及 T2主題的 4, 5, 6分區 C3-0 將消費 T1主題的 7, 8, 9 分區以及 T2主題的 7, 8, 9分區

可以看出，C1-0 消費者執行緒比其他消費者執行緒多消費了2個分區，這就是Range strategy的一個很明顯的弊端

RoundRobinAssignor（輪詢分區）

輪詢分區策略是把所有partition和所有consumer執行緒都列出來，然後按照hashcode進行排序。最後通過輪詢演算法分配partition給消費執行緒。如果所有consumer實例的訂閱是相同的，那麼partition會均勻分布。

在我們的例子裡面，假如按照 hashCode 排序完的topic-partitions組依次為T1-5, T1-3, T1-0, T1-8, T1-2, T1-1, T1-4, T1-7, T1-6, T1-9，我們的消費者執行緒排序為C1-0, C1-1, C2-0, C2-1，最後分區分配的結果為： C1-0 將消費 T1-5, T1-2, T1-6 分區； C1-1 將消費 T1-3, T1-1, T1-9 分區； C2-0 將消費 T1-0, T1-4 分區； C2-1 將消費 T1-8, T1-7 分區；

使用輪詢分區策略必須滿足兩個條件:

1. 每個主題的消費者實例具有相同數量的流
2. 每個消費者訂閱的主題必須是相同的

StrickyAssignor 分配策略

kafka在0.11.x版本支援了StrickyAssignor, 翻譯過來叫粘滯策略，它主要有兩個目的:

• 分區的分配儘可能的均勻
• 分區的分配儘可能和上次分配保持相同

當兩者發生衝突時， 第 一 個目標優先於第二個目標。鑒於這兩個目標， StickyAssignor分配策略的具體實現要比RangeAssignor和RoundRobinAssi gn or這兩種分配策略要複雜得多，假設我們有這樣一個場景:

假設消費組有3個消費者：C0,C1,C2，它們分別訂閱了4個Topic(t0,t1,t2,t3),並且每個主題有兩個分區(p0,p1),也就是說，整個消費組訂閱了8個分區：tOpO 、 tOpl 、 tlpO 、 tlpl 、 t2p0 、t2pl 、t3p0 、 t3pl 那麼最終的分配場景結果為 CO: tOpO、tlpl 、 t3p0 Cl: tOpl、t2p0 、 t3pl C2: tlpO、t2pl 這種分配方式有點類似於輪詢策略，但實際上並不是，因為假設這個時候，C1這個消費者掛了，就勢必會造成重新分區（reblance），如果是輪詢，那麼結果應該是 CO: tOpO、tlpO、t2p0、t3p0 C2: tOpl、tlpl、t2pl、t3pl 然後，strickyAssignor它是一種粘滯策略，所以它會滿足`分區的分配儘可能和上次分配保持相同，所以分配結果應該是 消費者CO: tOpO、tlpl 、 t3p0、t2p0 消費者C2: tlpO、t2pl、tOpl、t3pl 也就是說，C0和C2保留了上一次是的分配結果，並且把原來C1的分區分配給了C0和C2。這種策略的好處是使得分區發生變化時，由於分區的「粘性，減少了不必要的分區移動

誰來執行Rebalance以及管理consumer的group呢？

Kafka提供了一個角色：coordinator來執行對於consumer group的管理，Kafka提供了一個角色：coordinator來執行對於consumer group的管理，當consumer group的第一個consumer啟動的時候，它會去和kafka server確定誰是它們組的coordinator。之後該group內的所有成員都會和該coordinator進行協調通訊

如何確定coordinator

consumer group如何確定自己的coordinator是誰呢, 消費者向kafka集群中的任意一個broker發送一個GroupCoordinatorRequest請求，服務端會返回一個負載最小的broker節點的id，並將該broker設置為coordinator

JoinGroup的過程

在rebalance之前，需要保證coordinator是已經確定好了的，整個rebalance的過程分為兩個步驟，Join和Sync

join: 表示加入到consumer group中，在這一步中，所有的成員都會向coordinator發送joinGroup的請求。一旦所有成員都發送了joinGroup請求，那麼coordinator會選擇一個consumer擔任leader角色，並把組成員資訊和訂閱資訊發送消費者 leader選舉演算法比較簡單，如果消費組內沒有leader，那麼第一個加入消費組的消費者就是消費者leader，如果這個時候leader消費者退出了消費組，那麼重新選舉一個leader，這個選舉很隨意，類似於隨機演算法

protocol_metadata: 序列化後的消費者的訂閱資訊 leader_id：消費組中的消費者，coordinator會選擇一個座位leader，對應的就是member_id member_metadata 對應消費者的訂閱資訊 members：consumer group中全部的消費者的訂閱資訊 generation_id：年代資訊，類似於之前講解zookeeper的時候的epoch是一樣的，對於每一輪rebalance，generation_id都會遞增。主要用來保護consumer group。隔離無效的offset提交。也就是上一輪的consumer成員無法提交offset到新的consumer group中。

每個消費者都可以設置自己的分區分配策略，對於消費組而言，會從各個消費者上報過來的分區分配策略中選舉一個彼此都贊同的策略來實現整體的分區分配，這個"贊同"的規則是，消費組內的各個消費者會通過投票來決定

• 在joingroup階段，每個consumer都會把自己支援的分區分配策略發送到coordinator
• coordinator手機到所有消費者的分配策略，組成一個候選集
• 每個消費者需要從候選集里找出一個自己支援的策略，並且為這個策略投票
• 最終計算候選集中各個策略的選票數，票數最多的就是當前消費組的分配策略

Synchronizing Group State階段

完成分區分配之後，就進入了Synchronizing Group State階段，主要邏輯是向GroupCoordinator發送SyncGroupRequest請求，並且處理SyncGroupResponse響應，簡單來說，就是leader將消費者對應的partition分配方案同步給consumer group 中的所有consumer

每個消費者都會向coordinator發送syncgroup請求，不過只有leader節點會發送分配方案，其他消費者只是打打醬油而已。當leader把方案發給coordinator以後，coordinator會把結果設置到SyncGroupResponse中。這樣所有成員都知道自己應該消費哪個分區。

consumer group的分區分配方案是在客戶端執行的！Kafka將這個權利下放給客戶端主要是因為這樣做可以有更好的靈活性

總結

我們再來總結一下consumer group rebalance的過程 Ø 對於每個consumer group子集，都會在服務端對應一個GroupCoordinator進行管理，GroupCoordinator會在zookeeper上添加watcher，當消費者加入或者退出consumer group時，會修改zookeeper上保存的數據，從而觸發GroupCoordinator開始Rebalance操作 Ø 當消費者準備加入某個Consumer group或者GroupCoordinator發生故障轉移時，消費者並不知道GroupCoordinator的在網路中的位置，這個時候就需要確定GroupCoordinator，消費者會向集群中的任意一個Broker節點發送ConsumerMetadataRequest請求，收到請求的broker會返回一個response作為響應，其中包含管理當前ConsumerGroup的GroupCoordinator， Ø 消費者會根據broker的返回資訊，連接到groupCoordinator，並且發送HeartbeatRequest，發送心跳的目的是要要奧噶蘇GroupCoordinator這個消費者是正常在線的。當消費者在指定時間內沒有發送心跳請求，則GroupCoordinator會觸發Rebalance操作。

Ø 發起join group請求，兩種情況

• 如果GroupCoordinator返回的心跳包數據包含異常，說明GroupCoordinator因為前面說的幾種情況導致了Rebalance操作，那這個時候，consumer會發起join group請求
• 新加入到consumer group的consumer確定好了GroupCoordinator以後消費者會向GroupCoordinator發起join group請求，GroupCoordinator會收集全部消費者資訊之後，來確認可用的消費者，並從中選取一個消費者成為group_leader。並把相應的資訊（分區分配策略、leader_id、…）封裝成response返回給所有消費者，但是只有group leader會收到當前consumer group中的所有消費者資訊。當消費者確定自己是group leader以後，會根據消費者的資訊以及選定分區分配策略進行分區分配
• 接著進入Synchronizing Group State階段，每個消費者會發送SyncGroupRequest請求到GroupCoordinator，但是只有Group Leader的請求會存在分區分配結果，GroupCoordinator會根據Group Leader的分區分配結果形成SyncGroupResponse返回給所有的Consumer。
• consumer根據分配結果，執行相應的操作

到這裡為止，我們已經知道了消息的發送分區策略，以及消費者的分區消費策略和rebalance。對於應用層面來說，還有一個最重要的東西沒有講解，就是offset，他類似一個游標，表示當前消費的消息的位置。

如何保存消費端的消費位置

什麼是offset

前面在講解partition的時候，提到過offset， 每個topic可以劃分多個分區（每個Topic至少有一個分區），同一topic下的不同分區包含的消息是不同的。每個消息在被添加到分區時，都會被分配一個offset（稱之為偏移量），它是消息在此分區中的唯一編號，kafka通過offset保證消息在分區內的順序，offset的順序不跨分區，即kafka只保證在同一個分區內的消息是有序的；對於應用層的消費來說，每次消費一個消息並且提交以後，會保存當前消費到的最近的一個offset。那麼offset保存在哪裡？

offset在哪裡維護？

在kafka中，提供了一個consumer_offsets_* 的一個topic，把offset資訊寫入到這個topic中。 consumer_offsets——按保存了每個consumer group某一時刻提交的offset資訊。 __consumer_offsets 默認有50個分區。 根據前面我們演示的案例，我們設置了一個KafkaConsumerDemo的groupid。首先我們需要找到這個consumer_group保存在哪個分區中 properties.put(ConsumerConfig.GROUP_ID_CONFIG,"KafkaConsumerDemo"); 計算公式: Math.abs(「groupid」.hashCode())%groupMetadataTopicPartitionCount ; 由於默認情況下groupMetadataTopicPartitionCount有50個分區，計算得到的結果為:35, 意味著當前的consumer_group的位移資訊保存在__consumer_offsets的第35個分區 執行如下命令，可以查看當前consumer_goup中的offset位移提交的資訊

從輸出結果中，我們就可以看到test這個topic的offset的位移日誌

分區的副本機制

我們已經知道Kafka的每個topic都可以分為多個Partition，並且多個partition會均勻分布在集群的各個節點下。雖然這種方式能夠有效的對數據進行分片，但是對於每個partition來說，都是單點的，當其中一個partition不可用的時候，那麼這部分消息就沒辦法消費。所以kafka為了提高partition的可靠性而提供了副本的概念（Replica）,通過副本機制來實現冗餘備份。

每個分區可以有多個副本，並且在副本集合中會存在一個leader的副本，所有的讀寫請求都是由leader副本來進行處理。剩餘的其他副本都做為follower副本，follower副本會從leader副本同步消息日誌。 這個有點類似zookeeper中leader和follower的概念，但是具體的時間方式還是有比較大的差異。所以我們可以認為，副本集會存在一主多從的關係。

一般情況下，同一個分區的多個副本會被均勻分配到集群中的不同broker上，當leader副本所在的broker出現故障後，可以重新選舉新的leader副本繼續對外提供服務。通過這樣的副本機制來提高kafka集群的可用性。

創建一個帶副本機制的topic

通過下面的命令去創建帶2個副本的topic

然後我們可以在/tmp/kafka-log路徑下看到對應topic的副本資訊了。我們通過一個圖形的方式來表達。 針對secondTopic這個topic的3個分區對應的3個副本

如何知道那個各個分區中對應的leader是誰呢？

在zookeeper伺服器上，通過如下命令去獲取對應分區的資訊, 比如下面這個是獲取secondTopic第1個 分區的狀態資訊。

{"controller_epoch":12,"leader":0,"version":1,"leader_epoch":0,"isr":[0,1]} 或通過這個命令

leader表示當前分區的leader是那個broker-id。下圖中。綠色線條的表示該分區中的leader節點。其他節點就為follower

需要注意的是，kafka集群中的一個broker中最多只能有一個副本，leader副本所在的broker節點的分區叫leader節點，follower副本所在的broker節點的分區叫follower節點

副本的leader選舉

Kafka提供了數據複製演算法保證，如果leader副本所在的broker節點宕機或者出現故障，或者分區的leader節點發生故障，這個時候怎麼處理呢？ 那麼，kafka必須要保證從follower副本中選擇一個新的leader副本。那麼kafka是如何實現選舉的呢？ 要了解leader選舉，我們需要了解幾個概念 Kafka分區下有可能有很多個副本(replica)用於實現冗餘，從而進一步實現高可用。副本根據角色的不同可分為3類：

• leader副本：響應clients端讀寫請求的副本
• follower副本：被動地備份leader副本中的數據，不能響應clients端讀寫請求。
• ISR副本：包含了leader副本和所有與leader副本保持同步的follower副本——如何判定是否與leader同步後面會提到每個Kafka副本對象都有兩個重要的屬性：LEO和HW。注意是所有的副本，而不只是leader副本。
• LEO：即日誌末端位移(log end offset)，記錄了該副本底層日誌(log)中下一條消息的位移值。注意是下一條消息！也就是說，如果LEO=10，那麼表示該副本保存了10條消息，位移值範圍是[0, 9]。另外，leader LEO和follower LEO的更新是有區別的。我們後面會詳細說
• HW：即上面提到的水位值。對於同一個副本對象而言，其HW值不會大於LEO值。小於等於HW值的所有消息都被認為是「已備份」的（replicated）。同理，leader副本和follower副本的HW更新是有區別的 從生產者發出的 一 條消息首先會被寫入分區的leader 副本，不過還需要等待ISR集合中的所有follower副本都同步完之後才能被認為已經提交，之後才會更新分區的HW, 進而消費者可以消費到這條消息。

副本協同機制

剛剛提到了，消息的讀寫操作都只會由leader節點來接收和處理。follower副本只負責同步數據以及當leader副本所在的broker掛了以後，會從follower副本中選取新的leader。

寫請求首先由Leader副本處理，之後follower副本會從leader上拉取寫入的消息，這個過程會有一定的延遲，導致follower副本中保存的消息略少於leader副本，但是只要沒有超出閾值都可以容忍。但是如果一個follower副本出現異常，比如宕機、網路斷開等原因長時間沒有同步到消息，那這個時候，leader就會把它踢出去。kafka通過ISR集合來維護一個分區副本資訊

一個新leader被選舉並被接受客戶端的消息成功寫入。Kafka確保從同步副本列表中選舉一個副本為leader；leader負責維護和跟蹤ISR(in-Sync replicas ， 副本同步隊列)中所有follower滯後的狀態。當producer發送一條消息到broker後，leader寫入消息並複製到所有follower。消息提交之後才被成功複製到所有的同步副本。

ISR

ISR表示目前「可用且消息量與leader相差不多的副本集合，這是整個副本集合的一個子集」。怎麼去理解可用和相差不多這兩個詞呢？具體來說，ISR集合中的副本必須滿足兩個條件：

1. 副本所在節點必須維持著與zookeeper的連接
2. 副本最後一條消息的offset與leader副本的最後一條消息的offset之間的差值不能超過指定的閾值 (replica.lag.time.max.ms) replica.lag.time.max.ms：如果該follower在此時間間隔內一直沒有追上過leader的所有消息，則該follower就會被剔除isr列表
3. ISR數據保存在Zookeeper的 /brokers/topics/ /partitions//state 節點中

follower副本把leader副本LEO之前的日誌全部同步完成時，則認為follower副本已經追趕上了leader副本，這個時候會更新這個副本的lastCaughtUpTimeMs標識，kafk副本管理器會啟動一個副本過期檢查的定時任務，這個任務會定期檢查當前時間與副本的lastCaughtUpTimeMs的差值是否大於參數replica.lag.time.max.ms 的值，如果大於，則會把這個副本踢出ISR集合

如何處理所有的Replica不工作的情況

在ISR中至少有一個follower時，Kafka可以確保已經commit的數據不丟失，但如果某個Partition的所有Replica都宕機了，就無法保證數據不丟失了

1. 等待ISR中的任一個Replica「活」過來，並且選它作為Leader
2. 選擇第一個「活」過來的Replica（不一定是ISR中的）作為Leader

這就需要在可用性和一致性當中作出一個簡單的折衷。 如果一定要等待ISR中的Replica「活」過來，那不可用的時間就可能會相對較長。而且如果ISR中的所有Replica都無法「活」過來了，或者數據都丟失了，這個Partition將永遠不可用。 選擇第一個「活」過來的Replica作為Leader，而這個Replica不是ISR中的Replica，那即使它並不保證已經包含了所有已commit的消息，它也會成為Leader而作為consumer的數據源（所有讀寫都由Leader完成）。在我們課堂講的版本中，使用的是第一種策略。

副本數據同步原理

了解了副本的協同過程以後，還有一個最重要的機制，就是數據的同步過程。它需要解決

1. 怎麼傳播消息
2. 在向消息發送端返回ack之前需要保證多少個Replica已經接收到這個消息

下圖中，深紅色部分表示test_replica分區的leader副本，另外兩個節點上淺色部分表示follower副本

Producer在發布消息到某個Partition時，

• 先通過ZooKeeper找到該Partition的Leader get /brokers/topics/ /partitions/2/state ，然後無論該Topic的Replication Factor為多少（也即該Partition有多少個Replica），Producer只將該消息發送到該Partition的Leader。
• Leader會將該消息寫入其本地Log。每個Follower都從Leader pull數據。這種方式上，Follower存儲的數據順序與Leader保持一致。
• Follower在收到該消息並寫入其Log後，向Leader發送ACK。
• 一旦Leader收到了ISR中的所有Replica的ACK，該消息就被認為已經commit了，Leader將增加HW(HighWatermark)並且向Producer發送ACK。

LEO：即日誌末端位移(log end offset)，記錄了該副本底層日誌(log)中下一條消息的位移值。注意是下一條消息！也就是說，如果LEO=10，那麼表示該副本保存了10條消息，位移值範圍是[0, 9]。另外，leader LEO和follower LEO的更新是有區別的。

HW：即上面提到的水位值（Hight Water）。對於同一個副本對象而言，其HW值不會大於LEO值。小於等於HW值的所有消息都被認為是「已備份」的（replicated）。同理，leader副本和follower副本的HW更新是有區別的

通過下面這幅圖來表達LEO、HW的含義，隨著follower副本不斷和leader副本進行數據同步，follower副本的LEO會主鍵後移並且追趕到leader副本，這個追趕上的判斷標準是當前副本的LEO是否大於或者等於leader副本的HW，這個追趕上也會使得被踢出的follower副本重新加入到ISR集合中。 另外， 假如說下圖中的最右側的follower副本被踢出ISR集合，也會導致這個分區的HW發生變化，變成了3

數據丟失的問題

表達的含義是，至少需要多少個副本同步才能表示消息是提交的， 所以，當 min.insync.replicas=1的時候，一旦消息被寫入leader端log即被認為是「已提交」，而延遲一輪FETCH RPC更新HW值的設計使得follower HW值是非同步延遲更新的，倘若在這個過程中leader發生變更，那麼成為新leader的follower的HW值就有可能是過期的，使得clients端認為是成功提交的消息被刪除。

acks配置表示producer發送消息到broker上以後的確認值。有三個可選項 0：表示producer不需要等待broker的消息確認。這個選項時延最小但同時風險最大（因為當server宕機時，數據將會丟失）。 1：表示producer只需要獲得kafka集群中的leader節點確認即可，這個選擇時延較小同時確保了leader節點確認接收成功。 all(-1)：需要ISR中所有的Replica給予接收確認，速度最慢，安全性最高，但是由於ISR可能會縮小到僅包含一個Replica，所以設置參數為all並不能一定避免數據丟失，

數據丟失的解決方案

在kafka0.11.0.0版本之後，引入了一個leader epoch來解決這個問題，所謂的leader epoch實際上是一對值（epoch，offset），epoch代表leader的版本號，從0開始遞增，當leader發生過變更，epoch就+1，而offset則是對應這個epoch版本的leader寫入第一條消息的offset，比如 (0,0), (1,50) ,表示第一個leader從offset=0開始寫消息，一共寫了50條。第二個leader版本號是1，從offset=50開始寫，這個資訊會持久化在對應的分區的本地磁碟上，文件名是 /tmp/kafka-log/topic/leader-epoch-checkpoint 。 leader broker中會保存這樣一個快取，並且定期寫入到checkpoint文件中 當leader寫log時它會嘗試更新整個快取: 如果這個leader首次寫消息，則會在快取中增加一個條目；否則就不做更新。而每次副本重新成為leader時會查詢這部分快取，獲取出對應leader版本的offset

我們基於同樣的情況來分析，follower宕機並且恢復之後，有兩種情況，如果這個時候leader副本沒有掛，也就是意味著沒有發生leader選舉，那麼follower恢復之後並不會去截斷自己的日誌，而是先發送一個OffsetsForLeaderEpochRequest請求給到leader副本，leader副本收到請求之後返回當前的LEO。 如果follower副本的leaderEpoch和leader副本的epoch相同， leader的leo只可能大於或者等於follower副本的leo值，所以這個時候不會發生截斷 如果follower副本和leader副本的epoch值不同，那麼leader副本會查找follower副本傳過來的epoch+1在本地文件中存儲的StartOffset返回給follower副本，也就是新leader副本的LEO。這樣也避免了數據丟失的問題 如果leader副本宕機了重新選舉新的leader，那麼原本的follower副本就會變成leader，意味著epoch從0變成1，使得原本follower副本中LEO的值的到了保留。

Leader副本的選舉過程

1. KafkaController會監聽ZooKeeper的/brokers/ids節點路徑，一旦發現有broker掛了，執行下面的邏輯。這裡暫時先不考慮KafkaController所在broker掛了的情況，KafkaController掛了，各個broker會重新leader選舉出新的KafkaController
2. leader副本在該broker上的分區就要重新進行leader選舉，目前的選舉策略是 a) 優先從isr列表中選出第一個作為leader副本，這個叫優先副本，理想情況下有限副本就是該分區的leader副本 b) 如果isr列表為空，則查看該topic的unclean.leader.election.enable配置。 unclean.leader.election.enable：為true則代表允許選用非isr列表的副本作為leader，那麼此時就意味著數據可能丟失，為false的話，則表示不允許，直接拋出NoReplicaOnlineException異常，造成leader副本選舉失敗。 c) 如果上述配置為true，則從其他副本中選出一個作為leader副本，並且isr列表只包含該leader副本。一旦選舉成功，則將選舉後的leader和isr和其他副本資訊寫入到該分區的對應的zk路徑上。

消息的存儲

消息發送端發送消息到broker上以後，消息是如何持久化的呢？那麼接下來去分析下消息的存儲首先我們需要了解的是，kafka是使用日誌文件的方式來保存生產者和發送者的消息，每條消息都有一個offset值來表示它在分區中的偏移量。Kafka中存儲的一般都是海量的消息數據，為了避免日誌文件過大，Log並不是直接對應在一個磁碟上的日誌文件，而是對應磁碟上的一個目錄，這個目錄的命名規則是_

消息的文件存儲機制

一個topic的多個partition在物理磁碟上的保存路徑，路徑保存在 /tmp/kafka-logs/topic_partition，包含日誌文件、索引文件和時間索引文件

kafka是通過分段的方式將Log分為多個LogSegment，LogSegment是一個邏輯上的概念，一個LogSegment對應磁碟上的一個日誌文件和一個索引文件，其中日誌文件是用來記錄消息的。索引文件是用來保存消息的索引。那麼這個LogSegment是什麼呢？

LogSegment

假設kafka以partition為最小存儲單位，那麼我們可以想像當kafka producer不斷發送消息，必然會引起partition文件的無線擴張，這樣對於消息文件的維護以及被消費的消息的清理帶來非常大的挑戰，所以kafka 以segment為單位又把partition進行細分。每個partition相當於一個巨型文件被平均分配到多個大小相等的segment數據文件中（每個segment文件中的消息不一定相等），這種特性方便已經被消費的消息的清理，提高磁碟的利用率。

• log.segment.bytes=107370 (設置分段大小),默認是1gb，我們把這個值調小以後，可以看到日誌分段的效果
• 抽取其中3個分段來進行分析 image.png segment file由2大部分組成，分別為index file和data file，此2個文件一一對應，成對出現，後綴".index"和「.log」分別表示為segment索引文件、數據文件. segment文件命名規則：partion全局的第一個segment從0開始，後續每個segment文件名為上一個segment文件最後一條消息的offset值進行遞增。數值最大為64位long大小，20位數字字元長度，沒有數字用0填充

查看segment文件命名規則

通過下面這條命令可以看到kafka消息日誌的內容

假如第一個log文件的最後一個offset為:5376,所以下一個segment的文件命名為: 00000000000000005376.log。對應的index為00000000000000005376.index

segment中index和log的對應關係

從所有分段中，找一個分段進行分析 為了提高查找消息的性能，為每一個日誌文件添加2個索引索引文件：OffsetIndex 和 TimeIndex，分別對應.index以及.timeindex, TimeIndex索引文件格式：它是映射時間戳和相對offset 查看索引內容：

如圖所示，index中存儲了索引以及物理偏移量。log存儲了消息的內容。索引文件的元數據執行對應數據文件中message的物理偏移地址。舉個簡單的案例來說，以[4053,80899]為例，在log文件中，對應的是第4053條記錄，物理偏移量（position）為80899. position是ByteBuffer的指針位置

在partition中如何通過offset查找message

查找的演算法是

1. 根據offset的值，查找segment段中的index索引文件。由於索引文件命名是以上一個文件的最後一個offset進行命名的，所以，使用二分查找演算法能夠根據offset快速定位到指定的索引文件。
2. 找到索引文件後，根據offset進行定位，找到索引文件中的符合範圍的索引。（kafka採用稀疏索引的方式來提高查找性能）
3. 得到position以後，再到對應的log文件中，從position出開始查找offset對應的消息，將每條消息的offset與目標offset進行比較，直到找到消息

比如說，我們要查找offset=2490這條消息，那麼先找到00000000000000000000.index, 然後找到[2487,49111]這個索引，再到log文件中，根據49111這個position開始查找，比較每條消息的offset是否大於等於2490。最後查找到對應的消息以後返回

Log文件的消息內容分析

前面我們通過kafka提供的命令，可以查看二進位的日誌文件資訊，一條消息，會包含很多的欄位。

offset和position這兩個前面已經講過了、 createTime表示創建時間、keysize和valuesize表示key和value的大小、 compresscodec表示壓縮編碼、payload:表示消息的具體內容

日誌的清除策略以及壓縮策略

日誌清除策略

前面提到過，日誌的分段存儲，一方面能夠減少單個文件內容的大小，另一方面，方便kafka進行日誌清理。日誌的清理策略有兩個:

1. 根據消息的保留時間，當消息在kafka中保存的時間超過了指定的時間，就會觸發清理過程
2. 根據topic存儲的數據大小，當topic所佔的日誌文件大小大於一定的閥值，則可以開始刪除最舊的消息。kafka會啟動一個後台執行緒，定期檢查是否存在可以刪除的消息

通過log.retention.bytes和log.retention.hours這兩個參數來設置，當其中任意一個達到要求，都會執行刪除。 默認的保留時間是：7天

日誌壓縮策略

Kafka還提供了「日誌壓縮（Log Compaction）」功能，通過這個功能可以有效的減少日誌文件的大小，緩解磁碟緊張的情況，在很多實際場景中，消息的key和value的值之間的對應關係是不斷變化的，就像資料庫中的數據會不斷被修改一樣，消費者只關心key對應的最新的value。因此，我們可以開啟kafka的日誌壓縮功能，服務端會在後台啟動啟動Cleaner執行緒池，定期將相同的key進行合併，只保留最新的value值。日誌的壓縮原理是

磁碟存儲的性能問題

磁碟存儲的性能優化

我們現在大部分企業仍然用的是機械結構的磁碟，如果把消息以隨機的方式寫入到磁碟，那麼磁碟首先要做的就是定址，也就是定位到數據所在的物理地址，在磁碟上就要找到對應的柱面、磁頭以及對應的扇區；這個過程相對記憶體來說會消耗大量時間，為了規避隨機讀寫帶來的時間消耗，kafka採用順序寫的方式存儲數據。即使是這樣，但是頻繁的I/O操作仍然會造成磁碟的性能瓶頸

零拷貝

消息從發送到落地保存，broker維護的消息日誌本身就是文件目錄，每個文件都是二進位保存，生產者和消費者使用相同的格式來處理。在消費者獲取消息時，伺服器先從硬碟讀取數據到記憶體，然後把記憶體中的數據原封不動的通過socket發送給消費者。雖然這個操作描述起來很簡單，但實際上經歷了很多步驟。

作業系統將數據從磁碟讀入到內核空間的頁快取： ▪ 應用程式將數據從內核空間讀入到用戶空間快取中 ▪ 應用程式將數據寫回到內核空間到socket快取中 ▪ 作業系統將數據從socket緩衝區複製到網卡緩衝區，以便將數據經網路發出

通過「零拷貝」技術，可以去掉這些沒必要的數據複製操作，同時也會減少上下文切換次數。現代的unix作業系統提供一個優化的程式碼路徑，用於將數據從頁快取傳輸到socket；在Linux中，是通過sendfile系統調用來完成的。Java提供了訪問這個系統調用的方法：FileChannel.transferTo API 使用sendfile，只需要一次拷貝就行，允許作業系統將數據直接從頁快取發送到網路上。所以在這個優化的路徑中，只有最後一步將數據拷貝到網卡快取中是需要的

頁快取

頁快取是作業系統實現的一種主要的磁碟快取，但凡設計到快取的，基本都是為了提升i/o性能，所以頁快取是用來減少磁碟I/O操作的。 磁碟高速快取有兩個重要因素： 第一，訪問磁碟的速度要遠低於訪問記憶體的速度，若從處理器L1和L2高速快取訪問則速度更快。 第二，數據一旦被訪問，就很有可能短時間內再次訪問。正是由於基於訪問記憶體比磁碟快的多，所以磁碟的記憶體快取將給系統存儲性能帶來質的飛越。

當 一 個進程準備讀取磁碟上的文件內容時， 作業系統會先查看待讀取的數據所在的頁(page)是否在頁快取(pagecache)中，如果存在（命中）則直接返回數據， 從而避免了對物理磁碟的I/0操作；如果沒有命中， 則作業系統會向磁碟發起讀取請求並將讀取的數據頁存入頁快取， 之後再將數據返回給進程。 同樣，如果 一 個進程需要將數據寫入磁碟， 那麼作業系統也會檢測數據對應的頁是否在頁快取中，如果不存在， 則會先在頁快取中添加相應的頁， 最後將數據寫入對應的頁。被修改過後的頁也就變成了臟頁， 作業系統會在合適的時間把臟頁中的數據寫入磁碟， 以保持數據的 一 致性 Kafka中大量使用了頁快取， 這是Kafka實現高吞吐的重要因素之 一 。雖然消息都是先被寫入頁快取，然後由作業系統負責具體的刷盤任務的， 但在Kafka中同樣提供了同步刷盤及間斷性強制刷盤(fsync),可以通過 log.flush.interval.messages 和 log.flush.interval.ms 參數來控制。 同步刷盤能夠保證消息的可靠性，避免因為宕機導致頁快取數據還未完成同步時造成的數據丟失。但是實際使用上，我們沒必要去考慮這樣的因素以及這種問題帶來的損失，消息可靠性可以由多副本來解決，同步刷盤會帶來性能的影響。刷盤的操作由作業系統去完成即可

Kafka消息的可靠性

沒有一個中間件能夠做到百分之百的完全可靠，可靠性更多的還是基於幾個9的衡量指標，比如4個9、5個9. 軟體系統的可靠性只能夠無限去接近100%，但不可能達到100%。所以kafka如何是實現最大可能的可靠性呢？

• 分區副本， 你可以創建更多的分區來提升可靠性，但是分區數過多也會帶來性能上的開銷，一般來說，3個副本就能滿足對大部分場景的可靠性要求
• acks，生產者發送消息的可靠性，也就是我要保證我這個消息一定是到了broker並且完成了多副本的持久化，但這種要求也同樣會帶來性能上的開銷。它有幾個可選項： 1 ，生產者把消息發送到leader副本，leader副本在成功寫入到本地日誌之後就告訴生產者消息提交成功，但是如果isr集合中的follower副本還沒來得及同步leader副本的消息，leader掛了，就會造成消息丟失。 -1 ，消息不僅僅寫入到leader副本，並且被ISR集合中所有副本同步完成之後才告訴生產者已經提交成功，這個時候即使leader副本掛了也不會造成數據丟失。 0：表示producer不需要等待broker的消息確認。這個選項時延最小但同時風險最大（因為當server宕機時，數據將會丟失）。
• 保障消息到了broker之後，消費者也需要有一定的保證，因為消費者也可能出現某些問題導致消息沒有消費到。
• enable.auto.commit默認為true，也就是自動提交offset，自動提交是批量執行的，有一個時間窗口，這種方式會帶來重複提交或者消息丟失的問題，所以對於高可靠性要求的程式，要使用手動提交。對於高可靠要求的應用來說，寧願重複消費也不應該因為消費異常而導致消息丟失

來源：簡書 著作權歸作者所有。商業轉載請聯繫作者獲得授權，非商業轉載請註明出處。