在我們旅行于數(shù)據(jù)海洋的途中，如果把 Kafka 比作是一艘承載無(wú)數(shù)信息航行的快船，前文《Kafka實(shí)戰(zhàn)漫談：大數(shù)據(jù)領(lǐng)域的不敗王者》已經(jīng)講述了如何搭建起這艘快船，讓它在起風(fēng)的早晨開始了第一次航行。

但隨著大浪的拍打，我們必須讓它做好準(zhǔn)備，以應(yīng)對(duì)那些未知的暴風(fēng)雨。

今天，我們來(lái)談?wù)勅绾巫屵@艘快船變得更強(qiáng)壯——讓它有能力在風(fēng)急浪高時(shí)穩(wěn)穩(wěn)地前行，不至于讓寶貴的數(shù)據(jù)貨物沉入海底。

在 Kafka 這艘數(shù)據(jù)游輪載著數(shù)據(jù)航行時(shí)，我們這些開發(fā)者——也就是船上的水手來(lái)說(shuō)，Kafka 集群的高可用性、消息消費(fèi)的一致性和延時(shí)隊(duì)列等都是確保數(shù)據(jù)航行安全的關(guān)鍵特性。

所以，拿起你的望遠(yuǎn)鏡，讓我們來(lái)一探 Kafka 高級(jí)知識(shí)的奧秘吧！

一、背景

說(shuō)到 Kafka，許多人第一反應(yīng)可能是：噢，一個(gè)消息中間件？

是的，但它遠(yuǎn)不止如此。Kafka 的力量在于它的可伸縮性、可靠性以及高吞吐。

很多在業(yè)務(wù)中使用過(guò) Kafka 的小伙伴肯定知道，搭建起一個(gè)高效的 Kafka 集群，就像在你的航船上裝備強(qiáng)大的引擎。

而恰當(dāng)?shù)靥幚硐⒌闹貜?fù)消費(fèi)與延時(shí)，則猶如在粗獷的海浪中找到了平穩(wěn)的航道。

(1) 堅(jiān)不可摧的集群艦隊(duì)：Kafka 高可用

在高海拔時(shí)，Kafka 能夠展現(xiàn)它的“高”——高可用。

集群化部署后，即便是有節(jié)點(diǎn)失聯(lián)，系統(tǒng)也能自我修復(fù)。如同海中的艦隊(duì)，一艘艦船的失事并不意味著整個(gè)航線的中斷。

(2) 精確把握所有貨物：消費(fèi)去重

重復(fù)消費(fèi)在消息隊(duì)列中猶如誤投遞包裹。

在 Kafka 的運(yùn)送數(shù)據(jù)時(shí)，要確保每條消息只被消費(fèi)一次，我們得有高超的“航海術(shù)”——冪等性與事務(wù)。

(3) 總有延時(shí)的風(fēng)浪：延時(shí)隊(duì)列處理

和海底的珍珠一樣，每個(gè)船上的數(shù)據(jù)包都有其“價(jià)值”，然而在時(shí)間的尺度上，它們的價(jià)值是變化的。

所有才有詩(shī)人說(shuō)：人生遇到的每個(gè)人，出場(chǎng)順序真的很重要！這在消息中間件里也同樣適用。

人生沒(méi)有時(shí)光機(jī)，但 Kafka 的延時(shí)隊(duì)列有：它不僅可以發(fā)送實(shí)時(shí)消息，還能處理那些需要“重復(fù)呈現(xiàn)”的信息。

二、乘風(fēng)破浪的Kafka

接下來(lái)，讓我們更深入地探索這艘快船的強(qiáng)大之處，以及如何利用它。首先，我們搭建一個(gè) Kafka 集群。

1.創(chuàng)建三個(gè)broker

我們啟動(dòng)了三個(gè)容器，配置都略微不同——記住，每艘船都有其獨(dú)特的身份（broker.id）和自己的泊位（listeners）。

通過(guò)進(jìn)入 Zookeeper 的客戶端檢查 Kafka 的狀態(tài)，就如同站在燈塔上檢查艦隊(duì)——確保每艘船都已就緒并在正確的位置。

啟動(dòng)三個(gè)容器，server.properties 文件分別為：

