目錄

1、kafka是什么

一種高吞吐量的分布式、發(fā)布訂閱消息系統(tǒng)，它可以處理消費(fèi)者規(guī)模的，網(wǎng)站中的所有動(dòng)作流數(shù)據(jù)，具有高性能、持久化、多副本備份、橫向擴(kuò)展能力……

• 以時(shí)間復(fù)雜度為 O(1) 的方式提供消息持久化能力，即使對(duì) TB 級(jí)以上數(shù)據(jù)也能保證常數(shù)時(shí)間復(fù)雜度的訪問性能。
• 高吞吐率。即使在非常廉價(jià)的商用機(jī)器上也能做到單機(jī)支持每秒 100K 條以上消息的傳輸。
• 支持 Kafka Server 間的消息分區(qū)，及分布式消費(fèi)，同時(shí)保證每個(gè) Partition 內(nèi)的消息順序傳輸。
• 同時(shí)支持離線數(shù)據(jù)處理和實(shí)時(shí)數(shù)據(jù)處理。
• Scale out：支持在線水平擴(kuò)展。

image.png

2、消息從生產(chǎn)到消費(fèi)

2.1、發(fā)送數(shù)據(jù)

永遠(yuǎn)找leader！消息寫入leader后，follower是主動(dòng)的去leader進(jìn)行同步的！producer采用push模式將數(shù)據(jù)發(fā)布到broker，每條消息追加到分區(qū)中，順序?qū)懭氪疟P，所以保證同一分區(qū)內(nèi)的數(shù)據(jù)是有序的！

• partition在寫入的時(shí)候可以指定需要寫入的partition，如果有指定，則寫入對(duì)應(yīng)的partition。
• 如果沒有指定partition，但是設(shè)置了數(shù)據(jù)的key，則會(huì)根據(jù)key的值hash出一個(gè)partition。
• 如果既沒指定partition，又沒有設(shè)置key，則會(huì)輪詢選出一個(gè)partition。

image.png

發(fā)送數(shù)據(jù)可靠性保證：ACK機(jī)制！

0：代表producer往集群發(fā)送數(shù)據(jù)不需要等到集群的返回，不確保消息發(fā)送成功。安全性最低但是效率最高。
1：代表producer往集群發(fā)送數(shù)據(jù)只要leader應(yīng)答就可以發(fā)送下一條，只確保leader發(fā)送成功。
all：代表producer往集群發(fā)送數(shù)據(jù)需要所有的follower都完成從leader的同步才會(huì)發(fā)送下一條，確保leader發(fā)送成功和所有的副本都完成備份。安全性最高，但是效率最低。

敲黑板：所以這里可能是會(huì)丟消息的喲！

0：發(fā)丟了，生產(chǎn)端不知道，就是丟了。
1：保證leader不丟，但是如果leader掛了，恰好選了一個(gè)沒有ACK的follower，那也丟了。
all：保證leader和follower不丟，但是如果網(wǎng)絡(luò)擁塞，沒有收到ACK，會(huì)有重復(fù)發(fā)的問題。

2.2、保存數(shù)據(jù)

你以為是這樣的：

image.png

其實(shí)是這樣的：

image.png

操作系統(tǒng)本身有一層緩存，叫做 Page Cache，是在內(nèi)存里的緩存，我們也可以稱之為 OS Cache，意思就是操作系統(tǒng)自己管理的緩存。你在寫入磁盤文件的時(shí)候，可以直接寫入這個(gè) OS Cache 里，也就是僅僅寫入內(nèi)存中，接下來由操作系統(tǒng)自己決定什么時(shí)候把 OS Cache 里的數(shù)據(jù)真的刷入磁盤文件中。

Kafka提供了一個(gè)參數(shù)——producer.type來控制是不是主動(dòng)flush，如果Kafka寫入到mmap之后就立即flush然后再返回Producer叫 同步 (sync)；寫入mmap之后立即返回Producer不調(diào)用flush叫異步 (async)。

敲黑板：所以這里也可能是會(huì)丟消息的喲！

假如已經(jīng)寫在了OS cache中但是還沒來得及刷入磁盤，此時(shí)突然機(jī)器宕機(jī)，或者broker重啟，那就丟了。

2.3、消費(fèi)數(shù)據(jù)

