[[422790]]

本文轉(zhuǎn)載自微信公眾號(hào)「明哥的IT隨筆」，作者 IT明哥 。轉(zhuǎn)載本文請(qǐng)聯(lián)系明哥的IT隨筆公眾號(hào)。

1 前言

大家好，我是明哥!

KAFKA 作為開(kāi)源分布式事件流平臺(tái)，在大數(shù)據(jù)和微服務(wù)領(lǐng)域都有著廣泛的應(yīng)用場(chǎng)景，是實(shí)時(shí)流處理場(chǎng)景下消息隊(duì)列事實(shí)上的標(biāo)準(zhǔn)。用一句話(huà)概括，KAFKA 是實(shí)時(shí)數(shù)倉(cāng)的基石，是事件驅(qū)動(dòng)架構(gòu)的靈魂。

但是一些技術(shù)小伙伴，尤其是一些很早就開(kāi)始使用 KAFKA 的技術(shù)小伙伴們，對(duì) KAFKA 的發(fā)展趨勢(shì)和一些新特性，并不太熟悉，在使用過(guò)程中也踩了不少坑。

有鑒于此，我們接下來(lái)會(huì)有一個(gè) KAFKA 系列文章，專(zhuān)門(mén)講述 KAFKA 的這些新特性。

本文是該系列文章之一，講述 KAFAK 的冪等生產(chǎn)者。

以下是正文。

2 從歷史視角看 KAFKA 的發(fā)展

首先我們從歷史視角，看下 KAFKA 的發(fā)展：

• KAFKA 在2013年12月推出了一個(gè)重要的版本 0.8.0，該版本相當(dāng)重要，因?yàn)樗ㄟ^(guò) KAFKA-50 首次引進(jìn)了多副本機(jī)制，為容錯(cuò)打下了堅(jiān)實(shí)的基礎(chǔ);
• 然后在后續(xù)版本中逐步增添了很多新的功能特性：
  • 如逐步擺脫對(duì) zookeeper的依賴(lài);
  • 如支持 compact 清理策略;
  • 如支持 kafka tired storage;
  • 如生產(chǎn)者冪等性;
  • 如對(duì)事務(wù)的支持;
  • 如大的 kafka 生態(tài)的 kafka connect api, kafka stream api 以及 KSQL, 還有 kafka schema registry;
• 到目前為止(202109)，KAFKA 最新的穩(wěn)定版已經(jīng)演進(jìn)到了 2.8.0;
• KAFKA 已經(jīng)從最開(kāi)始僅僅作為一個(gè)高吞吐的消息中間件，發(fā)展到了如今實(shí)時(shí)流處理場(chǎng)景下消息隊(duì)列事實(shí)上的標(biāo)準(zhǔn)，用一句話(huà)概括，KAFKA 是實(shí)時(shí)數(shù)倉(cāng)的基石，是事件驅(qū)動(dòng)架構(gòu)的靈魂。
• 但是如今在市面上生產(chǎn)環(huán)境中，還不乏有使用早期版本如 0.8.0 版本的情況。

kafka-timeline

kafka-api

3 什么是冪等生產(chǎn)者?

我們知道，當(dāng) kafka producer 向 broker 中的 topic發(fā)送數(shù)據(jù)時(shí)，可能會(huì)因?yàn)榫W(wǎng)絡(luò)抖動(dòng)等各種原因，造成 producer 收不到 broker 的 ack 確認(rèn)信息。此時(shí) producer 有兩種選擇：

producer 可以選擇忽略沒(méi)有收到 ack 確認(rèn)消息，不做任何進(jìn)一步處理：此時(shí)有可能會(huì)丟失消息。(之所以說(shuō)有可能，是因?yàn)橄⒂锌赡軟](méi)有寫(xiě)到 broker 的topic 中，但也有可能已經(jīng)正確地寫(xiě)到了 broker 的 topic 中，只是回調(diào)的 ack 消息因網(wǎng)絡(luò)抖動(dòng) producer 沒(méi)有收到;)

producer 也可以選擇多次嘗試重發(fā)消息，直到收到ack 確認(rèn)消息或重試最大次數(shù)到達(dá): 此時(shí)有可能會(huì)造成消息的重復(fù)寫(xiě)，即 broker 端的 topic 中，重復(fù)地存儲(chǔ)了重試發(fā)送的這些消息;

producer 重發(fā)沒(méi)有收到 ack 確認(rèn)的消息, 也可能會(huì)造成 broker 端 topic 的 partition 中 消息的順序混亂，即因失敗重發(fā)的消息在部分沒(méi)有失敗不需要重發(fā)的消息之后。

