前言

Kafka最初由Linkedin公司開發(fā)，是一個(gè)分布式、支持分區(qū)的（partition）、多副本的（replica），基于zookeeper協(xié)調(diào)的分布式消息系統(tǒng)，它的最大的特性就是可以實(shí)時(shí)的處理大量數(shù)據(jù)以滿足各種需求場(chǎng)景：比如基于hadoop的批處理系統(tǒng)、低延遲的實(shí)時(shí)系統(tǒng)、storm/Spark流式處理引擎，web/nginx日志、訪問日志，消息服務(wù)等等，用scala語言編寫，Linkedin于2010年貢獻(xiàn)給了Apache基金會(huì)并成為頂級(jí)開源項(xiàng)目。

1.簡介

1.1? 特性

????1. 高吞吐量、低延遲：kafka每秒可以處理幾十萬條消息，它的延遲最低只有幾毫秒。

????2. 可擴(kuò)展性：kafka集群支持熱擴(kuò)展。

????3. 持久性、可靠性：消息被持久化到本地磁盤，并且支持?jǐn)?shù)據(jù)備份防止數(shù)據(jù)丟失。

????4. 容錯(cuò)性：允許集群中節(jié)點(diǎn)失敗（若副本數(shù)量為n,則允許n-1個(gè)節(jié)點(diǎn)失?。?/p>

????5. 高并發(fā)：支持?jǐn)?shù)千個(gè)客戶端同時(shí)讀寫

? ? 6.?不可靠性：為了高性能，降低了部分可靠性，消息存在丟失和重復(fù)的情況。

1.2?使用場(chǎng)景

基于Kafka的特性，一般應(yīng)用在日志收集，消息系統(tǒng)，流式處理等對(duì)吞吐量要求較高，但對(duì)可靠性要求較低的場(chǎng)景下。

與Kafka相比，RabbitMQ/RocketMQ更側(cè)重于消息的可靠性，一般用于金融或電商訂單業(yè)務(wù)。

2.架構(gòu)圖

3.消息存儲(chǔ)

3.1?消息格式

每個(gè)partition分區(qū)在broker上保存為一個(gè)文件目錄，命名為<topic_name>_<partition_id>。

每個(gè)partition目錄下包含多個(gè)相同大小的segment文件，并以文件內(nèi)首個(gè)消息的offset命名，擴(kuò)展名為.log。

segment文件內(nèi)消息存儲(chǔ)格式為：<offset> <message_size> <message>，每個(gè)partition分區(qū)的offset都是獨(dú)立并遞增的。

每個(gè)segment文件維護(hù)一個(gè)索引，擴(kuò)展名.index，支持針對(duì)offset的二分查找。

3.2?消息刪除

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

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

基于大小，默認(rèn)配置是 1073741824。

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

4.生產(chǎn)者設(shè)計(jì)

4.1?producer寫

寫入時(shí)需要指定topic，key和partition可選，如果partition沒有指定，則根據(jù)key做hash取模得到partition，如果key也沒有設(shè)置，則用輪詢。

每個(gè)topic-partition有一個(gè)發(fā)送隊(duì)列，業(yè)務(wù)將消息寫入隊(duì)列，后臺(tái)線程根據(jù)batch_size和linger.ms最大等待時(shí)間執(zhí)行批量發(fā)送，這樣會(huì)造成消息的延遲，但是卻減少網(wǎng)絡(luò)IO，提高了吞吐量。

4.2 集群寫

producer從broker集群中獲取當(dāng)前分區(qū)對(duì)應(yīng)的leader，并將消息發(fā)送給leader。

leader負(fù)責(zé)將消息寫入log，并等待其他副本更新。

其他從leader同步消息，并寫入log，返回給leader?ack。

leader收到所有副本返回的ack，判定消息寫入成功，返回給producer成功。

為提高吞吐量，默認(rèn)配置為當(dāng)leader寫入成功，就返回成功，此時(shí)如果leader服務(wù)掛掉，會(huì)造成數(shù)據(jù)丟失。

5.消費(fèi)者設(shè)計(jì)

5.1 消費(fèi)流程

