Kafka冪等性原理及實現剖析
1.概述
最近和一些同學交流的時候反饋說，在面試Kafka時，被問到Kafka元件組成部分、API使用、Consumer和Producer原理及作用等問題都能詳細作答。但是，問到一個平時不注意的問題，就是Kafka的冪等性，被卡主了。那麼，今天筆者就為大家來剖析一下Kafka的冪等性原理及實現。
2.內容
2.1 Kafka為啥需要冪等性？
Producer在生產傳送訊息時，難免會重複傳送訊息。Producer進行retry時會產生重試機制，發生訊息重複傳送。而引入冪等性後，重複傳送只會生成一條有效的訊息。Kafka作為分散式訊息系統，它的使用場景常見與分散式系統中，比如訊息推送系統、業務平臺系統（如物流平臺、銀行結算平臺等）。以銀行結算平臺來說，業務方作為上游把資料上報到銀行結算平臺，如果一份資料被計算、處理多次，那麼產生的影響會很嚴重。
2.2 影響Kafka冪等性的因素有哪些？
在使用Kafka時，需要確保Exactly-Once語義。分散式系統中，一些不可控因素有很多，比如網路、OOM、FullGC等。在Kafka Broker確認Ack時，出現網路異常、FullGC、OOM等問題時導致Ack超時，Producer會進行重複傳送。可能出現的情況如下：
2.3 Kafka的冪等性是如何實現的？
Kafka為了實現冪等性，它在底層設計架構中引入了ProducerID和SequenceNumber。那這兩個概念的用途是什麼呢？
- ProducerID：在每個新的Producer初始化時，會被分配一個唯一的ProducerID，這個ProducerID對客戶端使用者是不可見的。
- SequenceNumber：對於每個ProducerID，Producer傳送資料的每個Topic和Partition都對應一個從0開始單調遞增的SequenceNumber值。
2.3.1 冪等性引入之前的問題？
Kafka在引入冪等性之前，Producer向Broker傳送訊息，然後Broker將訊息追加到訊息流中後給Producer返回Ack訊號值。實現流程如下：
上圖的實現流程是一種理想狀態下的訊息傳送情況，但是實際情況中，會出現各種不確定的因素，比如在Producer在傳送給Broker的時候出現網路異常。比如以下這種異常情況的出現：
上圖這種情況，當Producer第一次傳送訊息給Broker時，Broker將訊息(x2,y2)追加到了訊息流中，但是在返回Ack訊號給Producer時失敗了（比如網路異常） 。此時，Producer端觸發重試機制，將訊息(x2,y2)重新發送給Broker，Broker接收到訊息後，再次將該訊息追加到訊息流中，然後成功返回Ack訊號給Producer。這樣下來，訊息流中就被重複追加了兩條相同的(x2,y2)的訊息。
2.3.2 冪等性引入之後解決了什麼問題？
面對這樣的問題，Kafka引入了冪等性。那麼冪等性是如何解決這類重複傳送訊息的問題的呢？下面我們可以先來看看流程圖：
同樣，這是一種理想狀態下的傳送流程。實際情況下，會有很多不確定的因素，比如Broker在傳送Ack訊號給Producer時出現網路異常，導致傳送失敗。異常情況如下圖所示：
當Producer傳送訊息(x2,y2)給Broker時，Broker接收到訊息並將其追加到訊息流中。此時，Broker返回Ack訊號給Producer時，發生異常導致Producer接收Ack訊號失敗。對於Producer來說，會觸發重試機制，將訊息(x2,y2)再次傳送，但是，由於引入了冪等性，在每條訊息中附帶了PID（ProducerID）和SequenceNumber。相同的PID和SequenceNumber傳送給Broker，而之前Broker快取過之前傳送的相同的訊息，那麼在訊息流中的訊息就只有一條(x2,y2)，不會出現重複傳送的情況。
2.3.3 ProducerID是如何生成的？
客戶端在生成Producer時，會例項化如下程式碼：
// 例項化一個Producer物件 Producer<String, String> producer = new KafkaProducer<>(props);
在org.apache.kafka.clients.producer.internals.Sender類中，在run()中有一個maybeWaitForPid()方法，用來生成一個ProducerID，實現程式碼如下：
private void maybeWaitForPid() {
if (transactionState == null)
return;
while (!transactionState.hasPid()) {
try {
Node node = awaitLeastLoadedNodeReady(requestTimeout);
if (node != null) {
ClientResponse response = sendAndAwaitInitPidRequest(node);
if (response.hasResponse() && (response.responseBody() instanceof InitPidResponse)) {
InitPidResponse initPidResponse = (InitPidResponse) response.responseBody();
transactionState.setPidAndEpoch(initPidResponse.producerId(), initPidResponse.epoch());
} else {
log.error("Received an unexpected response type for an InitPidRequest from {}. " +
"We will back off and try again.", node);
}
} else {
log.debug("Could not find an available broker to send InitPidRequest to. " +
"We will back off and try again.");
}
} catch (Exception e) {
log.warn("Received an exception while trying to get a pid. Will back off and retry.", e);
}
log.trace("Retry InitPidRequest in {}ms.", retryBackoffMs);
time.sleep(retryBackoffMs);
metadata.requestUpdate();
}
}
3.事務
與冪等性有關的另外一個特性就是事務。Kafka中的事務與資料庫的事務類似，Kafka中的事務屬性是指一系列的Producer生產訊息和消費訊息提交Offsets的操作在一個事務中，即原子性操作。對應的結果是同時成功或者同時失敗。
這裡需要與資料庫中事務進行區別，操作資料庫中的事務指一系列的增刪查改，對Kafka來說，操作事務是指一系列的生產和消費等原子性操作。
3.1 Kafka引入事務的用途？
在事務屬性引入之前，先引入Producer的冪等性，它的作用為：
- Producer多次傳送訊息可以封裝成一個原子性操作，即同時成功，或者同時失敗；
- 消費者&生產者模式下，因為Consumer在Commit Offsets出現問題時，導致重複消費訊息時，Producer重複生產訊息。需要將這個模式下Consumer的Commit Offsets操作和Producer一系列生產訊息的操作封裝成一個原子性操作。
產生的場景有：
比如，在Consumer中Commit Offsets時，當Consumer在消費完成時Commit的Offsets為100（假設最近一次Commit的Offsets為50），那麼執行觸發Balance時，其他Consumer就會重複消費訊息（消費的Offsets介於50~100之間的訊息）。
3.2 事務提供了哪些可使用的API？
Producer提供了五種事務方法，它們分別是：initTransactions()、beginTransaction()、sendOffsetsToTransaction()、commitTransaction()、abortTransaction()，程式碼定義在org.apache.kafka.clients.producer.Producer<K,V>介面中，具體定義介面如下：
// 初始化事務，需要注意確保transation.id屬性被分配
void initTransactions();
// 開啟事務
void beginTransaction() throws ProducerFencedException;
// 為Consumer提供的在事務內Commit Offsets的操作
void sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata> offsets,
String consumerGroupId) throws ProducerFencedException;
// 提交事務
void commitTransaction() throws ProducerFencedException;
// 放棄事務，類似於回滾事務的操作
void abortTransaction() throws ProducerFencedException;
3.3 事務的實際應用場景有哪些？
在Kafka事務中，一個原子性操作，根據操作型別可以分為3種情況。情況如下：
- 只有Producer生產訊息，這種場景需要事務的介入；
- 消費訊息和生產訊息並存，比如Consumer&Producer模式，這種場景是一般Kafka專案中比較常見的模式，需要事務介入；
- 只有Consumer消費訊息，這種操作在實際專案中意義不大，和手動Commit Offsets的結果一樣，而且這種場景不是事務的引入目的。
4.總結
Kafka的冪等性和事務是比較重要的特性，特別是在資料丟失和資料重複的問題上非常重要。Kafka引入冪等性，設計的原理也比較好理解。而事務與資料庫的事務特性類似，有資料庫使用的經驗對理解Kafka的事務也比較容易接受。
5.結束語
這篇部落格就和大家分享到這裡，如果大家在研究學習的過程當中有什麼問題，可以加群進行討論或傳送郵件給我，我會盡我所能為您解答，與君共勉！
另外，博主出書了《Kafka並不難學》和《Hadoop大資料探勘從入門到進階實戰》，喜歡的朋友或同學， 可以在公告欄那裡點選購買連結購買博主的書進行學習，在此感謝大家的支援。關注下面公眾號，根據提示，可免費獲取書籍的教學視訊。&n