因 producer 重發(fā)沒(méi)有收到 ack 確認(rèn)的消息造成數(shù)據(jù)重復(fù)的問(wèn)題，可以參見(jiàn)如下示意圖，圖中 message 7/8/9/10 即為重復(fù)的消息。

producer-resend-failure

KAFKA 的冪等生產(chǎn)者即 idempotent producer，就是解決上述問(wèn)題的：它可以確保消息被正確地投遞到 broker端，不會(huì)丟失沒(méi)有重復(fù)，而且是以正確的順序存儲(chǔ)在 topic 的各個(gè) partition 中。

4 如何啟用冪等生產(chǎn)者?

• 啟用冪等生產(chǎn)者，不涉及任何代碼層面的改動(dòng)，只涉及以下配置項(xiàng)的更改:
• enable.idempotence=true;//冪等生產(chǎn)者功能開(kāi)關(guān)
• message.send.max.retries=xx //發(fā)送失敗重試次數(shù)，可以配置很大比如10000000，甚至Integer.MAX_VALUE;
• max.in.flight.requests.per.connection=xx //xx <= 5, 代表每個(gè)連接中在途請(qǐng)求次數(shù)，有的博文說(shuō)該參數(shù)必須配置為=1，其實(shí)不然，只需要<=5即可(max.in.flight must be set <= 5 when enable.idempotence is true");
• Acks=All //ACK 確認(rèn)參數(shù)，可選 0/1/-1/ALL，-1 與 ALL 等價(jià)。在開(kāi)啟冪等生產(chǎn)者功能時(shí)，該參數(shù)必須配置為ALL/-1，即所有 ISR 都要確認(rèn)收到了消息，才認(rèn)為消息投遞成功(acks must be set to all when enable.idempotence is true");
• 在開(kāi)啟冪等生產(chǎn)者即 enable.idempotence=true 的情況下，也可以不配置參數(shù) max.in.flight.requests.per.connection 和參數(shù) Acks，此時(shí)這兩個(gè)參數(shù)會(huì)被自動(dòng)配置;

5 冪等生產(chǎn)者的原理是什么?

首先需要說(shuō)明下，在啟用冪等生產(chǎn)者的情況下，消息失敗時(shí)的重新發(fā)送，是由 kafka client 自動(dòng)實(shí)現(xiàn)的，對(duì)我們來(lái)講是透明的，我們不需要在代碼中重試發(fā)送。(事實(shí)上，在代碼中重試消息發(fā)送，反而會(huì)引起消息重復(fù)).

其內(nèi)部工作原理如下：

• 在 producer 端，每個(gè) producer 都被 broker 自動(dòng)分配了一個(gè) Producer Id (PID)， producer 向 broker 發(fā)送的每條消息，在內(nèi)部都附帶著該 pid 和一個(gè)遞增的 sequence number;
• 在 broker 端，broker 為每個(gè) topic 的每個(gè) partition 都維護(hù)了一個(gè)當(dāng)前寫(xiě)成功的消息的最大 PID-Sequence Number 元組;
• 當(dāng) broker 收到一個(gè)比當(dāng)前最大 PID-Sequence Number 元組小的 sequence number 消息時(shí)，就會(huì)丟棄該消息，以避免造成數(shù)據(jù)重復(fù)存儲(chǔ);
• 當(dāng) broker 失敗重新選舉新的 leader 時(shí), 以上去重機(jī)制仍然有效：因?yàn)?broker 的 topic 中存儲(chǔ)的消息體中附帶了 PID-sequence number 信息，且 leader 的所有消息都會(huì)被復(fù)制到 followers 中。當(dāng)某個(gè)原來(lái)的 follower 被選舉為新的 leader 時(shí)，它內(nèi)部的消息中已經(jīng)存儲(chǔ)了PID-sequence number 信息，也就可以執(zhí)行消息去重了。
• 冪等生產(chǎn)者，在 broker 端去重的工作原理，如下圖所示：圖片

6 冪等生產(chǎn)者與事務(wù)有何關(guān)系?

冪等生產(chǎn)者是 kafka 事務(wù)的必要不充分條件，即：

開(kāi)啟冪等生長(zhǎng)者，不一定需要開(kāi)啟事務(wù);

開(kāi)始 kafka 事務(wù)，必須要開(kāi)啟冪等生產(chǎn)者;

事實(shí)上，開(kāi)啟 kafka事務(wù)時(shí)，kafka 會(huì)自動(dòng)開(kāi)啟冪等生產(chǎn)者。