一、大數(shù)據(jù)框架三種語義‍‍‍‍

在分布式系統(tǒng)中，如kafka、spark、flink等構(gòu)成系統(tǒng)的任何節(jié)點(diǎn)都是被定義為可以彼此獨(dú)立失敗的。比如在 Kafka 中，broker 可能會(huì) crash，在 producer 推送數(shù)據(jù)至 topic 的過程中也可能會(huì)遇到網(wǎng)絡(luò)問題。根據(jù) producer 處理此類故障所采取的提交策略類型，有如下三種(以kafka為例)：

at-least-once：如果 producer 收到來自 Kafka broker 的確認(rèn)（ack）或者 acks = all，則表示該消息已經(jīng)寫入到 Kafka。但如果 producer ack 超時(shí)或收到錯(cuò)誤，則可能會(huì)重試發(fā)送消息，客戶端會(huì)認(rèn)為該消息未寫入 Kafka。如果 broker 在發(fā)送 Ack 之前失敗，但在消息成功寫入 Kafka 之后，此重試將導(dǎo)致該消息被寫入兩次，因此消息會(huì)被不止一次地傳遞給最終 consumer，這種策略可能導(dǎo)致重復(fù)的工作和不正確的結(jié)果。

at-most-once：如果在 ack 超時(shí)或返回錯(cuò)誤時(shí) producer 不重試，則該消息可能最終不會(huì)寫入 Kafka，因此不會(huì)傳遞給 consumer。在大多數(shù)情況下，這樣做是為了避免重復(fù)的可能性，業(yè)務(wù)上必須接收數(shù)據(jù)傳遞可能的丟失。

exactly-once：即使 producer 重試發(fā)送消息，消息也會(huì)保證最多一次地傳遞給最終consumer。該語義是最理想的，但也難以實(shí)現(xiàn)，因?yàn)樗枰⑾到y(tǒng)本身與生產(chǎn)和消費(fèi)消息的應(yīng)用程序進(jìn)行協(xié)作。

二、大數(shù)據(jù)框架故障階段(kafka為例)

理想狀況，網(wǎng)絡(luò)良好，代碼沒有錯(cuò)誤，則 Kafka 可以保證 exactly-once，但生產(chǎn)環(huán)境錯(cuò)綜復(fù)雜，故障幾乎無法避免，主要有：

1 框架自身故障(Broker)：Kafka 作為一個(gè)高可用、持久化系統(tǒng)，保證每條消息被持久化并且冗余多份（假設(shè)是 n 份），所以 Kafka 可以容忍 n-1 個(gè) broker 故障，意味著一個(gè)分區(qū)只要至少有一個(gè) broker 可用，分區(qū)就可用。Kafka 的副本協(xié)議保證了只要消息被成功寫入了主副本，它就會(huì)被復(fù)制到其他所有的可用副本（ISR）。

2 客戶端發(fā)送框架失?。≒roducer 到 Broker 的 RPC）：Kafka 的持久性依賴于生產(chǎn)者接收broker 的 ack 。沒有接收成功 ack 不代表生產(chǎn)請(qǐng)求本身失敗了。broker 可能在寫入消息后，發(fā)送 ack 給生產(chǎn)者的時(shí)候掛了，甚至 broker 也可能在寫入消息前就掛了。由于生產(chǎn)者沒有辦法知道錯(cuò)誤是什么造成的，所以它就只能認(rèn)為消息沒寫入成功，并且會(huì)重試發(fā)送。在一些情況下，這會(huì)造成同樣的消息在 Kafka 分區(qū)日志中重復(fù)，進(jìn)而造成消費(fèi)端多次收到這條消息。

3 客戶端也失敗(Producer)：Exactly-once delivery 也必須考慮客戶端失敗的情況。但是如何去區(qū)分客戶端是真的掛了（永久性宕機(jī)）還是說只是暫時(shí)丟失心跳？追求正確性的話，broker 應(yīng)該丟棄由 zombie producer 發(fā)送的消息。 consumer 也是如此，一旦新的客戶端實(shí)例已經(jīng)啟動(dòng)，它必須能夠從失敗實(shí)例的任何狀態(tài)中恢復(fù)，并從安全點(diǎn)( safe checkpoint )開始處理，這意味著消費(fèi)的偏移量必須始終與生成的輸出保持同步。

