本文全部Kafka原理性的描述除特殊說明外均基於Kafka 1.0.0版本。html
Kafka事務機制的實現主要是爲了支持sql
Exactly Once
即正好一次語義Exactly Once
《Kafka背景及架構介紹》一文中有說明Kafka在0.11.0.0以前的版本中只支持At Least Once
和At Most Once
語義,尚不支持Exactly Once
語義。緩存
可是在不少要求嚴格的場景下,如使用Kafka處理交易數據,Exactly Once
語義是必須的。咱們能夠經過讓下游系統具備冪等性來配合Kafka的At Least Once
語義來間接實現Exactly Once
。可是:服務器
所以,Kafka自己對Exactly Once
語義的支持就很是必要。架構
操做的原子性是指,多個操做要麼所有成功要麼所有失敗,不存在部分紅功部分失敗的可能。併發
實現原子性操做的意義在於:mvc
上文提到,實現Exactly Once
的一種方法是讓下游系統具備冪等處理特性,而在Kafka Stream中,Kafka Producer自己就是「下游」系統,所以若是能讓Producer具備冪等處理特性,那就可讓Kafka Stream在必定程度上支持Exactly once
語義。app
爲了實現Producer的冪等語義,Kafka引入了Producer ID
(即PID
)和Sequence Number
。每一個新的Producer在初始化的時候會被分配一個惟一的PID,該PID對用戶徹底透明而不會暴露給用戶。分佈式
對於每一個PID,該Producer發送數據的每一個<Topic, Partition>
都對應一個從0開始單調遞增的Sequence Number
。ide
相似地,Broker端也會爲每一個<PID, Topic, Partition>
維護一個序號,而且每次Commit一條消息時將其對應序號遞增。對於接收的每條消息,若是其序號比Broker維護的序號(即最後一次Commit的消息的序號)大一,則Broker會接受它,不然將其丟棄:
InvalidSequenceNumber
DuplicateSequenceNumber
上述設計解決了0.11.0.0以前版本中的兩個問題:
上述冪等設計只能保證單個Producer對於同一個<Topic, Partition>
的Exactly Once
語義。
另外,它並不能保證寫操做的原子性——即多個寫操做,要麼所有被Commit要麼所有不被Commit。
更不能保證多個讀寫操做的的原子性。尤爲對於Kafka Stream應用而言,典型的操做便是從某個Topic消費數據,通過一系列轉換後寫回另外一個Topic,保證從源Topic的讀取與向目標Topic的寫入的原子性有助於從故障中恢復。
事務保證可以使得應用程序將生產數據和消費數據看成一個原子單元來處理,要麼所有成功,要麼所有失敗,即便該生產或消費跨多個<Topic, Partition>
。
另外,有狀態的應用也能夠保證重啓後從斷點處繼續處理,也即事務恢復。
爲了實現這種效果,應用程序必須提供一個穩定的(重啓後不變)惟一的ID,也即Transaction ID
。Transactin ID
與PID
可能一一對應。區別在於Transaction ID
由用戶提供,而PID
是內部的實現對用戶透明。
另外,爲了保證新的Producer啓動後,舊的具備相同Transaction ID
的Producer即失效,每次Producer經過Transaction ID
拿到PID的同時,還會獲取一個單調遞增的epoch。因爲舊的Producer的epoch比新Producer的epoch小,Kafka能夠很容易識別出該Producer是老的Producer並拒絕其請求。
有了Transaction ID
後,Kafka可保證:
Transaction ID
的新的Producer實例被建立且工做時,舊的且擁有相同Transaction ID
的Producer將再也不工做。須要注意的是,上述的事務保證是從Producer的角度去考慮的。從Consumer的角度來看,該保證會相對弱一些。尤爲是不能保證全部被某事務Commit過的全部消息都被一塊兒消費,由於:
這一節所說的事務主要指原子性,也即Producer將多條消息做爲一個事務批量發送,要麼所有成功要麼所有失敗。
爲了實現這一點,Kafka 0.11.0.0引入了一個服務器端的模塊,名爲Transaction Coordinator
,用於管理Producer發送的消息的事務性。
該Transaction Coordinator
維護Transaction Log
,該log存於一個內部的Topic內。因爲Topic數據具備持久性,所以事務的狀態也具備持久性。
Producer並不直接讀寫Transaction Log
,它與Transaction Coordinator
通訊,而後由Transaction Coordinator
將該事務的狀態插入相應的Transaction Log
。
Transaction Log
的設計與Offset Log
用於保存Consumer的Offset相似。
許多基於Kafka的應用,尤爲是Kafka Stream應用中同時包含Consumer和Producer,前者負責從Kafka中獲取消息,後者負責將處理完的數據寫回Kafka的其它Topic中。
爲了實現該場景下的事務的原子性,Kafka須要保證對Consumer Offset的Commit與Producer對發送消息的Commit包含在同一個事務中。不然,若是在兩者Commit中間發生異常,根據兩者Commit的順序可能會形成數據丟失和數據重複:
At Least Once
語義,可能形成數據重複。At Most Once
語義,可能形成數據丟失。爲了區分寫入Partition的消息被Commit仍是Abort,Kafka引入了一種特殊類型的消息,即Control Message
。該類消息的Value內不包含任何應用相關的數據,而且不會暴露給應用程序。它只用於Broker與Client間的內部通訊。
對於Producer端事務,Kafka以Control Message的形式引入一系列的Transaction Marker
。Consumer便可經過該標記斷定對應的消息被Commit了仍是Abort了,而後結合該Consumer配置的隔離級別決定是否應該將該消息返回給應用程序。
Producer<String, String> producer = new KafkaProducer<String, String>(props); // 初始化事務,包括結束該Transaction ID對應的未完成的事務(若是有) // 保證新的事務在一個正確的狀態下啓動 producer.initTransactions(); // 開始事務 producer.beginTransaction(); // 消費數據 ConsumerRecords<String, String> records = consumer.poll(100); try{ // 發送數據 producer.send(new ProducerRecord<String, String>("Topic", "Key", "Value")); // 發送消費數據的Offset,將上述數據消費與數據發送歸入同一個Transaction內 producer.sendOffsetsToTransaction(offsets, "group1"); // 數據發送及Offset發送均成功的狀況下,提交事務 producer.commitTransaction(); } catch (ProducerFencedException | OutOfOrderSequenceException | AuthorizationException e) { // 數據發送或者Offset發送出現異常時,終止事務 producer.abortTransaction(); } finally { // 關閉Producer和Consumer producer.close(); consumer.close(); }
Transaction Coordinator
因爲Transaction Coordinator
是分配PID和管理事務的核心,所以Producer要作的第一件事情就是經過向任意一個Broker發送FindCoordinator
請求找到Transaction Coordinator
的位置。
注意:只有應用程序爲Producer配置了Transaction ID
時纔可以使用事務特性,也才須要這一步。另外,因爲事務性要求Producer開啓冪等特性,所以經過將transactional.id
設置爲非空從而開啓事務特性的同時也須要經過將enable.idempotence
設置爲true來開啓冪等特性。
找到Transaction Coordinator
後,具備冪等特性的Producer必須發起InitPidRequest
請求以獲取PID。
注意:只要開啓了冪等特性即必須執行該操做,而無須考慮該Producer是否開啓了事務特性。
* 若是事務特性被開啓 *
InitPidRequest
會發送給Transaction Coordinator
。若是Transaction Coordinator
是第一次收到包含有該Transaction ID
的InitPidRequest請求,它將會把該<TransactionID, PID>
存入Transaction Log
,如上圖中步驟2.1所示。這樣可保證該對應關係被持久化,從而保證即便Transaction Coordinator
宕機該對應關係也不會丟失。
除了返回PID外,InitPidRequest
還會執行以下任務:
注意:InitPidRequest
的處理過程是同步阻塞的。一旦該調用正確返回,Producer便可開始新的事務。
另外,若是事務特性未開啓,InitPidRequest
可發送至任意Broker,而且會獲得一個全新的惟一的PID。該Producer將只能使用冪等特性以及單一Session內的事務特性,而不能使用跨Session的事務特性。
Kafka從0.11.0.0版本開始,提供beginTransaction()
方法用於開啓一個事務。調用該方法後,Producer本地會記錄已經開啓了事務,但Transaction Coordinator
只有在Producer發送第一條消息後才認爲事務已經開啓。
這一階段,包含了整個事務的數據處理過程,而且包含了多種請求。
AddPartitionsToTxnRequest
一個Producer可能會給多個<Topic, Partition>
發送數據,給一個新的<Topic, Partition>
發送數據前,它須要先向Transaction Coordinator
發送AddPartitionsToTxnRequest
。
Transaction Coordinator
會將該<Transaction, Topic, Partition>
存於Transaction Log
內,並將其狀態置爲BEGIN
,如上圖中步驟4.1所示。有了該信息後,咱們才能夠在後續步驟中爲每一個Topic, Partition>
設置COMMIT或者ABORT標記(如上圖中步驟5.2所示)。
另外,若是該<Topic, Partition>
爲該事務中第一個<Topic, Partition>
,Transaction Coordinator
還會啓動對該事務的計時(每一個事務都有本身的超時時間)。
ProduceRequest
Producer經過一個或多個ProduceRequest
發送一系列消息。除了應用數據外,該請求還包含了PID,epoch,和Sequence Number
。該過程如上圖中步驟4.2所示。
AddOffsetsToTxnRequest
爲了提供事務性,Producer新增了sendOffsetsToTransaction
方法,該方法將多組消息的發送和消費放入同一批處理內。
該方法先判斷在當前事務中該方法是否已經被調用並傳入了相同的Group ID。如果,直接跳到下一步;若不是,則向Transaction Coordinator
發送AddOffsetsToTxnRequests
請求,Transaction Coordinator
將對應的全部<Topic, Partition>
存於Transaction Log
中,並將其狀態記爲BEGIN
,如上圖中步驟4.3所示。該方法會阻塞直到收到響應。
TxnOffsetCommitRequest
做爲sendOffsetsToTransaction
方法的一部分,在處理完AddOffsetsToTxnRequest
後,Producer也會發送TxnOffsetCommit
請求給Consumer Coordinator
從而將本事務包含的與讀操做相關的各<Topic, Partition>
的Offset持久化到內部的__consumer_offsets
中,如上圖步驟4.4所示。
在此過程當中,Consumer Coordinator
會經過PID和對應的epoch來驗證是否應該容許該Producer的該請求。
這裏須要注意:
__consumer_offsets
的Offset信息在當前事務Commit前對外是不可見的。也即在當前事務被Commit前,可認爲該Offset還沒有Commit,也即對應的消息還沒有被完成處理。Consumer Coordinator
並不會當即更新緩存中相應<Topic, Partition>
的Offset,由於此時這些更新操做還沒有被COMMIT或ABORT。一旦上述數據寫入操做完成,應用程序必須調用KafkaProducer
的commitTransaction
方法或者abortTransaction
方法以結束當前事務。
EndTxnRequest
commitTransaction
方法使得Producer寫入的數據對下游Consumer可見。abortTransaction
方法經過Transaction Marker
將Producer寫入的數據標記爲Aborted
狀態。下游的Consumer若是將isolation.level
設置爲READ_COMMITTED
,則它讀到被Abort的消息後直接將其丟棄而不會返回給客戶程序,也即被Abort的消息對應用程序不可見。
不管是Commit仍是Abort,Producer都會發送EndTxnRequest
請求給Transaction Coordinator
,並經過標誌位標識是應該Commit仍是Abort。
收到該請求後,Transaction Coordinator
會進行以下操做
PREPARE_COMMIT
或PREPARE_ABORT
消息寫入Transaction Log
,如上圖中步驟5.1所示WriteTxnMarker
請求以Transaction Marker
的形式將COMMIT
或ABORT
信息寫入用戶數據日誌以及Offset Log
中,如上圖中步驟5.2所示COMPLETE_COMMIT
或COMPLETE_ABORT
信息寫入Transaction Log
中,如上圖中步驟5.3所示補充說明:對於commitTransaction
方法,它會在發送EndTxnRequest
以前先調用flush方法以確保全部發送出去的數據都獲得相應的ACK。對於abortTransaction
方法,在發送EndTxnRequest
以前直接將當前Buffer中的事務性消息(若是有)所有丟棄,但必須等待全部被髮送但還沒有收到ACK的消息發送完成。
上述第二步是實現將一組讀操做與寫操做做爲一個事務處理的關鍵。由於Producer寫入的數據Topic以及記錄Comsumer Offset的Topic會被寫入相同的Transactin Marker
,因此這一組讀操做與寫操做要麼所有COMMIT要麼所有ABORT。
WriteTxnMarkerRequest
上面提到的WriteTxnMarkerRequest
由Transaction Coordinator
發送給當前事務涉及到的每一個<Topic, Partition>
的Leader。收到該請求後,對應的Leader會將對應的COMMIT(PID)
或者ABORT(PID)
控制信息寫入日誌,如上圖中步驟5.2所示。
該控制消息向Broker以及Consumer代表對應PID的消息被Commit了仍是被Abort了。
這裏要注意,若是事務也涉及到__consumer_offsets
,即該事務中有消費數據的操做且將該消費的Offset存於__consumer_offsets
中,Transaction Coordinator
也須要向該內部Topic的各Partition的Leader發送WriteTxnMarkerRequest
從而寫入COMMIT(PID)
或COMMIT(PID)
控制信息。
寫入最終的COMPLETE_COMMIT
或COMPLETE_ABORT
消息
寫完全部的Transaction Marker
後,Transaction Coordinator
會將最終的COMPLETE_COMMIT
或COMPLETE_ABORT
消息寫入Transaction Log
中以標明該事務結束,如上圖中步驟5.3所示。
此時,Transaction Log
中全部關於該事務的消息所有能夠移除。固然,因爲Kafka內數據是Append Only的,不可直接更新和刪除,這裏說的移除只是將其標記爲null從而在Log Compact時再也不保留。
另外,COMPLETE_COMMIT
或COMPLETE_ABORT
的寫入並不須要獲得全部Rreplica的ACK,由於若是該消息丟失,能夠根據事務協議重發。
補充說明,若是參與該事務的某些<Topic, Partition>
在被寫入Transaction Marker
前不可用,它對READ_COMMITTED
的Consumer不可見,但不影響其它可用<Topic, Partition>
的COMMIT或ABORT。在該<Topic, Partition>
恢復可用後,Transaction Coordinator
會從新根據PREPARE_COMMIT
或PREPARE_ABORT
向該<Topic, Partition>
發送Transaction Marker
。
PID
與Sequence Number
的引入實現了寫操做的冪等性At Least Once
語義實現了單一Session內的Exactly Once
語義Transaction Marker
與PID
提供了識別消息是否應該被讀取的能力,從而實現了事務的隔離性Transaction Marker
)來實現事務中涉及的全部讀寫操做同時對外可見或同時對外不可見InvalidProducerEpoch
這是一種Fatal Error,它說明當前Producer是一個過時的實例,有Transaction ID
相同但epoch更新的Producer實例被建立並使用。此時Producer會中止並拋出Exception。
InvalidPidMapping
Transaction Coordinator
沒有與該Transaction ID
對應的PID。此時Producer會經過包含有Transaction ID
的InitPidRequest
請求建立一個新的PID。
NotCorrdinatorForGTransactionalId
該Transaction Coordinator
不負責該當前事務。Producer會經過FindCoordinatorRequest
請求從新尋找對應的Transaction Coordinator
。
InvalidTxnRequest
違反了事務協議。正確的Client實現不該該出現這種Exception。若是該異常發生了,用戶須要檢查本身的客戶端實現是否有問題。
CoordinatorNotAvailable
Transaction Coordinator
仍在初始化中。Producer只須要重試便可。
DuplicateSequenceNumber
發送的消息的序號低於Broker預期。該異常說明該消息已經被成功處理過,Producer能夠直接忽略該異常並處理下一條消息
InvalidSequenceNumber
這是一個Fatal Error,它說明發送的消息中的序號大於Broker預期。此時有兩種可能
max.inflight.requests.per.connection
被強制設置爲1,而acks
被強制設置爲all。故前面消息重試期間,後續消息不會被髮送,也即不會發生亂序。而且只有ISR中全部Replica都ACK,Producer纔會認爲消息已經被髮送,也即不存在Broker端數據丟失問題。InvalidTransactionTimeout
InitPidRequest
調用出現的Fatal Error。它代表Producer傳入的timeout時間不在可接受範圍內,應該中止Producer並報告給用戶。
Transaction Coordinator
失敗PREPARE_COMMIT/PREPARE_ABORT
前失敗Producer經過FindCoordinatorRequest
找到新的Transaction Coordinator
,並經過EndTxnRequest
請求發起COMMIT
或ABORT
流程,新的Transaction Coordinator
繼續處理EndTxnRequest
請求——寫PREPARE_COMMIT
或PREPARE_ABORT
,寫Transaction Marker
,寫COMPLETE_COMMIT
或COMPLETE_ABORT
。
PREPARE_COMMIT/PREPARE_ABORT
後失敗此時舊的Transaction Coordinator
可能已經成功寫入部分Transaction Marker
。新的Transaction Coordinator
會重複這些操做,因此部分Partition中可能會存在重複的COMMIT
或ABORT
,但只要該Producer在此期間沒有發起新的事務,這些重複的Transaction Marker
就不是問題。
COMPLETE_COMMIT/ABORT
後失敗舊的Transaction Coordinator
可能已經寫完了COMPLETE_COMMIT
或COMPLETE_ABORT
但在返回EndTxnRequest
以前失敗。該場景下,新的Transaction Coordinator
會直接給Producer返回成功。
transaction.timeout.ms
當Producer失敗時,Transaction Coordinator
必須可以主動的讓某些進行中的事務過時。不然沒有Producer的參與,Transaction Coordinator
沒法判斷這些事務應該如何處理,這會形成:
Transaction Coordinator
須要維護大量的事務狀態,大量佔用內存Transaction Log
內也會存在大量數據,形成新的Transaction Coordinator
啓動緩慢READ_COMMITTED
的Consumer須要緩存大量的消息,形成沒必要要的內存浪費甚至是OOMTransaction ID
不一樣的Producer交叉寫同一個Partition,當一個Producer的事務狀態不更新時,READ_COMMITTED
的Consumer爲了保證順序消費而被阻塞爲了不上述問題,Transaction Coordinator
會週期性遍歷內存中的事務狀態Map,並執行以下操做
BEGIN
而且其最後更新時間與當前時間差大於transaction.remove.expired.transaction.cleanup.interval.ms
(默認值爲1小時),則主動將其終止:1)未避免原Producer臨時恢復與當前終止流程衝突,增長該Producer對應的PID的epoch,並確保將該更新的信息寫入Transaction Log
;2)以更新後的epoch回滾事務,從而使得該事務相關的全部Broker都更新其緩存的該PID的epoch從而拒絕舊Producer的寫操做PREPARE_COMMIT
,完成後續的COMMIT流程————向各<Topic, Partition>
寫入Transaction Marker
,在Transaction Log
內寫入COMPLETE_COMMIT
PREPARE_ABORT
,完成後續ABORT流程Transaction ID
某Transaction ID
的Producer可能很長時間再也不發送數據,Transaction Coordinator
不必再保存該Transaction ID
與PID
等的映射,不然可能會形成大量的資源浪費。所以須要有一個機制探測再也不活躍的Transaction ID
並將其信息刪除。
Transaction Coordinator
會週期性遍歷內存中的Transaction ID
與PID
映射,若是某Transaction ID
沒有對應的正在進行中的事務而且它對應的最後一個事務的結束時間與當前時間差大於transactional.id.expiration.ms
(默認值是7天),則將其從內存中刪除並在Transaction Log
中將其對應的日誌的值設置爲null從而使得Log Compact可將其記錄刪除。
Kafka的事務機制與《MVCC PostgreSQL實現事務和多版本併發控制的精華》一文中介紹的PostgreSQL經過MVCC實現事務的機制很是相似,對於事務的回滾,並不須要刪除已寫入的數據,都是將寫入數據的事務標記爲Rollback/Abort從而在讀數據時過濾該數據。
Kafka的事務機制與《分佈式事務(一)兩階段提交及JTA》一文中所介紹的兩階段提交機制看似類似,都分PREPARE階段和最終COMMIT階段,但又有很大不一樣。
PREPARE_COMMIT
仍是PREPARE_ABORT
,而且只須在Transaction Log
中標記便可,無須其它組件參與。而兩階段提交的PREPARE須要發送給全部的分佈式事務參與方,而且事務參與方須要儘量準備好,並根據準備狀況返回Prepared
或Non-Prepared
狀態給事務管理器。PREPARE_COMMIT
或PREPARE_ABORT
,則肯定該事務最終的結果應該是被COMMIT
或ABORT
。而分佈式事務中,PREPARE後由各事務參與方返回狀態,只有全部參與方均返回Prepared
狀態纔會真正執行COMMIT,不然執行ROLLBACKTransaction Coordinator
實例,而分佈式事務中只有一個事務管理器Zookeeper的原子廣播協議與兩階段提交以及Kafka事務機制有類似之處,但又有各自的特色
Transaction Coordinator
實例,擴展性較好。而Zookeeper寫操做只能在Leader節點進行,因此其寫性能遠低於讀性能。
http://www.cnblogs.com/birdstudio/p/7373057.html
http://www.heartthinkdo.com/?p=2040
http://www.infoq.com/cn/articles/kafka-analysis-part-8