01 冪等性如此重要

Kafka作為分佈式MQ，大量用於分佈式系統中，如消息推送系統、業務平臺系統（如結算平臺），就拿結算來說，業務方作為上游把數據打到結算平臺，如果一份數據被計算、處理了多次，產生的後果將會特別嚴重。

02 哪些因素影響冪等性

使用Kafka時,需要保證exactly-once語義。要知道在分佈式系統中，出現網絡分區是不可避免的，如果kafka broker 在回覆ack時，出現網絡故障或者是full gc導致ack timeout，producer將會重發，如何保證producer重試時不造成重複or亂序？又或者producer 掛了，新的producer並沒有old producer的狀態數據，這個時候如何保證冪等？即使Kafka 發送消息滿足了冪等，consumer拉取到消息後，把消息交給線程池workers，workers線程對message的處理可能包含異步操作，又會出現以下情況：

先commit，再執行業務邏輯：提交成功，處理失敗。造成丟失
先執行業務邏輯，再commit：提交失敗，執行成功。造成重複執行
先執行業務邏輯，再commit：提交成功，異步執行fail。造成丟失

本文將針對以上問題作出討論

03 Kafka保證發送冪等性

針對以上的問題，kafka在0.11版新增了冪等型producer和事務型producer。前者解決了單會話冪等性等問題，後者解決了多會話冪等性。

單會話冪等性

為解決producer重試引起的亂序和重複。Kafka增加了pid和seq。Producer中每個RecordBatch都有一個單調遞增的seq; Broker上每個tp也會維護pid-seq的映射，並且每Commit都會更新lastSeq。這樣recordBatch到來時，broker會先檢查RecordBatch再保存數據：如果batch中 baseSeq(第一條消息的seq)比Broker維護的序號(lastSeq)大1，則保存數據，否則不保存(inSequence方法)。

引申：Kafka producer 對有序性做了哪些處理

假設我們有5個請求，batch1、batch2、batch3、batch4、batch5；如果只有batch2 ack failed，3、4、5都保存了，那2將會隨下次batch重發而造成重複。我們可以設置max.in.flight.requests.per.connection=1（客戶端在單個連接上能夠發送的未響應請求的個數）來解決亂序，但降低了系統吞吐。

新版本kafka設置enable.idempotence=true後能夠動態調整max-in-flight-request。正常情況下max.in.flight.requests.per.connection大於1。當重試請求到來且時，batch 會根據 seq重新添加到隊列的合適位置，並把max.in.flight.requests.per.connection設為1，這樣它前面的 batch序號都比它小，只有前面的都發完了，它才能發。

多會話冪等性

在單會話冪等性中介紹，kafka通過引入pid和seq來實現單會話冪等性，但正是引入了pid，當應用重啟時，新的producer並沒有old producer的狀態數據。可能重複保存。

Kafka事務通過隔離機制來實現多會話冪等性

kafka事務引入了transactionId 和Epoch，設置transactional.id後，一個transactionId只對應一個pid, 且Server 端會記錄最新的 Epoch 值。這樣有新的producer初始化時，會向TransactionCoordinator發送InitPIDRequest請求， TransactionCoordinator 已經有了這個 transactionId對應的 meta，會返回之前分配的 PID，並把 Epoch 自增 1 返回，這樣當old producer恢復過來請求操作時，將被認為是無效producer拋出異常。如果沒有開啟事務，TransactionCoordinator會為新的producer返回new pid，這樣就起不到隔離效果，因此無法實現多會話冪等。

04 Consumer端冪等性

如上所述，consumer拉取到消息後，把消息交給線程池workers，workers對message的handle可能包含異步操作，又會出現以下情況：