4 框架發(fā)送消費(fèi)端失?。˙roker到 Consumer 的 RPC）

三、Exactly-once底層實(shí)現(xiàn)原理

3.1、依賴業(yè)務(wù)控制

• 對(duì)生產(chǎn)者：

每個(gè)分區(qū)只有一個(gè)生產(chǎn)者寫入消息，當(dāng)出現(xiàn)異?；虺瑫r(shí)，生產(chǎn)者查詢此分區(qū)最后一個(gè)消息，用于決定后續(xù)操作時(shí)重傳還是繼續(xù)發(fā)送。

為每個(gè)消息增加唯一主鍵，生產(chǎn)者不做處理，由消費(fèi)者根據(jù)主鍵去重。

• 對(duì)消費(fèi)者：

關(guān)閉自動(dòng)提交 offset 的功能，不使用 Offsets Topic 這個(gè)內(nèi)部 Topic 記錄其 offset，而是由消費(fèi)者自動(dòng)保存 offset。將 offset 和消息處理放在一個(gè)事務(wù)里面，事務(wù)執(zhí)行成功認(rèn)為消息被消費(fèi)，否則事務(wù)回滾需要重新處理。當(dāng)出現(xiàn)消費(fèi)者重啟或者 Rebalance 操作，可以從數(shù)據(jù)庫(kù)找到對(duì)應(yīng)的 offset，然后調(diào)用 KafkaConsumer.seek() 設(shè)置消費(fèi)者位置，從此 offset 開始消費(fèi)。

3.2、依賴 Kafka

3.2.1、冪等性：每個(gè)分區(qū)中精確一次且有序（Idempotence: Exactly-once in order semantics per partition）

Kafka 在0.11.0.0之前的版本中只支持 At Least Once 和 At Most Once 語義，尚不支持 Exactly Once 語義。

Kafka 0.11.0.0版本引入了冪等語義。 一個(gè)冪等性的操作就是一種被執(zhí)行多次造成的影響和只執(zhí)行一次造成的影響一樣的操作。

如果出現(xiàn)導(dǎo)致生產(chǎn)者重試的錯(cuò)誤，同樣的消息，仍由同樣的生產(chǎn)者發(fā)送多次，將只被寫到 Kafka broker 的日志中一次。

對(duì)于單個(gè)分區(qū)，冪等生產(chǎn)者不會(huì)因?yàn)樯a(chǎn)者或 broker 故障而產(chǎn)生多條重復(fù)消息。

想要開啟這個(gè)特性，獲得每個(gè)分區(qū)內(nèi)的精確一次語義，也就是說沒有重復(fù)，沒有丟失，并且有序的語義，只需要 producer 配置 enable.idempotence=true。

這個(gè)特性是怎么實(shí)現(xiàn)的呢？每個(gè)新的 Producer 在初始化的時(shí)候會(huì)被分配一個(gè)唯一的 PID，該P(yáng)ID對(duì)用戶完全透明而不會(huì)暴露給用戶。在底層，它和 TCP 的工作原理有點(diǎn)像，每一批發(fā)送到 Kafka 的消息都將包含 PID 和一個(gè)從 0 開始單調(diào)遞增序列號(hào)。

Broker 將使用這個(gè)序列號(hào)來刪除重復(fù)的發(fā)送。和只能在瞬態(tài)內(nèi)存中的連接中保證不重復(fù)的 TCP 不同，這個(gè)序列號(hào)被持久化到副本日志，所以，即使分區(qū)的 leader 掛了，其他的 broker 接管了leader，新 leader 仍可以判斷重新發(fā)送的是否重復(fù)了。這種機(jī)制的開銷非常低：每批消息只有幾個(gè)額外的字段。這種特性比非冪等的生產(chǎn)者只增加了可忽略的性能開銷。

如果消息序號(hào)比 Broker 維護(hù)的序號(hào)大 1 以上，說明中間有數(shù)據(jù)尚未寫入，也即亂序，此時(shí) Broker 拒絕該消息。

如果消息序號(hào)小于等于 Broker 維護(hù)的序號(hào)，說明該消息已被保存，即為重復(fù)消息，Broker直接丟棄該消息。

總結(jié)來說，producer 端發(fā)送消息時(shí)，生成全局唯一自增pid，和broker中數(shù)據(jù)的pid進(jìn)行對(duì)比，多則刪除，少則通知producer端重新發(fā)送。

