Kafka

kafka默認(rèn)情況下，提供的是至少一次的可靠性保障。即broker保障已提交的消息的發(fā)送，但是遇上某些意外情況

如：網(wǎng)絡(luò)抖動(dòng)，超時(shí)等問題，導(dǎo)致Producer沒有收到broker返回的數(shù)據(jù)ack，則Producer會(huì)繼續(xù)重試發(fā)送消息，從而導(dǎo)致消息重復(fù)發(fā)送。

如果我們禁止Producer的失敗重試發(fā)送功能，消息要么寫入成功，要么寫入失敗，但絕不會(huì)重復(fù)發(fā)送。

這樣就是最多一次的消息保障模式。但對(duì)于消息組件，排除特殊業(yè)務(wù)場(chǎng)景，我們追求的一定是精確一次的消息保障模式。

kafka通過 冪等性（Idempotence）和事務(wù)（Transaction） 的機(jī)制，提供了這種精確的消息保障。

在之前的舊版本中，Kafka只能支持兩種語義：At most once和At least once。而Kafka在 0.11.0.0 版本支持增加了對(duì)冪等的支持。冪等是針對(duì)生產(chǎn)者角度的特性。冪等可以保證上生產(chǎn)者發(fā)送的消息，不會(huì)丟失，而且不會(huì)重復(fù)。

冪等性要解決的問題？

在 0.11.0 之前，Kafka 通過 Producer 端和 Server 端的相關(guān)配置可以做到 數(shù)據(jù)不丟 ，也就是 at least once，但是在一些情況下，可能會(huì)導(dǎo)致數(shù)據(jù)重復(fù)

比如：網(wǎng)絡(luò)請(qǐng)求延遲等導(dǎo)致的重試操作，在發(fā)送請(qǐng)求重試時(shí) Server 端并不知道這條請(qǐng)求是否已經(jīng)處理（沒有記錄之前的狀態(tài)信息）

所以就會(huì)有可能導(dǎo)致數(shù)據(jù)請(qǐng)求的重復(fù)發(fā)送，這是 Kafka 自身的機(jī)制（異常時(shí)請(qǐng)求重試機(jī)制）導(dǎo)致的數(shù)據(jù)重復(fù)。

對(duì)于大多數(shù)應(yīng)用而言，數(shù)據(jù)保證不丟是可以滿足其需求的，但是對(duì)于一些其他的應(yīng)用場(chǎng)景（比如支付數(shù)據(jù)等），它們是要求精確計(jì)數(shù)的，這時(shí)候如果上游數(shù)據(jù)有重復(fù)，下游應(yīng)用只能在消費(fèi)數(shù)據(jù)時(shí)進(jìn)行相應(yīng)的去重操作，應(yīng)用在去重時(shí)，最常用的手段就是根據(jù)唯一 id 鍵做 check 去重。

在這種場(chǎng)景下，因?yàn)樯嫌紊a(chǎn)導(dǎo)致的數(shù)據(jù)重復(fù)問題，會(huì)導(dǎo)致所有有精確計(jì)數(shù)需求的下游應(yīng)用都需要做這種復(fù)雜的、重復(fù)的去重處理。

試想一下：如果在發(fā)送時(shí)，系統(tǒng)就能保證 exactly once，這對(duì)下游將是多么大的解脫。

這就是冪等性要解決的問題，主要是解決數(shù)據(jù)重復(fù)的問題，正如前面所述，數(shù)據(jù)重復(fù)問題，通用的解決方案就是加唯一 id，然后根據(jù) id 判斷數(shù)據(jù)是否重復(fù)，Producer 的冪等性也是這樣實(shí)現(xiàn)的。

Kafka 是怎么保證冪等性的？

Kafka為了實(shí)現(xiàn)冪等性，它在底層設(shè)計(jì)架構(gòu)中引入了ProducerID和SequenceNumber。