# 1,2,3
broker.id = 3
# 49092，49093,49094
listeners=PLAINTEXT://172.16.30.34:49094

校驗(yàn)是否啟動(dòng)成功，進(jìn)入 zookeeper，查看：

# 進(jìn)入客戶端
./bin/zkCli.sh

# 查看broker
ls /brokers/ids

可以看到各自的 ID 都不同，說(shuō)明 3 個(gè)節(jié)點(diǎn)啟動(dòng)成功了：

2. 集群中的副本

我們?cè)谏掀恼轮幸呀?jīng)介紹了分區(qū)和副本的概念，現(xiàn)在，結(jié)合集群中的節(jié)點(diǎn)，我們進(jìn)一步來(lái)理解它們。

有備無(wú)患，副本就好比是同型船的兄弟，一旦領(lǐng)軍的船只（leader）遇難，它們（follower）中的一員就會(huì)站出來(lái)，扛起大旗。

Kafka 的讀寫操作都發(fā)生在 leader 上，leader 還負(fù)責(zé)把數(shù)據(jù)同步給 foller。

當(dāng) leader 掛了，經(jīng)過(guò)主從選舉，從 follower 中選出一個(gè)新的 leader。

在這里，Kafka 的調(diào)度能力展露無(wú)遺。

3. 關(guān)于集群消費(fèi)

就像團(tuán)隊(duì)合作的捕魚作業(yè)，一個(gè) message 可能會(huì)被多個(gè)消費(fèi)者（在不同的船上）處理，但恰到好處的通信保證了魚群的精確分配。

在集群中，我們可以借助以下命令來(lái)發(fā)送、消費(fèi)消息：

# 向集群發(fā)送消息
./kafka-console-producer.sh --broker-list 172.16.30.34:49092,172.16.30.34:49093,172.16.30.34:49094 --topic my-replicated-topic

# 消費(fèi)集群消息
./kafka-console-consumer.sh --bootstrap-server 172.16.30.34:49092,172.16.30.34:49093,172.16.30.34:49094 --topic my-replicated-topic

# 指定消費(fèi)組消費(fèi)集群消息
./kafka-console-consumer.sh --bootstrap-server 172.16.30.34:49092,172.16.30.34:49093,172.16.30.34:49094 --from-beginning --consumer-property group.id=testGroup1 --topic my-replicated-topic

可以看到，消息發(fā)送成功了：

消息消費(fèi)成功：

三、kafka 集群中的關(guān)鍵角色

1. controller

控制器如船隊(duì)的指揮官，遇見(jiàn)有需要改變的情況時(shí)能及時(shí)做出應(yīng)答，無(wú)論是船只的增減，抑或是航線的變更。

每個(gè) broker 在啟動(dòng)時(shí)會(huì)向 zk 創(chuàng)建一個(gè)臨時(shí)序號(hào)節(jié)點(diǎn)【比如上面創(chuàng)建的broker節(jié)點(diǎn) 1,2,3】，獲得的最小序號(hào) broker 會(huì)作為集群中的 controller，負(fù)責(zé)以下幾件事：

• 當(dāng)集群中有一個(gè)副本的 leader 掛掉，需要在集群中選舉出一個(gè)新的 leader，選舉的規(guī)則是從 ISR 集合的最左邊元素獲?。ū热?ISR 集合為 【2,1,3】，當(dāng) leader 為 2 并且掛了時(shí)，ISR 為 【1,3】，就將 broker-1 上的副本作為新的 leader）；
• 當(dāng)集群中的 broker 新增或減少時(shí)，controller 會(huì)同步信息給其他 broker；
• 當(dāng)集群中有分區(qū)新增或減少時(shí)，controller 會(huì)同步信息給其他 broker。

2. rebalance 機(jī)制

每一個(gè)水手都有其特定的崗位，如同 Kafka 在消費(fèi)者與分區(qū)間實(shí)現(xiàn)的再平衡——這是一種資源優(yōu)化的藝術(shù)，和分配負(fù)載均衡的請(qǐng)求類似。

在 Kafka 中，再平衡需要一個(gè)前提就是：消費(fèi)組中的消費(fèi)者沒(méi)有指定分區(qū)來(lái)消費(fèi)。如果對(duì)消息指定了分區(qū)，rebalance 就不會(huì)生效。