3.2.2、事務(wù)：跨分區(qū)原子寫入

上述冪等設(shè)計(jì)只能保證單個(gè) Producer 對(duì)于同一個(gè) <Topic, Partition> 的 Exactly Once 語義。

Kafka 現(xiàn)在通過新的事務(wù) API 支持跨分區(qū)原子寫入。這將允許一個(gè)生產(chǎn)者發(fā)送一批到不同分區(qū)的消息，這些消息要么全部對(duì)任何一個(gè)消費(fèi)者可見，要么對(duì)任何一個(gè)消費(fèi)者都不可見。這個(gè)特性也允許在一個(gè)事務(wù)中處理消費(fèi)數(shù)據(jù)和提交消費(fèi)偏移量，從而實(shí)現(xiàn)端到端的精確一次語義。

為了實(shí)現(xiàn)這種效果，應(yīng)用程序必須提供一個(gè)穩(wěn)定的（重啟后不變）唯一的 ID，也即Transaction ID 。 Transactin ID 與 PID 可能一一對(duì)應(yīng)。區(qū)別在于 Transaction ID 由用戶提供，將生產(chǎn)者的 transactional.id 配置項(xiàng)設(shè)置為某個(gè)唯一ID。而 PID 是內(nèi)部的實(shí)現(xiàn)對(duì)用戶透明。

另外，為了保證新的 Producer 啟動(dòng)后，舊的具有相同 Transaction ID 的 Producer 失效，每次 Producer 通過 Transaction ID 拿到 PID 的同時(shí)，還會(huì)獲取一個(gè)單調(diào)遞增的 epoch。由于舊的 Producer 的 epoch 比新 Producer 的 epoch 小，Kafka 可以很容易識(shí)別出該 Producer 是老的 Producer 并拒絕其請(qǐng)求。

有了Transaction ID后，Kafka可保證：

• 跨Session的數(shù)據(jù)冪等發(fā)送。當(dāng)具有相同Transaction ID的新的Producer實(shí)例被創(chuàng)建且工作時(shí)，舊的且擁有相同Transaction ID的Producer將不再工作。
• 跨Session的事務(wù)恢復(fù)。如果某個(gè)應(yīng)用實(shí)例宕機(jī)，新的實(shí)例可以保證任何未完成的舊的事務(wù)要么Commit要么Abort，使得新實(shí)例從一個(gè)正常狀態(tài)開始工作。

需要注意的是，上述的事務(wù)保證是從Producer的角度去考慮的。從Consumer的角度來看，該保證會(huì)相對(duì)弱一些。尤其是不能保證所有被某事務(wù)Commit過的所有消息都被一起消費(fèi)，因?yàn)椋?/p>

• 對(duì)于壓縮的Topic而言，同一事務(wù)的某些消息可能被其它版本覆蓋
• 事務(wù)包含的消息可能分布在多個(gè)Segment中（即使在同一個(gè)Partition內(nèi)），當(dāng)老的Segment被刪除時(shí)，該事務(wù)的部分?jǐn)?shù)據(jù)可能會(huì)丟失
• Consumer在一個(gè)事務(wù)內(nèi)可能通過seek方法訪問任意Offset的消息，從而可能丟失部分消息
• Consumer可能并不需要消費(fèi)某一事務(wù)內(nèi)的所有Partition，因此它將永遠(yuǎn)不會(huì)讀取組成該事務(wù)的所有消息

四、事務(wù)中Offset的提交

許多基于Kafka的應(yīng)用，尤其是Kafka Stream應(yīng)用中同時(shí)包含Consumer和Producer，前者負(fù)責(zé)從Kafka中獲取消息，后者負(fù)責(zé)將處理完的數(shù)據(jù)寫回Kafka的其它Topic中。

為了實(shí)現(xiàn)該場(chǎng)景下的事務(wù)的原子性，Kafka需要保證對(duì)Consumer Offset的Commit與Producer對(duì)發(fā)送消息的Commit包含在同一個(gè)事務(wù)中。否則，如果在二者Commit中間發(fā)生異常，根據(jù)二者Commit的順序可能會(huì)造成數(shù)據(jù)丟失和數(shù)據(jù)重復(fù)：

