[[270715]]

Kafka 中的術(shù)語

broker：中間的kafka cluster，存儲(chǔ)消息，是由多個(gè)server組成的集群。

topic：kafka給消息提供的分類方式。broker用來存儲(chǔ)不同topic的消息數(shù)據(jù)。

producer：往broker中某個(gè)topic里面生產(chǎn)數(shù)據(jù)。

consumer：從broker中某個(gè)topic獲取數(shù)據(jù)。

Kafka 中的術(shù)語設(shè)計(jì)：

1、Broker

中間的kafka cluster，存儲(chǔ)消息，是由多個(gè)server組成的集群。

2、topic與消息

kafka將所有消息組織成多個(gè)topic的形式存儲(chǔ)，而每個(gè)topic又可以拆分成多個(gè)partition，每個(gè)partition又由一個(gè)一個(gè)消息組成。每個(gè)消息都被標(biāo)識(shí)了一個(gè)遞增序列號(hào)代表其進(jìn)來的先后順序，并按順序存儲(chǔ)在partition中。

這樣，消息就以一個(gè)個(gè)id的方式，組織起來。

producer選擇一個(gè)topic，生產(chǎn)消息，消息會(huì)通過分配策略append到某個(gè)partition末尾。

consumer選擇一個(gè)topic，通過id指定從哪個(gè)位置開始消費(fèi)消息。消費(fèi)完成之后保留id，下次可以從這個(gè)位置開始繼續(xù)消費(fèi)，也可以從其他任意位置開始消費(fèi)。

上面的id在kafka中稱為offset，這種組織和處理策略提供了如下好處：

消費(fèi)者可以根據(jù)需求，靈活指定offset消費(fèi)。

保證了消息不變性，為并發(fā)消費(fèi)提供了線程安全的保證。每個(gè)consumer都保留自己的offset，互相之間不干擾，不存在線程安全問題。

消息訪問的并行高效性。每個(gè)topic中的消息被組織成多個(gè)partition，partition均勻分配到集群server中。生產(chǎn)、消費(fèi)消息的時(shí)候，會(huì)被路由到指定partition，減少競(jìng)爭(zhēng)，增加了程序的并行能力。

增加消息系統(tǒng)的可伸縮性。每個(gè)topic中保留的消息可能非常龐大，通過partition將消息切分成多個(gè)子消息，并通過負(fù)責(zé)均衡策略將partition分配到不同server。這樣當(dāng)機(jī)器負(fù)載滿的時(shí)候，通過擴(kuò)容可以將消息重新均勻分配。

保證消息可靠性。消息消費(fèi)完成之后不會(huì)刪除，可以通過重置offset重新消費(fèi)，保證了消息不會(huì)丟失。

靈活的持久化策略。可以通過指定時(shí)間段(如最近一天)來保存消息，節(jié)省broker存儲(chǔ)空間。

備份高可用性。消息以partition為單位分配到多個(gè)server，并以partition為單位進(jìn)行備份。備份策略為：1個(gè)leader和N個(gè)followers，leader接受讀寫請(qǐng)求，followers被動(dòng)復(fù)制leader。leader和followers會(huì)在集群中打散，保證partition高可用。

3、Partitions

每個(gè)Topics劃分為一個(gè)或者多個(gè)Partition,并且Partition中的每條消息都被標(biāo)記了一個(gè)sequential id ,也就是offset,并且存儲(chǔ)的數(shù)據(jù)是可配置存儲(chǔ)時(shí)間的

4、producer

producer生產(chǎn)消息需要如下參數(shù)：

topic：往哪個(gè)topic生產(chǎn)消息。

partition：往哪個(gè)partition生產(chǎn)消息。

key：根據(jù)該key將消息分區(qū)到不同partition。

message：消息。

5、consumer

傳統(tǒng)消息系統(tǒng)有兩種模式：

隊(duì)列

發(fā)布訂閱

kafka通過consumer group將兩種模式統(tǒng)一處理：每個(gè)consumer將自己標(biāo)記consumer group名稱，之后系統(tǒng)會(huì)將consumer group按名稱分組，將消息復(fù)制并分發(fā)給所有分組，每個(gè)分組只有一個(gè)consumer能消費(fèi)這條消息。如下圖：

