架構(gòu)層面

利用partiton并行處理

kafka每個Topic都包含一個或多個Partition，不同Partition可位于不同節(jié)點。同時Partition在物理上對應(yīng)一個本地文件夾，每個Partition包含一個或多個Segment，每個Segment包含一個數(shù)據(jù)文件和一個與之對應(yīng)的索引文件。在邏輯上，可以把一個Partition當(dāng)作一個非常長的數(shù)組，可通過這個“數(shù)組”的索引（offset）去訪問其數(shù)據(jù)。

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

多Consumer消費同一個Topic時，同一條消息只會被同一Consumer Group內(nèi)的一個Consumer所消費。而數(shù)據(jù)并非按消息為單位分配，而是以Partition為單位分配，也即同一個Partition的數(shù)據(jù)只會被一個Consumer所消費（在不考慮Rebalance的前提下）。

如果Consumer的個數(shù)多于Partition的個數(shù)，那么會有部分Consumer無法消費該Topic的任何數(shù)據(jù)，也即當(dāng)Consumer個數(shù)超過Partition后，增加Consumer并不能增加并行度。

ISR-可用性和一致性的動態(tài)平衡

Kafka的數(shù)據(jù)復(fù)制是以Partition為單位的。而多個備份間的數(shù)據(jù)復(fù)制，通過Follower向Leader拉取數(shù)據(jù)完成。從一這點來講，Kafka的數(shù)據(jù)復(fù)制方案接近于Master-Slave方案。不同的是，Kafka既不是完全的同步復(fù)制，也不是完全的異步復(fù)制，而是基于ISR的動態(tài)復(fù)制方案。

ISR，也即In-sync Replica。每個Partition的Leader都會維護(hù)這樣一個列表，該列表中，包含了所有與之同步的Replica（包含Leader自己）。每次數(shù)據(jù)寫入時，只有ISR中的所有Replica都復(fù)制完，Leader才會將其置為Commit，它才能被Consumer所消費。

這種方案，與同步復(fù)制非常接近。但不同的是，這個ISR是由Leader動態(tài)維護(hù)的。如果Follower不能緊“跟上”Leader，它將被Leader從ISR中移除，待它又重新“跟上”Leader后，會被Leader再次加加ISR中。每次改變ISR后，Leader都會將最新的ISR持久化到Zookeeper中。

使用ISR方案的原因

由于Leader可移除不能及時與之同步的Follower，故與同步復(fù)制相比可避免最慢的Follower拖慢整體速度，也即ISR提高了系統(tǒng)可用性。

ISR中的所有Follower都包含了所有Commit過的消息，而只有Commit過的消息才會被Consumer消費，故從Consumer的角度而言，ISR中的所有Replica都始終處于同步狀態(tài)，從而與異步復(fù)制方案相比提高了數(shù)據(jù)一致性。

ISR可動態(tài)調(diào)整，極限情況下，可以只包含Leader，極大提高了可容忍的宕機(jī)的Follower的數(shù)量。與Majority Quorum方案相比，容忍相同個數(shù)的節(jié)點失敗，所要求的總節(jié)點數(shù)少了近一半。

ISR相關(guān)配置說明

Broker的min.insync.replicas參數(shù)指定了Broker所要求的ISR最小長度，默認(rèn)值為1。也即極限情況下ISR可以只包含Leader。但此時如果Leader宕機(jī)，則該Partition不可用，可用性得不到保證。

只有被ISR中所有Replica同步的消息才被Commit，但Producer發(fā)布數(shù)據(jù)時，Leader并不需要ISR中的所有Replica同步該數(shù)據(jù)才確認(rèn)收到數(shù)據(jù)。Producer可以通過acks參數(shù)指定最少需要多少個Replica確認(rèn)收到該消息才視為該消息發(fā)送成功。acks的默認(rèn)值是1，即Leader收到該消息后立即告訴Producer收到該消息，此時如果在ISR中的消息復(fù)制完該消息前Leader宕機(jī)，那該條消息會丟失。而如果將該值設(shè)置為0，則Producer發(fā)送完數(shù)據(jù)后，立即認(rèn)為該數(shù)據(jù)發(fā)送成功，不作任何等待，而實際上該數(shù)據(jù)可能發(fā)送失敗，并且Producer的Retry機(jī)制將不生效。更推薦的做法是，將acks設(shè)置為all或者-1，此時只有ISR中的所有Replica都收到該數(shù)據(jù)（也即該消息被Commit），Leader才會告訴Producer該消息發(fā)送成功，從而保證不會有未知的數(shù)據(jù)丟失。