• 如果先Commit Producer發(fā)送數(shù)據(jù)的事務(wù)再Commit Consumer的Offset，即At Least Once語義，可能造成數(shù)據(jù)重復(fù)。
• 如果先Commit Consumer的Offset，再Commit Producer數(shù)據(jù)發(fā)送事務(wù)，即At Most Once語義，可能造成數(shù)據(jù)丟失。

五、分布式事務(wù)經(jīng)見實(shí)現(xiàn)機(jī)制

5.1 兩階段提交

Kafka的事務(wù)機(jī)制與《分布式事務(wù)（一）兩階段提交及JTA》一文中所介紹的兩階段提交機(jī)制看似相似，都分PREPARE階段和最終COMMIT階段，但又有很大不同。

• Kafka事務(wù)機(jī)制中，PREPARE時(shí)即要指明是PREPARE_COMMIT還是PREPARE_ABORT并且只須在Transaction Log中標(biāo)記即可，無須其它組件參與。而兩階段提交的PREPARE需要發(fā)送給所有的分布式事務(wù)參與方，并且事務(wù)參與方需要盡可能準(zhǔn)備好，并根據(jù)準(zhǔn)備情況返回Prepared或Non-Prepared狀態(tài)給事務(wù)管理器。
• Kafka事務(wù)中，一但發(fā)起PREPARE_COMMIT或PREPARE_ABORT則確定該事務(wù)最終的結(jié)果應(yīng)該是被COMMIT或ABORT。而分布式事務(wù)中，PREPARE后由各事務(wù)參與方返回狀態(tài)，只有所有參與方均返回Prepared狀態(tài)才會(huì)真正執(zhí)行COMMIT，否則執(zhí)行ROLLBACK
• Kafka事務(wù)機(jī)制中，某幾個(gè)Partition在COMMIT或ABORT過程中變?yōu)椴豢捎茫挥绊懺揚(yáng)artition不影響其它Partition。兩階段提交中，若唯一收到COMMIT命令參與者Crash，其它事務(wù)參與方無法判斷事務(wù)狀態(tài)從而使得整個(gè)事務(wù)阻塞
• Kafka事務(wù)機(jī)制引入事務(wù)超時(shí)機(jī)制，有效避免了掛起的事務(wù)影響其它事務(wù)的問題
• Kafka事務(wù)機(jī)制中存在多個(gè)Transaction Coordinator實(shí)例，而分布式事務(wù)中只有一個(gè)事務(wù)管理器

兩階段提交原理

二階段提交的算法思路可以概括為：協(xié)調(diào)者詢問參與者是否準(zhǔn)備好了提交，并根據(jù)所有參與者的反饋情況決定向所有參與者發(fā)送commit或者rollback指令（協(xié)調(diào)者向所有參與者發(fā)送相同的指令）。

所謂的兩個(gè)階段是指

• 準(zhǔn)備階段 又稱投票階段。在這一階段，協(xié)調(diào)者詢問所有參與者是否準(zhǔn)備好提交，參與者如果已經(jīng)準(zhǔn)備好提交則回復(fù)Prepared，否則回復(fù)Non-Prepared。
• 提交階段又稱執(zhí)行階段。協(xié)調(diào)者如果在上一階段收到所有參與者回復(fù)的Prepared，則在此階段向所有參與者發(fā)送commit指令，所有參與者立即執(zhí)行commit操作；否則協(xié)調(diào)者向所有參與者發(fā)送rollback指令，參與者立即執(zhí)行rollback操作。

5.2 Zookeeper

Zookeeper的原子廣播協(xié)議與兩階段提交以及Kafka事務(wù)機(jī)制有相似之處，但又有各自的特點(diǎn)

• Kafka事務(wù)可COMMIT也可ABORT。而Zookeeper原子廣播協(xié)議只有COMMIT沒有ABORT。當(dāng)然，Zookeeper不COMMIT某消息也即等效于ABORT該消息的更新。
• Kafka存在多個(gè)Transaction Coordinator實(shí)例，擴(kuò)展性較好。而Zookeeper寫操作只能在Leader節(jié)點(diǎn)進(jìn)行，所以其寫性能遠(yuǎn)低于讀性能。
• Kafka事務(wù)是COMMIT還是ABORT完全取決于Producer即客戶端。而Zookeeper原子廣播協(xié)議中某條消息是否被COMMIT取決于是否有一大半FOLLOWER ACK該消息。