由于partition分區(qū)只對(duì)應(yīng)一個(gè)consumer，所以推薦consumer的個(gè)數(shù)和分區(qū)個(gè)數(shù)一樣，這樣能形成點(diǎn)對(duì)點(diǎn)，處理效率是最高的，如果consumer個(gè)數(shù)小于分區(qū)個(gè)數(shù)，則每個(gè)consumer會(huì)被分配多個(gè)partition，如果大于分區(qū)個(gè)數(shù)，則超出的consumer節(jié)點(diǎn)不會(huì)分配到partition，消費(fèi)不到數(shù)據(jù)。

consumer從broker集群獲取當(dāng)前分區(qū)的leader，并從leader批量pull消息，并提交消息的offset，為提高吞吐量，默認(rèn)為5秒自動(dòng)提交一次offset，但這樣會(huì)造成消息的丟失和重復(fù)。

5.2?已消費(fèi)offset

Kafka集群在每個(gè)partition上為每個(gè)消費(fèi)組維護(hù)一個(gè)已消費(fèi)offset，每次consumer消費(fèi)完成并提交后，集群都會(huì)更新這個(gè)offset。

在歷史版本中，這個(gè)offset信息是維護(hù)在zookeeper中，新版本維護(hù)在 __consumer_offsets 這個(gè) Topic 中。

5.3 rebalanced機(jī)制

Rebalance 本質(zhì)上是一種協(xié)議，規(guī)定了一個(gè) Consumer Group 下的所有 consumer 如何達(dá)成一致，來分配訂閱 Topic 的每個(gè)分區(qū)。

Rebalance 的觸發(fā)條件主要有2個(gè)：

????1.組成員個(gè)數(shù)發(fā)生變化，增加組員或者減少組員。

????2.訂閱的partition分區(qū)數(shù)發(fā)生變化。

5.4?心跳監(jiān)測(cè)

集群通過消費(fèi)端的兩個(gè)線程來監(jiān)測(cè)狀態(tài)，一個(gè)是心跳線程，一個(gè)是用戶poll線程。

心跳線程根據(jù)heartbeat.interval.ms參數(shù)(默認(rèn)3s)，定時(shí)向集群發(fā)送心跳包，心跳線程用于快速監(jiān)測(cè)消費(fèi)端的故障，盡早rebalance。

用戶poll線程從集群循環(huán)拉取消息，如果兩次poll的時(shí)間間隔超過了max.poll.interval.ms(默認(rèn)300s)，則認(rèn)定消費(fèi)端故障，執(zhí)行rebalance。

Kafka 0.10版本之前心跳包是放在poll線程去發(fā)的，這樣導(dǎo)致為了滿足業(yè)務(wù)處理時(shí)間，heartbeat.interval.ms時(shí)間要設(shè)置的很大才行，如果消費(fèi)端出了故障，心跳監(jiān)測(cè)不能馬上檢查到。

6.?高可用設(shè)計(jì)

Kafka在0.8以前的版本中，并不提供High Availablity機(jī)制，一旦一個(gè)或多個(gè)Broker宕機(jī)，則宕機(jī)期間其上所有Partition都無法繼續(xù)提供服務(wù)。

6.1?副本機(jī)制

同一個(gè)Partition會(huì)有多個(gè)Replication，并選出一個(gè)Leader，Producer和Consumer只與這個(gè)Leader交互，其它Replica作為Follower從Leader中復(fù)制數(shù)據(jù)。

6.2 zookeeper管理

引入zookeeper來管理broker的動(dòng)態(tài)加入和離開，實(shí)現(xiàn)故障發(fā)現(xiàn)和leader選舉。

zookeeper同時(shí)也會(huì)管理consumer的動(dòng)態(tài)加入與離開，Producer不需要管理，隨便一臺(tái)計(jì)算機(jī)都可以作為Producer向Kakfa Broker發(fā)消息。

7.?高吞吐設(shè)計(jì)

Kafka基于頁緩存計(jì)算+磁盤順序?qū)?，?shí)現(xiàn)了寫入數(shù)據(jù)的超高性能。

基于零拷貝技術(shù)，提高了讀取數(shù)據(jù)的性能。

7.1? 頁緩存技術(shù)

文件讀寫并不是直接訪問磁盤，而是利用到了操作系統(tǒng)的page-cache(頁緩存)，所以寫磁盤文件其實(shí)就是在寫內(nèi)存。

7.2 磁盤順序?qū)?/h2>

普通的機(jī)械磁盤隨機(jī)寫的性能極差，也就是隨便找到文件的某個(gè)位置來寫數(shù)據(jù)。