并且，當(dāng)消費(fèi)組中的消費(fèi)組和分區(qū)關(guān)系發(fā)生變化時(shí)，rebalance 才會(huì)觸發(fā)。這時(shí)，消息的分區(qū)會(huì)遵循以下幾個(gè)策略中的一種（可配置）：

• range：根據(jù)公式計(jì)算得到每個(gè)消費(fèi)者去消費(fèi)哪個(gè)分區(qū)，前面的消費(fèi)者分區(qū) = 分區(qū)總數(shù)/消費(fèi)者數(shù)量+1，后面的消費(fèi)者 = 分區(qū)總數(shù)/消費(fèi)者數(shù)量；
• 輪詢：幾個(gè)消費(fèi)者輪流消費(fèi)分區(qū)；
• sticky：粘合策略，當(dāng)需要 rebalance 時(shí)，會(huì)在之前已經(jīng)分配的基礎(chǔ)上調(diào)整，且不會(huì)改變之前的分配情況。如果這個(gè)策略未打開，則需要重新進(jìn)行全部分區(qū)的分配。

3. HW 和 LEO

HW（high-weight，高水位）和 LEO（log-end-offset）是衡量副本最后消息位置的兩個(gè)重要指標(biāo)，它們就像是船上的測(cè)深儀，確保了數(shù)據(jù)不被過(guò)早或不當(dāng)?shù)靥幚怼?/p>

HW 是已完成 lead-follower 同步的位置，消費(fèi)者無(wú)法消費(fèi)到 HW 線之前的消息。并且，在完成同步以后，HW 線才更新，以防止消息丟失。

LEO 是指某個(gè)副本最后消費(fèi)消息的位置，根據(jù)木桶效應(yīng)，HW 一定不高于 LEO。

四、kafka 中的優(yōu)化問(wèn)題

1. 如何防止消息丟失

在 Kafka 的海域里，防止消息的丟失恰至關(guān)鍵。這就需要水手精準(zhǔn)的操作——生產(chǎn)者要如同技術(shù)精湛的引導(dǎo)者，消費(fèi)者像觀望遠(yuǎn)方的瞭望者，及時(shí)地做出反饋。

對(duì)于生產(chǎn)者來(lái)說(shuō)，可以采用以下方式來(lái)防止：

• 使用同步發(fā)送
• 把 ack 設(shè)為 1（0為異步進(jìn)行數(shù)據(jù)復(fù)制，-1為保證有一個(gè)副本復(fù)制完成，1為全同步）
• 同步的分區(qū)數(shù) >= 2

對(duì)于消費(fèi)者來(lái)說(shuō)，需要將自動(dòng)提交改為手動(dòng)提交，并且當(dāng)消費(fèi)完成后再進(jìn)行 ack 應(yīng)答。

這相當(dāng)于網(wǎng)絡(luò)中的握手過(guò)程，消息包收到以后，給出反饋；如果沒(méi)有收到消息，就讓發(fā)送端或者 Kafka 重新發(fā)一次，以防止消息還沒(méi)消費(fèi)就丟失了。

2. 如何防止重復(fù)消費(fèi)

再精確的海圖也免不了失誤時(shí)出現(xiàn)。為避免消息被重復(fù)消費(fèi)，生產(chǎn)者可能需要更謹(jǐn)慎，而消費(fèi)者需要有追蹤每條消息唯一性的能力。

為了防止消息丟失，當(dāng)生產(chǎn)者發(fā)送完消息后，會(huì)根據(jù)有無(wú)收到 ack 應(yīng)答去決定是否重新發(fā)送消息。

當(dāng)網(wǎng)絡(luò)抖動(dòng)或者其它原因，導(dǎo)致生產(chǎn)者沒(méi)有收到 ack 時(shí)，消費(fèi)者可能會(huì)收到兩條或多條相同的消息，造成重復(fù)消費(fèi)。

解決方案有以下兩種：

• 生產(chǎn)者關(guān)閉重試機(jī)制；
• 消費(fèi)者消費(fèi)消息時(shí)用冪等性保證：1）數(shù)據(jù)庫(kù)唯一索引；2）Redis 分布式鎖。

