Kafka

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

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

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

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

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

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

冪等性要解決的問題？

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

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

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

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

在這種場景下，因為上游生產(chǎn)導致的數(shù)據(jù)重復問題，會導致所有有精確計數(shù)需求的下游應用都需要做這種復雜的、重復的去重處理。

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

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

Kafka 是怎么保證冪等性的？

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

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

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

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

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

開啟冪等性配置

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

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

Kafka冪等性的局限性

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

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

事務

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

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

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

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

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

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

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

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

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

最新評論