詳細講解一下分布式事務(wù)-兩階段提交的實現(xiàn)原理
作者：全棧程序員老馬
Seata（Simple Extensible Autonomous Transaction Architecture）是一款開源的分布式事務(wù)解決方案，提供了AT、TCC、SAGA和XA四種事務(wù)模式。其中，AT（Auto Transaction）模式是基于兩階段提交協(xié)議改進而來，通過數(shù)據(jù)源代理和全局鎖機制實現(xiàn)了對業(yè)務(wù)代碼幾乎零侵入的分布式事務(wù)支持。
一、為什么需要分布式事務(wù)
隨著微服務(wù)架構(gòu)和分布式系統(tǒng)的普及，一個業(yè)務(wù)操作往往需要調(diào)用多個服務(wù)，修改多個數(shù)據(jù)源的數(shù)據(jù)。例如：
- 電商系統(tǒng)中的下單操作：需要扣減庫存、創(chuàng)建訂單、支付等多個操作
- 銀行轉(zhuǎn)賬操作：需要從一個賬戶扣款，另一個賬戶加款
這些操作需要作為一個整體要么全部成功，要么全部失敗，這就需要分布式事務(wù)來保證。
二、兩階段提交（2PC) 原理
兩階段提交（Two-Phase Commit，簡稱2PC）是分布式系統(tǒng)中實現(xiàn)分布式事務(wù)的經(jīng)典算法。它將事務(wù)的提交過程分為兩個階段：
- 準備階段（Prepare Phase）：協(xié)調(diào)者詢問所有參與者是否可以提交
- 提交/回滾階段（Commit/Rollback Phase）：根據(jù)準備階段的結(jié)果決定提交或回滾
角色劃分：
- 協(xié)調(diào)者（Coordinator）：事務(wù)的發(fā)起者，負責協(xié)調(diào)所有參與者
- 參與者（Participant/Cohort）：事務(wù)的實際執(zhí)行者，負責本地事務(wù)的執(zhí)行
兩階段提交流程：
（1）準備階段
- 協(xié)調(diào)者向所有參與者發(fā)送prepare請求
- 參與者執(zhí)行本地事務(wù)但不提交，記錄undo/redo日志
- 參與者向協(xié)調(diào)者反饋響應(yīng)：
成功：返回"同意"
失敗：返回"中止"
（2）提交/回滾階段
情況1：所有參與者都返回"同意"
- 協(xié)調(diào)者向所有參與者發(fā)送commit請求
- 參與者完成本地事務(wù)提交
- 參與者向協(xié)調(diào)者發(fā)送ack響應(yīng)
- 協(xié)調(diào)者收到所有ack后完成事務(wù)
情況2：任一參與者返回"中止"或超時
- 協(xié)調(diào)者向所有參與者發(fā)送rollback請求
- 參與者利用undo日志回滾本地事務(wù)
- 參與者向協(xié)調(diào)者發(fā)送ack響應(yīng)
- 協(xié)調(diào)者收到所有ack后中斷事務(wù)
三、基本實現(xiàn)示例
首先定義協(xié)調(diào)者和參與者的接口：
public interface Coordinator {
void startTransaction(List<Participant> participants);
boolean prepare();
void commit();
void rollback();
}
public interface Participant {
boolean prepare();
void commit();
void rollback();
}參與都實現(xiàn)代碼：
public class DatabaseParticipant implements Participant {
private Connection connection;
private String transactionId;
public DatabaseParticipant(Connection connection, String transactionId) {
this.connection = connection;
this.transactionId = transactionId;
}
@Override
public boolean prepare() {
try {
// 設(shè)置不自動提交
connection.setAutoCommit(false);
// 執(zhí)行SQL但不提交
// 這里應(yīng)該有實際的業(yè)務(wù)SQL，如：
// PreparedStatement ps = connection.prepareStatement("UPDATE account SET balance = balance - 100 WHERE user_id = 1");
// ps.executeUpdate();
// 記錄redo/undo日志
logRedoUndo();
return true;
} catch (SQLException e) {
return false;
}
}
@Override
public void commit() {
try {
connection.commit();
cleanRedoUndo();
} catch (SQLException e) {
// 處理異常
}
}
@Override
public void rollback() {
try {
connection.rollback();
cleanRedoUndo();
} catch (SQLException e) {
// 處理異常
}
}
private void logRedoUndo() {
// 實現(xiàn)記錄redo/undo日志的邏輯
}
private void cleanRedoUndo() {
// 清理日志
}
}協(xié)調(diào)者實現(xiàn)代碼：
public class TwoPhaseCommitCoordinator implements Coordinator {
private List<Participant> participants;
private String transactionId;
public TwoPhaseCommitCoordinator(String transactionId) {
this.transactionId = transactionId;
}
@Override
public void startTransaction(List<Participant> participants) {
this.participants = participants;
}
@Override
public boolean prepare() {
for (Participant participant : participants) {
if (!participant.prepare()) {
return false;
}
}
return true;
}
@Override
public void commit() {
if (prepare()) {
for (Participant participant : participants) {
participant.commit();
}
} else {
rollback();
}
}
@Override
public void rollback() {
for (Participant participant : participants) {
try {
participant.rollback();
} catch (Exception e) {
// 記錄日志，繼續(xù)回滾其他參與者
}
}
}
}使用示例：
public class TwoPhaseCommitExample {
public static void main(String[] args) {
// 模擬兩個數(shù)據(jù)庫參與者
Connection conn1 = getConnection(); // 獲取第一個數(shù)據(jù)庫連接
Connection conn2 = getConnection(); // 獲取第二個數(shù)據(jù)庫連接
Participant participant1 = new DatabaseParticipant(conn1, "tx123");
Participant participant2 = new DatabaseParticipant(conn2, "tx123");
Coordinator coordinator = new TwoPhaseCommitCoordinator("tx123");
coordinator.startTransaction(Arrays.asList(participant1, participant2));
try {
coordinator.commit();
System.out.println("事務(wù)提交成功");
} catch (Exception e) {
coordinator.rollback();
System.out.println("事務(wù)回滾");
}
}
private static Connection getConnection() {
// 實際應(yīng)用中應(yīng)該從數(shù)據(jù)源獲取連接
return null;
}
}四、Seata AT模式
Seata（Simple Extensible Autonomous Transaction Architecture）是一款開源的分布式事務(wù)解決方案，提供了AT、TCC、SAGA和XA四種事務(wù)模式。其中，AT（Auto Transaction）模式是基于兩階段提交協(xié)議改進而來，通過數(shù)據(jù)源代理和全局鎖機制實現(xiàn)了對業(yè)務(wù)代碼幾乎零侵入的分布式事務(wù)支持。
包含組件：
Transaction Coordinator (TC，事務(wù)協(xié)調(diào)器)
- 獨立部署的服務(wù)，維護全局事務(wù)和分支事務(wù)的狀態(tài)
- 負責協(xié)調(diào)全局事務(wù)的提交或回滾
- 管理全局鎖的獲取與釋放
Transaction Manager (TM，事務(wù)管理器)
- 嵌入在應(yīng)用中，負責定義全局事務(wù)邊界
- 通過@GlobalTransactional注解標記分布式事務(wù)方法
- 向TC發(fā)起全局事務(wù)的開始、提交或回滾指令
Resource Manager (RM，資源管理器)
- 管理分支事務(wù)上的資源
- 向TC注冊分支事務(wù)并報告狀態(tài)
- 驅(qū)動分支事務(wù)的提交或回滾
- 負責生成和操作undo log
AT模式的整體流程：
（1）業(yè)務(wù)執(zhí)行與本地提交
- 解析SQL：攔截業(yè)務(wù)SQL，解析SQL類型（INSERT/UPDATE/DELETE）、表、條件等信息
- 查詢前鏡像：根據(jù)SQL條件查詢修改前的數(shù)據(jù)快照（before image）
- 執(zhí)行業(yè)務(wù)SQL：執(zhí)行用戶的實際業(yè)務(wù)SQL
- 查詢后鏡像：根據(jù)主鍵查詢修改后的數(shù)據(jù)快照（after image）
- 插入回滾日志：將前后鏡像和業(yè)務(wù)SQL信息組成undo log記錄，插入到undo_log表
- 注冊分支事務(wù)：向TC注冊分支事務(wù)并獲取全局鎖
- 提交本地事務(wù)：業(yè)務(wù)SQL和undo log在同一個本地事務(wù)中提交
- 上報執(zhí)行結(jié)果：將本地事務(wù)執(zhí)行結(jié)果上報給TC
（2）全局提交或回滾
全局提交：
- TC異步通知各分支事務(wù)提交
- RM異步刪除對應(yīng)的undo log記錄
- 釋放全局鎖
全局回滾：
- TC通知各分支事務(wù)回滾
- RM根據(jù)undo log中的before image生成補償SQL并執(zhí)行
- 校驗數(shù)據(jù)一致性（對比after image與當前數(shù)據(jù)）
- 刪除undo log記錄
- 釋放全局鎖
Seata的詳細信息，請查看官網(wǎng)：https://seata.apache.org/zh-cn/docs/dev/mode/at-mode
五、兩階段提交的問題
（1）協(xié)調(diào)者單點故障
如果在第二階段協(xié)調(diào)者宕機，部分參與者收到commit而部分沒收到，系統(tǒng)將處于不一致狀態(tài)。
解決方法：記錄事務(wù)日志，協(xié)調(diào)者恢復(fù)后能繼續(xù)處理。
（2）網(wǎng)絡(luò)分區(qū)
網(wǎng)絡(luò)分區(qū)可能導(dǎo)致部分參與者無法收到協(xié)調(diào)者的指令。
為了解決2PC的網(wǎng)絡(luò)阻塞問題，引入了3PC：
- CanCommit階段：詢問參與者是否可以提交
- PreCommit階段：預(yù)提交，執(zhí)行事務(wù)但不提交
- DoCommit階段：實際提交
3PC通過引入超時機制減少了阻塞，但增加了復(fù)雜度。
責任編輯：武曉燕
來源：
全棧程序員老馬