消費(fèi)者通過pull模式主動(dòng)的去kafka集群拉取消息，與producer相同的是，消費(fèi)者在拉取消息的時(shí)候也是找leader去拉取。

多個(gè)消費(fèi)者可以組成一個(gè)消費(fèi)者組（consumer group），每個(gè)消費(fèi)者組都有一個(gè)組id！同一個(gè)消費(fèi)組者的消費(fèi)者可以消費(fèi)同一topic下不同分區(qū)的數(shù)據(jù)，但是不會(huì)組內(nèi)多個(gè)消費(fèi)者消費(fèi)同一分區(qū)的數(shù)據(jù)。

image.png

在早期的版本中，消費(fèi)者將消費(fèi)到的offset維護(hù)zookeeper中，consumer每間隔一段時(shí)間上報(bào)一次，這里容易導(dǎo)致重復(fù)消費(fèi)，且性能不好！在新的版本中消費(fèi)者消費(fèi)到的offset已經(jīng)直接維護(hù)在kafka集群的__consumer_offsets這個(gè)topic中！

消費(fèi)消息的時(shí)候可以大致分為兩個(gè)階段：1、標(biāo)示該消息已被消費(fèi)（commit記錄一下）；2、處理消息。

敲黑板：所以這里也可能是會(huì)丟消息的喲！

先commit，但是在處理消息的異常了，此時(shí)這條消息就丟了。
先處理，再commit，不丟。

3、消息可靠性保證

• At most once 消息可能會(huì)丟，但絕不會(huì)重復(fù)傳輸
• At least once 消息絕不會(huì)丟，但可能會(huì)重復(fù)傳輸
• Exactly once 每條消息肯定會(huì)被傳輸一次且僅傳輸一次，很多時(shí)候這是用戶所想要的

從 Producer 向 broker 發(fā)送消息時(shí)，通過ACK機(jī)制保證不丟消息，但是不可靠，依賴kafka的參數(shù)配置：

• 0：發(fā)丟了，生產(chǎn)端不知道，就是丟了，對(duì)應(yīng)At most once。
• 1：保證leader不丟，但是如果leader掛了，恰好選了一個(gè)沒有ACK的follower，那也丟了，對(duì)應(yīng)At most once。
• all：保證leader和follower不丟，但是如果網(wǎng)絡(luò)擁塞，沒有收到ACK，會(huì)有重復(fù)發(fā)的問題，對(duì)應(yīng)At least once。

在broker存儲(chǔ)消息時(shí)，通過主動(dòng)flush來保證可靠性，但是如果沒設(shè)置強(qiáng)制flush，那就有可能丟了。

從 broker 到 Consumer 消費(fèi)消息時(shí)，數(shù)據(jù)處理與 commit 的順序在很大程度上決定了消息從 broker 和 consumer 的可靠性：

• 讀完消息先 commit 再處理消息。這種模式下，如果 Consumer 在 commit 后還沒來得及處理消息就 crash 了，下次重新開始工作后就無法讀到剛剛已提交而未處理的消息，這就對(duì)應(yīng)于 At most once
• 讀完消息先處理再 commit。這種模式下，如果在處理完消息之后 commit 之前 Consumer crash 了，下次重新開始工作時(shí)還會(huì)處理剛剛未 commit 的消息，實(shí)際上該消息已經(jīng)被處理過了。這就對(duì)應(yīng)于 At least once。在很多使用場(chǎng)景下，消息的處理往往具有冪等性，即多次處理這一條消息跟只處理一次是等效的，那就可以認(rèn)為是 Exactly once。
• 如果一定要做到 Exactly once，就需要協(xié)調(diào) offset 和實(shí)際操作的輸出。精典的做法是引入兩階段提交。如果能讓 offset 和操作輸入存在同一個(gè)地方，會(huì)更簡(jiǎn)潔和通用。這種方式可能更好，因?yàn)樵S多輸出系統(tǒng)可能不支持兩階段提交。比如，Consumer 拿到數(shù)據(jù)后可能把數(shù)據(jù)放到 HDFS，如果把最新的 offset 和數(shù)據(jù)本身一起寫到 HDFS，那就可以保證數(shù)據(jù)的輸出和 offset 的更新要么都完成，要么都不完成，間接實(shí)現(xiàn) Exactly once。