• ProducerID：在每個(gè)新的Producer初始化時(shí)，會(huì)被分配一個(gè)唯一的ProducerID，這個(gè)ProducerID對(duì)客戶端使用者是不可見的。
• SequenceNumber：對(duì)于每個(gè)ProducerID，Producer發(fā)送數(shù)據(jù)的每個(gè)Topic和Partition都對(duì)應(yīng)一個(gè)從0開始單調(diào)遞增的SequenceNumber值。

當(dāng)Producer發(fā)送消息(x2,y2)給Broker時(shí)，Broker接收到消息并將其追加到消息流中。此時(shí)，Broker返回Ack信號(hào)給Producer時(shí)，發(fā)生異常導(dǎo)致Producer接收Ack信號(hào)失敗。

對(duì)于Producer來說，會(huì)觸發(fā)重試機(jī)制，將消息(x2,y2)再次發(fā)送，但是，由于引入了冪等性，在每條消息中附帶了PID（ProducerID）和SequenceNumber。

相同的PID和SequenceNumber發(fā)送給Broker，而之前Broker緩存過之前發(fā)送的相同的消息，那么在消息流中的消息就只有一條(x2,y2)，不會(huì)出現(xiàn)重復(fù)發(fā)送的情況。

開啟冪等性配置

只需要把 Producer 的配置 enable.idempotence 設(shè)置為 true 即可

props.put(“enable.idempotence”, ture)
//或者
props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG， true)

Kafka冪等性的局限性

開啟enable.idempotence后，kafka就會(huì)自動(dòng)幫你做好消息去重的一系列工作。底層具體實(shí)現(xiàn)原理很簡(jiǎn)單，就是用空間換時(shí)間的優(yōu)化思路，即在broker端多存一些字段來標(biāo)識(shí)數(shù)據(jù)的唯一性。當(dāng)Producer發(fā)送了具有相同字段值的消息后，broker會(huì)進(jìn)行匹配去重，丟棄重復(fù)的數(shù)據(jù)。實(shí)際的代碼沒這么簡(jiǎn)單，但大致是這么個(gè)處理邏輯。

官方的這個(gè)冪等實(shí)現(xiàn)看似簡(jiǎn)單高效，但也存在他的局限性。他只能保證單分區(qū)上的冪等性，即一個(gè)冪等性Producer只能夠保證某個(gè)topic的一個(gè)分區(qū)上不出現(xiàn)重復(fù)消息，無法實(shí)現(xiàn)多分區(qū)的冪等。此外，如果Producer重啟，也會(huì)導(dǎo)致冪等重置。

事務(wù)

對(duì)于多分區(qū)保證冪等的場(chǎng)景，則需要事務(wù)特性來處理了。

kafka的事務(wù)跟我們常見數(shù)據(jù)庫(kù)事務(wù)概念差不多，也是提供經(jīng)典的ACID，即原子（Atomicity）、一致性 (Consistency)、隔離性 (Isolation) 和持久性 (Durability)。

事務(wù)Producer保證消息寫入分區(qū)的原子性，即這批消息要么全部寫入成功，要么全失敗。

此外，Producer重啟回來后，kafka依然保證它們發(fā)送消息的精確一次處理。事務(wù)特性的配置也很簡(jiǎn)單：

和冪等Producer一樣，開啟enable.idempotence = true設(shè)置Producer端參數(shù)transctional.id事務(wù)Producer的代碼稍微也有點(diǎn)不一樣，需要調(diào)一些事務(wù)處理的API。

數(shù)據(jù)的發(fā)送需要放在beginTransaction和commitTransaction之間。Consumer端的代碼也需要加上isolation.level參數(shù)，用以處理事務(wù)提交的數(shù)據(jù)。示例代碼:

producer.initTransactions();
try {
     producer.beginTransaction();
     producer.send(record1);
     producer.send(record2);
     producer.commitTransaction();
} catch (KafkaException e) {
     producer.abortTransaction();
}

事務(wù)Producer雖然在多分區(qū)的數(shù)據(jù)處理上保證了冪等，但是處理性能上相應(yīng)的是會(huì)有一些下降的。

到此這篇關(guān)于Java消息隊(duì)列中的Kafka如何保證冪等性的文章就介紹到這了,更多相關(guān)Java的Kafka保證冪等性內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

最新評(píng)論