分區分配策略
客戶端可以自定義分區分配策略, 當然也需要考慮分區消費之后的offset提交, 是否有沖突。
消費者協調器和組協調器
a. 消費者的不同分區策略, 消費者之間的負載均衡(新消費者加入或者存量消費者退出), 需要broker做必要的協調。
b. Kafka按照消費組管理消費者, 鑒于offset提交最終都是在某個broker節點上完成。該broker扮演GroupCoordinator角色, 具體的選擇則是通過hash快速定位。
c. client端存在一個ClientCoordinator與目標的GroupCoordinator進行通信實現最終協調;
d. 具體過程如下
關于__consumer_offset
__consumer_offset是一個特殊的topic, 用于存儲每個topic中partition中client提交的offset。其中的數據保留時間通過offset.retention.minutes配置。如果consumer消費消息的間隔超過了配置時間, 則offset會丟失, consumer再次獲取offset時會因為沒有存量的offset而自動重置(auto.offset.reset)。該topic下的消息清理采用壓縮策略(僅保留最新消息)。Kafka中會有定時清理任務清理過期的消費位移。
消息發送QoS
- at-least-once, 至少一次, 消息不會丟失, 但消息會重復;
- at-most-once, 至多一次, 消息不會重復, 但可能會丟失;
- exact-once, 恰好一次, 消息肯定被傳輸且只傳輸一次;(如果開發即時消息系統, 那么這個語義就是我們的目標)
默認情況下, Kafka producer在發送時, 如果消息發送失敗會自動進行重試, 重試過程可能會導致消息重復。而一旦發送成功, Kafka通過多副本機制保證消息一定會被保存。因此從consumer角度觀察, producer發送的結果, 其QoS是at-least-once。如果需要exact-once, 則需要啟用Kafka的冪等特性。
冪等
-
配置參數
enable.idompotence=true
retrics > 0
max.in.flight.requests.per.connection <=5
ack=-1 -
實現細節
首先冪等是partition級別, broker端自動為producer分配一個PID, 并維護PID->分區(序列號 lastSeq) 的狀態。當producer發送消息時, 必須攜帶該序列號newSeq。broker端收到消息時做校驗:
a. newSeq = lastSeq+1, broker接收;
b. newSeq > lastSeq+1, 中間存在消息丟失, 拋出OutOfOrderException;
c. newSeq < lastSeq+1, 消息存在重復, 直接丟棄即可.
事務消息
如果要實現跨parition的exact-once語義, 則需要基于事務消息。一般來說事務有ACID的特性, 但這個是數據庫事務的通用場景。Kafka下消息需要考慮生產和消費, 這里的事務消息更多是生產端的事務消息。消費端可能會因為某些原因無法以事務的形式消費。比如:
- 對于采用日志壓縮策略的主題而言, 事務中的消息被清理（對相同key的消息后寫入的消息會覆蓋之前寫入的消息）;
- 事務涉及的分區多個日志段, 如果老的日志分段被刪除, 對應的消息也會消失;
- 消費者通過seek消費消息, 造成消息遺漏;
- 消費者在消費時沒有消費到事務涉及的所有分區, 因此不能讀取事務中的所有消息;
總的來說, 事務保證了生產者可以以事務的方式實現消息發送的exact
-once語義, 但消息清理和消費并未引入事務約束。
實現原理
- 開啟冪等;
- 設置事務ID, transactional.id;
- 生產者通過事務ID得到PID和producer epoch, 進而實現跨生產者會話的消息發送和事務恢復。前者保證相同transactionId的生產者僅有1個可以有效發送消息, 后者保證如果事務消息發送后宕機新恢復出來的生產者可以繼續提交或者終止事務。其中包含2個方面, 生產者的唯一性, 其關聯的在途事務的可見性和可操作性。
- broker端為支持事務消息引入了事務協調器, 與組協調器類似, 用于處理事務的提交和終止。
- 具體交互流程如下
事務存儲
- 日志存儲按Topic, Partition和LogSegment層級存儲, 事務消息也不例外;
- 與普通消息的區別是, 事務消息更多適用于發送一組消息的場景, 具體到LogSegment就是有一組連續的消息, 因此Kafka引入了ControlBatch消息來標志消息結束。
- 事務消息的開始在哪里呢? 嚴格來說, producer跨分區發送成功后, consumer是無法恢復出原有的順序, 在分區級別僅可以做到與某個事務關聯的一組消息(通過消息的屬性標志是否為事務消息), 結束通過ControlBatch標志一組消息結束。
小結
本文討論了Kafka發送消息的三種語義at-least-once, at-most-once, exact-once，并針對exact-once的單分區實現(冪等控制)和跨分區實現(事務消息)做簡要介紹, 希望能幫助你梳理出Kafka broker端對消息發送QoS實現的基本脈絡, 為進一步學習打基礎。