由于生產(chǎn)者關(guān)閉重試后，可能會(huì)造成消息丟失，所以我們更推薦讓消費(fèi)者用冪等性或者事務(wù)來(lái)防止重復(fù)消費(fèi)，這在其它的消息隊(duì)列中也同樣適用。

3. 如何做到消息的順序消費(fèi)

就像在粗獷的風(fēng)浪中維持航道一樣困難，Kafka 的有序消費(fèi)要求很高。有時(shí)需要犧牲性能，專注于單一的分區(qū)和消費(fèi)者。

• 生產(chǎn)者：使用同步發(fā)送，ack 設(shè)置為 1
• 消費(fèi)者：主題只能設(shè)置為一個(gè) partition 分區(qū)，消費(fèi)組中只能有一個(gè)消費(fèi)者

Kafka 的順序消費(fèi)會(huì)嚴(yán)重地犧牲性能，所以使用時(shí)需要做出權(quán)衡。

4. 如何解決消息積壓的問(wèn)題

消息積壓如同海上的風(fēng)暴，短時(shí)間內(nèi)的消費(fèi)者可能支撐不過(guò)生產(chǎn)者的速度，這時(shí)候可能需要更多的船只和船員加入作業(yè)。

當(dāng)消費(fèi)者的消費(fèi)速度，遠(yuǎn)遠(yuǎn)趕不上生產(chǎn)消息的速度一段時(shí)間后，kafka 會(huì)堆積大量未消費(fèi)的消息。

這會(huì)導(dǎo)致消費(fèi)者尋址的速度越來(lái)越慢，kafka 對(duì)外提供服務(wù)的性能也越來(lái)越差，從而可能會(huì)造成整個(gè)服務(wù)鏈變慢，導(dǎo)致服務(wù)雪崩。

一般，我們通過(guò)多線程或橫向擴(kuò)展的方式來(lái)解決：

• 消費(fèi)者使用多線程消費(fèi)，充分利用機(jī)器的性能；
• 在同一個(gè)消費(fèi)組中創(chuàng)建多個(gè)消費(fèi)者，部署到其它機(jī)器上，一起消費(fèi)。

5. 實(shí)現(xiàn)延時(shí)隊(duì)列

有時(shí)，信息需要在特定的時(shí)刻獲取，就比如火車票待支付訂單 30min 后會(huì)自動(dòng)過(guò)期。

Kafka 也可設(shè)定的延時(shí)隊(duì)列，就像是烹飪師傅精準(zhǔn)控制火候，讓美食在最佳時(shí)間完美呈現(xiàn)。

方案如下：

• Kafka 中創(chuàng)建相應(yīng)的主題，并創(chuàng)建消費(fèi)者消費(fèi)該主題的消息，消息中帶有創(chuàng)建的時(shí)間戳；
• 消費(fèi)消息時(shí)判斷，未支付訂單消息的創(chuàng)建時(shí)間是否已經(jīng)超過(guò) 30 分鐘：1）如果是，就修改訂單狀態(tài)為超時(shí)取消；2）否則，記錄當(dāng)前消息的 offset，并等待 1 分鐘后，再次向 kafka 中拉取該 offset 的消息進(jìn)行判斷，直到支付訂單或超時(shí)取消。

五、總結(jié)

過(guò)去，信號(hào)火一點(diǎn)即傳，如今，Kafka 使數(shù)據(jù)能在毫秒之間抵達(dá)世界的任何角落。

高可用性的秘密不在于單艦的堅(jiān)固，而在于整個(gè)艦隊(duì)的協(xié)同。

去重、延時(shí)、消費(fèi)順序的保證，使得 Kafka 成為一個(gè)強(qiáng)大且靈活的通信工具。

Kafka 也不僅僅是一個(gè)消息隊(duì)列，它是大數(shù)據(jù)時(shí)代的一艘高速航船，隨時(shí)準(zhǔn)備穿梭于信息的海洋，破浪前行。而我們，作為水手和指揮官，應(yīng)當(dāng)熟悉船上每一個(gè)細(xì)節(jié)，以確保我們的貨物——數(shù)據(jù)，能安全、準(zhǔn)確地到達(dá)預(yù)定的港口。

好了，現(xiàn)在，各位船長(zhǎng)，讓我們起航吧！