于是推理出兩個(gè)極端情況：

• 當(dāng)所有consumer的consumer group相同時(shí)，系統(tǒng)變成隊(duì)列模式
• 當(dāng)每個(gè)consumer的consumer group都不相同時(shí)，系統(tǒng)變成發(fā)布訂閱

注意：

1、Consumer Groups 提供了topics和partitions的隔離， 如上圖Consumer Group A中的consumer-C2掛掉，consumer-C1會(huì)接收P1,P2，即一個(gè)consumer Group中有其他consumer掛掉后能夠重新平衡。如下圖：

2、多consumer并發(fā)消費(fèi)消息時(shí)，容易導(dǎo)致消息亂序，通過限制消費(fèi)者為同步，可以保證消息有序，但是這大大降低了程序的并發(fā)性。

kafka通過partition的概念，保證了partition內(nèi)消息有序性，緩解了上面的問題。partition內(nèi)消息會(huì)復(fù)制分發(fā)給所有分組，每個(gè)分組只有一個(gè)consumer能消費(fèi)這條消息。這個(gè)語義保證了某個(gè)分組消費(fèi)某個(gè)分區(qū)的消息，是同步而非并發(fā)的。如果一個(gè)topic只有一個(gè)partition，那么這個(gè)topic并發(fā)消費(fèi)有序，否則只是單個(gè)partition有序。

一般消息系統(tǒng)，consumer存在兩種消費(fèi)模型：

push：優(yōu)勢(shì)在于消息實(shí)時(shí)性高。劣勢(shì)在于沒有考慮consumer消費(fèi)能力和飽和情況，容易導(dǎo)致producer壓垮consumer。

pull：優(yōu)勢(shì)在可以控制消費(fèi)速度和消費(fèi)數(shù)量，保證consumer不會(huì)出現(xiàn)飽和。劣勢(shì)在于當(dāng)沒有數(shù)據(jù)，會(huì)出現(xiàn)空輪詢，消耗cpu。

kafka采用pull，并采用可配置化參數(shù)保證當(dāng)存在數(shù)據(jù)并且數(shù)據(jù)量達(dá)到一定量的時(shí)候，consumer端才進(jìn)行pull操作，否則一直處于block狀態(tài)。kakfa采用整數(shù)值consumer position來記錄單個(gè)分區(qū)的消費(fèi)狀態(tài)，并且單個(gè)分區(qū)單個(gè)消息只能被consumer group內(nèi)的一個(gè)consumer消費(fèi)，維護(hù)簡(jiǎn)單開銷小。消費(fèi)完成，broker收到確認(rèn)，position指向下次消費(fèi)的offset。由于消息不會(huì)刪除，在完成消費(fèi)，position更新之后，consumer依然可以重置offset重新消費(fèi)歷史消息。

消息發(fā)送語義

producer視角

消息最多發(fā)送一次：producer異步發(fā)送消息，或者同步發(fā)消息但重試次數(shù)為0。

消息至少發(fā)送一次：producer同步發(fā)送消息，失敗、超時(shí)都會(huì)重試。

消息發(fā)且僅發(fā)一次：后續(xù)版本支持。

consumer視角

消息最多消費(fèi)一次：consumer先讀取消息，再確認(rèn)position，***處理消息。

消息至少消費(fèi)一次：consumer先讀取消息，再處理消息，***確認(rèn)position。

消息消費(fèi)且僅消費(fèi)一次。

注意：

如果消息處理后的輸出端(如db)能保證消息更新冪等性，則多次消費(fèi)也能保證exactly once語義。

如果輸出端能支持兩階段提交協(xié)議，則能保證確認(rèn)position和處理輸出消息同時(shí)成功或者同時(shí)失敗。

在消息處理的輸出端存儲(chǔ)更新后的position，保證了確認(rèn)position和處理輸出消息的原子性(簡(jiǎn)單、通用)。

可用性

在kafka中，正常情況下所有node處于同步中狀態(tài)，當(dāng)某個(gè)node處于非同步中狀態(tài)，也就意味著整個(gè)系統(tǒng)出問題，需要做容錯(cuò)處理。

同步中代表了：

該node與zookeeper能連通。

該node如果是follower，那么consumer position與leader不能差距太大(差額可配置)。