如果是追加文件末尾按照順序來寫數(shù)據(jù)的話，和寫內(nèi)存的性能是差不多的。

7.3?零拷貝技術(shù)

正常的數(shù)據(jù)發(fā)送流程：將數(shù)據(jù)從page-cache拷貝到應(yīng)用程序的進(jìn)程緩存中，然后調(diào)用write方法，將數(shù)據(jù)再拷貝到內(nèi)核socket發(fā)送緩沖區(qū)中，再經(jīng)過網(wǎng)卡發(fā)送出去。

零拷貝發(fā)送流程是：僅僅拷貝socket的描述符，然后數(shù)據(jù)就直接從page-cache中發(fā)送到網(wǎng)卡，節(jié)省了兩次數(shù)據(jù)的拷貝。

零拷貝的好處有：

? ? 1.避免操作系統(tǒng)內(nèi)核緩沖區(qū)之間進(jìn)行數(shù)據(jù)拷貝操作。

? ? 2.?避免操作系統(tǒng)內(nèi)核和用戶應(yīng)用程序地址空間這兩者之間進(jìn)行數(shù)據(jù)拷貝操作。

? ? 3.?減少內(nèi)核和用戶進(jìn)程的上下文切換。

? ? 4.?數(shù)據(jù)傳輸盡量讓 DMA 來做，解放了cpu。

8.?不可靠特性

Kafka是為了高吞吐量設(shè)計(jì)的，在滿足性能的前提下，不可避免的會(huì)帶來一些不可靠問題。

8.1?消息丟失

生產(chǎn)者丟失

生產(chǎn)者采用定時(shí)批量發(fā)送數(shù)據(jù)，如果期間生產(chǎn)者進(jìn)程掛掉，消息來不及發(fā)送出去，則消息丟失，解決辦法是減少消息發(fā)送的最大等待時(shí)間，比如可以配置為5ms，從而減少消息丟失的數(shù)量和幾率。

集群丟失

Kafka默認(rèn)是同步寫入，只要leader寫入成功就返回成功，此時(shí)如果leader掛掉，其他副本還沒來得及同步消息，則消息丟失，解決辦法是配置為等待所有副本寫入成功后，才返回成功，此時(shí)會(huì)降低寫入的性能，影響吞吐量。

消費(fèi)者丟失

消費(fèi)者設(shè)置為自動(dòng)提交時(shí)，如果消息被提交后，還沒來得及處理，進(jìn)程掛掉，此時(shí)消息丟失，解決辦法是改為手動(dòng)提交，犧牲性能。

8.2 重復(fù)消費(fèi)

重復(fù)消費(fèi)問題無法完全避免，如果業(yè)務(wù)系統(tǒng)不能容忍消息重復(fù)，需要自己實(shí)現(xiàn)冪等性。

生產(chǎn)者重復(fù)

生產(chǎn)者發(fā)送完消息，因?yàn)榫W(wǎng)絡(luò)問題沒有收到response，此時(shí)會(huì)重發(fā)消息，造成消息重復(fù)。

消費(fèi)者重復(fù)

消費(fèi)者設(shè)置為自動(dòng)提交時(shí)，如果業(yè)務(wù)層消息處理時(shí)間太久，超過了max.poll.interval.ms（默認(rèn)300s），則判定消費(fèi)端故障產(chǎn)生rebalance，再次poll時(shí)仍獲取到之前的消息，導(dǎo)致重復(fù)。解決辦法是減少max.poll.records(poll的消息個(gè)數(shù))，盡量保證消息處理的夠快。

在自動(dòng)提交模式下，只要集群產(chǎn)生rebalance，已處理過但來不及提交的消息都會(huì)被再消費(fèi)一次，導(dǎo)致重復(fù)。

8.3 同分區(qū)消息亂序

生產(chǎn)者發(fā)送消息時(shí)，如果前一個(gè)消息未響應(yīng)，可以繼續(xù)發(fā)送消息，如果前一個(gè)消息最終超時(shí)導(dǎo)致重發(fā)，則會(huì)出現(xiàn)消息亂序。

配置max.in.flight.requests.per.connection：限制客戶端在單個(gè)連接上能夠發(fā)送的未響應(yīng)請(qǐng)求的個(gè)數(shù)。設(shè)置此值是1表示kafka broker在響應(yīng)請(qǐng)求之前client不能再向同一個(gè)broker發(fā)送請(qǐng)求，但吞吐量會(huì)下降