兩階段提交協(xié)議（2PC）

兩階段提交協(xié)議（Two－Phase Commit，2PC）是很常用的解決分布式事務(wù)問(wèn)題的方式，它可以保證在分布式事務(wù)中，要么所有參與進(jìn)程都提交事務(wù)，要么都取消，即實(shí)現(xiàn) ACID 中的 A （原子性）。

在數(shù)據(jù)一致性的環(huán)境下，其代表的含義是：要么所有備份數(shù)據(jù)同時(shí)更改某個(gè)數(shù)值，要么都不改，以此來(lái)達(dá)到數(shù)據(jù)的強(qiáng)一致性。

兩階段提交協(xié)議中有兩個(gè)重要角色，協(xié)調(diào)者（Coordinator）和參與者（Participant），其中協(xié)調(diào)者只有一個(gè)，起到分布式事務(wù)的協(xié)調(diào)管理作用，參與者有多個(gè)。

顧名思義，兩階段提交將提交過(guò)程劃分為連續(xù)的兩個(gè)階段：表決階段（Voting）和提交階段（Commit）。

兩階段提交協(xié)議過(guò)程如下圖所示：

兩階段提交協(xié)議

第一階段：表決階段

協(xié)調(diào)者向所有參與者發(fā)送一個(gè) VOTE＿REQUEST 消息。

當(dāng)參與者接收到 VOTE＿REQUEST 消息，向協(xié)調(diào)者發(fā)送 VOTE＿COMMIT 消息作為回應(yīng)，告訴協(xié)調(diào)者自己已經(jīng)做好準(zhǔn)備提交準(zhǔn)備，如果參與者沒有準(zhǔn)備好或遇到其他故障，就返回一個(gè) VOTE＿ABORT 消息，告訴協(xié)調(diào)者目前無(wú)法提交事務(wù)。

第二階段：提交階段

協(xié)調(diào)者收集來(lái)自各個(gè)參與者的表決消息。如果所有參與者一致認(rèn)為可以提交事務(wù)，那么協(xié)調(diào)者決定事務(wù)的最終提交，在此情形下協(xié)調(diào)者向所有參與者發(fā)送一個(gè) GLOBAL＿COMMIT 消息，通知參與者進(jìn)行本地提交；如果所有參與者中有任意一個(gè)返回消息是 VOTE＿ABORT，協(xié)調(diào)者就會(huì)取消事務(wù)，向所有參與者廣播一條 GLOBAL＿ABORT 消息通知所有的參與者取消事務(wù)。

每個(gè)提交了表決信息的參與者等候協(xié)調(diào)者返回消息，如果參與者接收到一個(gè) GLOBAL＿COMMIT 消息，那么參與者提交本地事務(wù)，否則如果接收到 GLOBAL＿ABORT 消息，則參與者取消本地事務(wù)。

兩階段提交協(xié)議在 Flink 中的應(yīng)用

Flink 的兩階段提交思路：

我們從 Flink 程序啟動(dòng)到消費(fèi) Kafka 數(shù)據(jù)，最后到 Flink 將數(shù)據(jù) Sink 到 Kafka 為止，來(lái)分析 Flink 的精準(zhǔn)一次處理。

當(dāng) Checkpoint 啟動(dòng)時(shí)，JobManager 會(huì)將檢查點(diǎn)分界線（checkpoint battier）注入數(shù)據(jù)流，checkpoint barrier 會(huì)在算子間傳遞下去，如下如所示：

Flink 精準(zhǔn)一次處理：Checkpoint 啟動(dòng)

Source 端：Flink Kafka Source 負(fù)責(zé)保存 Kafka 消費(fèi) offset，當(dāng) Chckpoint 成功時(shí) Flink 負(fù)責(zé)提交這些寫入，否則就終止取消掉它們，當(dāng) Chckpoint 完成位移保存，它會(huì)將 checkpoint barrier（檢查點(diǎn)分界線） 傳給下一個(gè) Operator，然后每個(gè)算子會(huì)對(duì)當(dāng)前的狀態(tài)做個(gè)快照，保存到狀態(tài)后端（State Backend）。

對(duì)于 Source 任務(wù)而言，就會(huì)把當(dāng)前的 offset 作為狀態(tài)保存起來(lái)。下次從 Checkpoint 恢復(fù)時(shí)，Source 任務(wù)可以重新提交偏移量，從上次保存的位置開始重新消費(fèi)數(shù)據(jù)，如下圖所示：

Flink 精準(zhǔn)一次處理：checkpoint barrier 及 offset 保存Slink 端：從 Source 端開始，每個(gè)內(nèi)部的 transform 任務(wù)遇到 checkpoint barrier（檢查點(diǎn)分界線）時(shí)，都會(huì)把狀態(tài)存到 Checkpoint 里。數(shù)據(jù)處理完畢到 Sink 端時(shí)，Sink 任務(wù)首先把數(shù)據(jù)寫入外部 Kafka，這些數(shù)據(jù)都屬于預(yù)提交的事務(wù)（還不能被消費(fèi)），此時(shí)的 Pre－commit 預(yù)提交階段下 Data Sink 在保存狀態(tài)到狀態(tài)后端的同時(shí)還必須預(yù)提交它的外部事務(wù)，如下圖所示：

Flink 精準(zhǔn)一次處理：預(yù)提交到外部系統(tǒng)

當(dāng)所有算子任務(wù)的快照完成（所有創(chuàng)建的快照都被視為是 Checkpoint 的一部分），也就是這次的 Checkpoint 完成時(shí)，JobManager 會(huì)向所有任務(wù)發(fā)通知，確認(rèn)這次 Checkpoint 完成，此時(shí) Pre－commit 預(yù)提交階段才算完成。才正式到兩階段提交協(xié)議的第二個(gè)階段：commit 階段。該階段中 JobManager 會(huì)為應(yīng)用中每個(gè) Operator 發(fā)起 Checkpoint 已完成的回調(diào)邏輯。

本例中的 Data Source 和窗口操作無(wú)外部狀態(tài)，因此在該階段，這兩個(gè) Opeartor 無(wú)需執(zhí)行任何邏輯，但是 Data Sink 是有外部狀態(tài)的，此時(shí)我們必須提交外部事務(wù)，當(dāng) Sink 任務(wù)收到確認(rèn)通知，就會(huì)正式提交之前的事務(wù)，Kafka 中未確認(rèn)的數(shù)據(jù)就改為“已確認(rèn)”，數(shù)據(jù)就真正可以被消費(fèi)了，如下圖所示：

Flink 精準(zhǔn)一次處理：數(shù)據(jù)精準(zhǔn)被消費(fèi)

注：Flink 由 JobManager 協(xié)調(diào)各個(gè) TaskManager 進(jìn)行 Checkpoint 存儲(chǔ)，Checkpoint 保存在 StateBackend（狀態(tài)后端） 中，默認(rèn) StateBackend 是內(nèi)存級(jí)的，也可以改為文件級(jí)的進(jìn)行持久化保存。

最后，一張圖總結(jié)下 Flink 的 EOS：

Flink 端到端精準(zhǔn)一次處理

此圖建議保存，總結(jié)全面且簡(jiǎn)明扼要，再也不慫面試官！