某個(gè)分區(qū)內(nèi)同步中的node組成一個(gè)集合，即該分區(qū)的ISR。

kafka通過兩個(gè)手段容錯(cuò)：

數(shù)據(jù)備份：以partition為單位備份，副本數(shù)可設(shè)置。當(dāng)副本數(shù)為N時(shí)，代表1個(gè)leader，N-1個(gè)followers，followers可以視為leader的consumer，拉取leader的消息，append到自己的系統(tǒng)中

failover：

1. 當(dāng)leader處于非同步中時(shí)，系統(tǒng)從followers中選舉新leader

2. 當(dāng)某個(gè)follower狀態(tài)變?yōu)榉峭街袝r(shí)，leader會(huì)將此follower剔除ISR，當(dāng)此follower恢復(fù)并完成數(shù)據(jù)同步之后再次進(jìn)入 ISR。

另外，kafka有個(gè)保障：當(dāng)producer生產(chǎn)消息時(shí)，只有當(dāng)消息被所有ISR確認(rèn)時(shí)，才表示該消息提交成功。只有提交成功的消息，才能被consumer消費(fèi)。

因此，當(dāng)有N個(gè)副本時(shí)，N個(gè)副本都在ISR中，N-1個(gè)副本都出現(xiàn)異常時(shí)，系統(tǒng)依然能提供服務(wù)。

假設(shè)N副本全掛了，node恢復(fù)后會(huì)面臨同步數(shù)據(jù)的過程，這期間ISR中沒有node，會(huì)導(dǎo)致該分區(qū)服務(wù)不可用。kafka采用一種降級(jí)措施來處理：選舉***個(gè)恢復(fù)的node作為leader提供服務(wù)，以它的數(shù)據(jù)為基準(zhǔn)，這個(gè)措施被稱為臟leader選舉。由于leader是主要提供服務(wù)的，kafka broker將多個(gè)partition的leader均分在不同的server上以均攤風(fēng)險(xiǎn)。每個(gè)parition都有l(wèi)eader，如果在每個(gè)partition內(nèi)運(yùn)行選主進(jìn)程，那么會(huì)導(dǎo)致產(chǎn)生非常多選主進(jìn)程。kakfa采用一種輕量級(jí)的方式：從broker集群中選出一個(gè)作為controller，這個(gè)controller監(jiān)控掛掉的broker，為上面的分區(qū)批量選主。

一致性

上面的方案保證了數(shù)據(jù)高可用，有時(shí)高可用是體現(xiàn)在對(duì)一致性的犧牲上。如果希望達(dá)到強(qiáng)一致性，可以采取如下措施：

禁用臟leader選舉，ISR沒有node時(shí)，寧可不提供服務(wù)也不要未完全同步的node。

設(shè)置最小ISR數(shù)量min_isr，保證消息至少要被min_isr個(gè)node確認(rèn)才能提交。

持久化

基于以下幾點(diǎn)事實(shí)，kafka重度依賴磁盤而非內(nèi)存來存儲(chǔ)消息。

硬盤便宜，內(nèi)存貴

順序讀+預(yù)讀取操作，能提高緩存***率

操作系統(tǒng)利用富余的內(nèi)存作為pagecache，配合預(yù)讀取(read-ahead)+寫回(write-back)技術(shù)，從cache讀數(shù)據(jù)，寫到cache就返回(操作系統(tǒng)后臺(tái)flush)，提高用戶進(jìn)程響應(yīng)速度

java對(duì)象實(shí)際大小比理想大小要大，使得將消息存到內(nèi)存成本很高

當(dāng)堆內(nèi)存占用不斷增加時(shí)，gc抖動(dòng)較大

基于文件順序讀寫的設(shè)計(jì)思路，代碼編寫簡(jiǎn)單

在持久化數(shù)據(jù)結(jié)構(gòu)的選擇上，kafka采用了queue而不是Btree

kafka只有簡(jiǎn)單的根據(jù)offset讀和append操作，所以基于queue操作的時(shí)間復(fù)雜度為O(1),而基于Btree操作的時(shí)間復(fù)雜度為O(logN)

在大量文件讀寫的時(shí)候，基于queue的read和append只需要一次磁盤尋址，而Btree則會(huì)涉及多次。磁盤尋址過程極大降低了讀寫性能