分布式事务之2PC两阶段提交

1. 分布式事务概述

1.1 问题背景

在分布式系统中，业务操作可能跨越多个服务或数据库（如订单服务、库存服务、支付服务），传统单机事务（ACID）无法满足跨网络节点的数据一致性需求。

• 网络不可靠：服务间调用可能失败或超时。
  
• 数据一致性：不同节点间的状态需最终一致。
  
• 性能与可用性：避免长时间锁资源导致系统阻塞。
  

分布式事务的核心目标是确保 跨服务/数据库的操作要么全部成功，要么全部回滚。
2. 两阶段提交（2PC）

原理

• 阶段一（Prepare）：协调者询问所有参与者是否可提交，参与者锁定资源并返回“同意”或“拒绝”。
  
• 阶段二（Commit/Rollback）：若所有参与者同意，协调者发送提交命令；否则发送回滚命令。
  

以下是一个简化的 Java 两阶段提交（2PC） 具体实现示例，包含协调者（Coordinator）和参与者（Participant）的核心逻辑。代码通过模拟数据库操作展示2PC的关键流程：
1. 参与者（Participant）实现

每个参与者代表一个独立的数据库或服务，需支持准备（Prepare）、提交（Commit）、回滚（Rollback）操作。

1. import java.util.concurrent.atomic.AtomicBoolean;
3. /**
4. * 参与者（如数据库或服务）
5. */
6. public class Participant {
7.     private String name;          // 参与者名称（如"DB1"）
8.     private AtomicBoolean prepared = new AtomicBoolean(false);  // 准备状态
9.     private AtomicBoolean committed = new AtomicBoolean(false); // 提交状态
11.     public Participant(String name) {
12.         this.name = name;
13.     }
15.     /**
16.      * 阶段一：准备操作（锁定资源）
17.      * @return true表示准备成功，false表示失败
18.      */
19.     public boolean prepare() {
20.         try {
21.             // 模拟资源锁定，实际可能为操作数据库
22.             System.out.println(name + ": Trying to prepare...");
23.             Thread.sleep(100);  // 模拟网络延迟
24.             boolean success = Math.random() > 0.2;  // 80%概率成功
25.             if (success) {
26.                 prepared.set(true);
27.                 System.out.println(name + ": Prepared successfully.");
28.                 return true;
29.             } else {
30.                 System.out.println(name + ": Prepare failed.");
31.                 return false;
32.             }
33.         } catch (InterruptedException e) {
34.             Thread.currentThread().interrupt();
35.             return false;
36.         }
37.     }
39.     /**
40.      * 阶段二：提交操作
41.      */
42.     public void commit() {
43.         if (prepared.get()) {
44.             // 实际提交事务（如更新数据库）
45.             committed.set(true);
46.             System.out.println(name + ": Committed.");
47.         } else {
48.             System.out.println(name + ": Cannot commit without preparation.");
49.         }
50.     }
52.     /**
53.      * 阶段二：回滚操作
54.      */
55.     public void rollback() {
56.         if (prepared.get()) {
57.             // 实际回滚事务（如恢复数据）
58.             prepared.set(false);
59.             System.out.println(name + ": Rolled back.");
60.         } else {
61.             System.out.println(name + ": No need to rollback.");
62.         }
63.     }
65.     // 检查是否已提交
66.     public boolean isCommitted() {
67.         return committed.get();
68.     }
69. }

复制代码

2. 协调者（Coordinator）实现

协调者负责管理所有参与者，驱动两阶段提交流程。

1. import java.util.List;
3. /**
4. * 协调者（事务管理器）
5. */
6. public class Coordinator {
7.     private List<Participant> participants;
9.     public Coordinator(List<Participant> participants) {
10.         this.participants = participants;
11.     }
13.     /**
14.      * 执行两阶段提交事务
15.      * @return true表示事务成功提交，false表示失败
16.      */
17.     public boolean executeTransaction() {
18.         System.out.println("===== Phase 1: Prepare =====");
19.         boolean allPrepared = participants.stream()
20.                 .allMatch(Participant::prepare);
22.         System.out.println("===== Phase 2: Commit/Rollback =====");
23.         if (allPrepared) {
24.             participants.forEach(Participant::commit);
25.             System.out.println("Transaction committed successfully.");
26.             return true;
27.         } else {
28.             participants.forEach(Participant::rollback);
29.             System.out.println("Transaction rolled back due to failures.");
30.             return false;
31.         }
32.     }
33. }

复制代码

3. 客户端测试代码

模拟包含两个参与者的分布式事务场景。

1. import java.util.Arrays;
3. public class TwoPhaseCommitDemo {
4.     public static void main(String[] args) {
5.         // 创建两个参与者（如数据库DB1和DB2）
6.         Participant db1 = new Participant("DB1");
7.         Participant db2 = new Participant("DB2");
9.         // 创建协调者并关联参与者
10.         Coordinator coordinator = new Coordinator(Arrays.asList(db1, db2));
12.         // 执行两阶段提交事务
13.         boolean success = coordinator.executeTransaction();
15.         // 输出最终状态
16.         System.out.println("\nFinal Status:");
17.         System.out.println("DB1 Committed: " + db1.isCommitted());
18.         System.out.println("DB2 Committed: " + db2.isCommitted());
19.         System.out.println("Transaction Result: " + (success ? "SUCCESS" : "FAILURE"));
20.     }
21. }

复制代码

4. 运行结果示例

成功场景（所有参与者准备成功）

1. ===== Phase 1: Prepare =====
2. DB1: Trying to prepare...
3. DB1: Prepared successfully.
4. DB2: Trying to prepare...
5. DB2: Prepared successfully.
6. ===== Phase 2: Commit/Rollback =====
7. DB1: Committed.
8. DB2: Committed.
9. Transaction committed successfully.
11. Final Status:
12. DB1 Committed: true
13. DB2 Committed: true
14. Transaction Result: SUCCESS

复制代码

失败场景（某一参与者准备失败）

1. ===== Phase 1: Prepare =====
2. DB1: Trying to prepare...
3. DB1: Prepared successfully.
4. DB2: Trying to prepare...
5. DB2: Prepare failed.
6. ===== Phase 2: Commit/Rollback =====
7. DB1: Rolled back.
8. DB2: No need to rollback.
9. Transaction rolled back due to failures.
11. Final Status:
12. DB1 Committed: false
13. DB2 Committed: false
14. Transaction Result: FAILURE

复制代码

5. 关键点说明

• 阶段一（Prepare）
  
  
  
  • 协调者询问所有参与者是否可以提交。
    
  • 参与者锁定资源并记录操作日志。
    
  • 任一参与者失败则整个事务回滚。
    
  
  
• 阶段二（Commit/Rollback）
  
  
  
  • 若所有参与者准备成功，协调者发送提交命令。
    
  • 若任一参与者失败，协调者发送回滚命令。
    
  
  
• 代码简化说明
  
  
  
  • 实际应用中需处理网络超时、重试和持久化日志。
    
  • 分布式场景下需使用RPC或HTTP替代本地方法调用。
    
  • 生产环境建议使用成熟的XA协议实现（如Atomikos、Narayana）。
    
  
  

6. 2PC的局限性

• 同步阻塞：参与者在Prepare阶段后需阻塞等待协调者指令。
  
• 单点故障：协调者宕机可能导致事务悬挂。
  
• 数据不一致：协调者与参与者在Commit阶段同时宕机时，可能部分提交。
  

关注微信公众号，查看更多技术文章。

来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！