具體實現(xiàn)層面

高效實用磁盤

1順序?qū)懘疟P

將寫磁盤的過程變?yōu)轫樞驅(qū)?，可極大提高對磁盤的利用率。

Kafka的整個設(shè)計中，Partition相當(dāng)于一個非常長的數(shù)組，而Broker接收到的所有消息順序?qū)懭脒@個大數(shù)組中。同時Consumer通過Offset順序消費這些數(shù)據(jù)，并且不刪除已經(jīng)消費的數(shù)據(jù)，從而避免了隨機(jī)寫磁盤的過程。

由于磁盤有限，不可能保存所有數(shù)據(jù)，實際上作為消息系統(tǒng)Kafka也沒必要保存所有數(shù)據(jù)，需要刪除舊的數(shù)據(jù)。而這個刪除過程，并非通過使用“讀-寫”模式去修改文件，而是將Partition分為多個Segment，每個Segment對應(yīng)一個物理文件，通過刪除整個文件的方式去刪除Partition內(nèi)的數(shù)據(jù)。這種方式清除舊數(shù)據(jù)的方式，也避免了對文件的隨機(jī)寫操作。

2.充分利用PageCache

使用Page Cache的好處如下：

1.I/O Scheduler會將連續(xù)的小塊寫組裝成大塊的物理寫從而提高性能；

2.I/O Scheduler會嘗試將一些寫操作重新按順序排好，從而減少磁盤頭的移動時間；

3.充分利用所有空閑內(nèi)存（非JVM內(nèi)存）。如果使用應(yīng)用層Cache（即JVM堆內(nèi)存），會增加GC負(fù)擔(dān)；

4.讀操作可直接在Page Cache內(nèi)進(jìn)行。如果消費和生產(chǎn)速度相當(dāng)，甚至不需要通過物理磁盤（直接通過Page Cache）交換數(shù)據(jù)；

5.如果進(jìn)程重啟，JVM內(nèi)的Cache會失效，但Page Cache仍然可用。

Broker收到數(shù)據(jù)后，寫磁盤時只是將數(shù)據(jù)寫入Page Cache，并不保證數(shù)據(jù)一定完全寫入磁盤。從這一點看，可能會造成機(jī)器宕機(jī)時，Page Cache內(nèi)的數(shù)據(jù)未寫入磁盤從而造成數(shù)據(jù)丟失。但是這種丟失只發(fā)生在機(jī)器斷電等造成操作系統(tǒng)不工作的場景，而這種場景完全可以由Kafka層面的Replication機(jī)制去解決。如果為了保證這種情況下數(shù)據(jù)不丟失而強(qiáng)制將Page Cache中的數(shù)據(jù)Flush到磁盤，反而會降低性能。也正因如此，Kafka雖然提供了flush.messages和flush.ms兩個參數(shù)將Page Cache中的數(shù)據(jù)強(qiáng)制Flush到磁盤，但是Kafka并不建議使用。

如果數(shù)據(jù)消費速度與生產(chǎn)速度相當(dāng)，甚至不需要通過物理磁盤交換數(shù)據(jù)，而是直接通過Page Cache交換數(shù)據(jù)。同時，F(xiàn)ollower從Leader Fetch數(shù)據(jù)時，也可通過Page Cache完成。

3.支持多Disk Drive

Broker的log.dirs配置項，允許配置多個文件夾。如果機(jī)器上有多個Disk Drive，可將不同的Disk掛載到不同的目錄，然后將這些目錄都配置到log.dirs里。Kafka會盡可能將不同的Partition分配到不同的目錄，也即不同的Disk上，從而充分利用了多Disk的優(yōu)勢。

零拷貝

Kafka中存在大量的網(wǎng)絡(luò)數(shù)據(jù)持久化到磁盤（Producer到Broker）和磁盤文件通過網(wǎng)絡(luò)發(fā)送（Broker到Consumer）的過程。這一過程的性能直接影響Kafka的整體吞吐量。

傳統(tǒng)模式下這一過程實際上發(fā)生了四次數(shù)據(jù)拷貝。首先通過系統(tǒng)調(diào)用將文件數(shù)據(jù)讀入到內(nèi)核態(tài)Buffer（DMA拷貝），然后應(yīng)用程序?qū)?nèi)存態(tài)Buffer數(shù)據(jù)讀入到用戶態(tài)Buffer（CPU拷貝），接著用戶程序通過Socket發(fā)送數(shù)據(jù)時將用戶態(tài)Buffer數(shù)據(jù)拷貝到內(nèi)核態(tài)Buffer（CPU拷貝），最后通過DMA拷貝將數(shù)據(jù)拷貝到NIC Buffer。同時，還伴隨著四次上下文切換。

Linux 2.4+內(nèi)核通過sendfile系統(tǒng)調(diào)用，提供了零拷貝。數(shù)據(jù)通過DMA拷貝到內(nèi)核態(tài)Buffer后，直接通過DMA拷貝到NIC Buffer，無需CPU拷貝。除了減少數(shù)據(jù)拷貝外，因為整個讀文件-網(wǎng)絡(luò)發(fā)送由一個sendfile調(diào)用完成，整個過程只有兩次上下文切換，因此大大提高了性能。

注： transferTo和transferFrom并不保證一定能使用零拷貝。實際上是否能使用零拷貝與操作系統(tǒng)相關(guān)，如果操作系統(tǒng)提供sendfile這樣的零拷貝系統(tǒng)調(diào)用，則這兩個方法會通過這樣的系統(tǒng)調(diào)用充分利用零拷貝的優(yōu)勢，否則并不能通過這兩個方法本身實現(xiàn)零拷貝。

減少網(wǎng)絡(luò)開銷

1.批處理

批處理是一種常用的用于提高I/O性能的方式。對Kafka而言，批處理既減少了網(wǎng)絡(luò)傳輸?shù)腛verhead，又提高了寫磁盤的效率。

Kafka 0.8.2開始支持新的Producer API，將同步Producer和異步Producer結(jié)合,send方法并非立即將消息發(fā)送出去，而是通過batch.size和linger.ms控制實際發(fā)送頻率，從而實現(xiàn)批量發(fā)送。

由于每次網(wǎng)絡(luò)傳輸，除了傳輸消息本身以外，還要傳輸非常多的網(wǎng)絡(luò)協(xié)議本身的一些內(nèi)容（稱為Overhead），所以將多條消息合并到一起傳輸，可有效減少網(wǎng)絡(luò)傳輸?shù)腛verhead，進(jìn)而提高了傳輸效率。

雖然Broker持續(xù)從網(wǎng)絡(luò)接收數(shù)據(jù)，但是寫磁盤并非每秒都在發(fā)生，而是間隔一段時間寫一次磁盤，并且每次寫磁盤的數(shù)據(jù)量都非常大（最高達(dá)到718MB/S）。

2.數(shù)據(jù)壓縮，降低網(wǎng)絡(luò)負(fù)載

Kafka從0.7開始，即支持將數(shù)據(jù)壓縮后再傳輸給Broker。除了可以將每條消息單獨壓縮然后傳輸外，Kafka還支持在批量發(fā)送時，將整個Batch的消息一起壓縮后傳輸。數(shù)據(jù)壓縮的一個基本原理是，重復(fù)數(shù)據(jù)越多壓縮效果越好。因此將整個Batch的數(shù)據(jù)一起壓縮能更大幅度減小數(shù)據(jù)量，從而更大程度提高網(wǎng)絡(luò)傳輸效率。

Broker接收消息后，并不直接解壓縮，而是直接將消息以壓縮后的形式持久化到磁盤。Consumer Fetch到數(shù)據(jù)后再解壓縮。因此Kafka的壓縮不僅減少了Producer到Broker的網(wǎng)絡(luò)傳輸負(fù)載，同時也降低了Broker磁盤操作的負(fù)載，也降低了Consumer與Broker間的網(wǎng)絡(luò)傳輸量，從而極大得提高了傳輸效率，提高了吞吐量。

3.高效的序列化方式

Kafka消息的Key和Payload（或者說Value）的類型可自定義，只需同時提供相應(yīng)的序列化器和反序列化器即可。因此用戶可以通過使用快速且緊湊的序列化-反序列化方式（如Avro，Protocal Buffer）來減少實際網(wǎng)絡(luò)傳輸和磁盤存儲的數(shù)據(jù)規(guī)模，從而提高吞吐率。這里要注意，如果使用的序列化方法太慢，即使壓縮比非常高，最終的效率也不一定高。