分布式事务系列（三）：三阶段提交协议（3PC）深度解析

1
2
3
4
5
6

时间线分析：
T1: 协调者发送Prepare，所有参与者回复YES
T2: 协调者决定COMMIT，开始发送Commit消息
T3: 协调者在发送过程中崩溃
T4: 部分参与者收到Commit，部分未收到
结果: 系统状态不一致，未收到的参与者永久阻塞

1

INITIAL → WAIT_CAN_COMMIT → WAIT_PRE_COMMIT → COMMITTED

1

INITIAL → UNCERTAIN → PREPARED → COMMITTED

1
2
3
4

Message: CAN_COMMIT?
TransactionID: TXN_3PC_001
Query: "Can you commit this transaction?"
RequireResponse: true

1
2

响应时间：通常 < 50ms（无实际操作）
系统状态：所有参与者进入 CAN_COMMIT 状态

1
2
3
4

Message: PRE_COMMIT
TransactionID: TXN_3PC_001
Instruction: "Prepare to commit - lock resources"
Timeout: 30 seconds

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

// 根据第一阶段的投票结果决定下一步操作
if (allParticipantsVotedYes()) {
    // 所有参与者都同意，发送预提交指令
    sendPreCommitToAllParticipants();
    setCoordinatorState(State.PRE_COMMIT_SENT);
} else {
    // 有参与者拒绝，直接中止事务
    sendAbortToAllParticipants();
    setCoordinatorState(State.ABORTED);
}

1
2
3

状态转换：CAN_COMMIT → PRE_COMMIT
资源状态：已锁定，事务已执行但未提交
回复：所有参与者发送 ACK

1
2
3
4

Message: DO_COMMIT
TransactionID: TXN_3PC_001
Instruction: "Commit the transaction"
Final: true

1
2
3

状态转换：PRE_COMMIT → COMMITTED
最终状态：事务成功提交，所有数据已持久化
总耗时：约 150-300ms（比2PC多一轮消息）

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

// 参与者的超时处理逻辑
public class ThreePCParticipant {

    public void handleDoCommitTimeout() {
        if (currentState == PRE_COMMIT) {
            // 3PC的关键改进：默认提交
            logger.info("DoCommit timeout in PRE_COMMIT state, auto committing");

            // 自动提交事务
            commitTransaction();
            currentState = COMMITTED;

            // 通知其他参与者（可选）
            notifyOtherParticipants(COMMITTED);
        }
    }
}

1
2

分区A: 协调者 + 参与者1
分区B: 参与者2 + 参与者3

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

public class PartitionRecovery {

    public void handleNetworkPartition() {
        // 1. 检测到网络分区
        if (detectPartition()) {

            // 2. 尝试联系其他参与者
            List<ParticipantState> otherStates = queryOtherParticipants();

            // 3. 基于状态协商决策
            Decision decision = makeConsensusDecision(otherStates);

            // 4. 执行决策
            executeDecision(decision);
        }
    }

    private Decision makeConsensusDecision(List<ParticipantState> states) {
        // 如果大多数在PRE_COMMIT状态，则提交
        long preCommitCount = states.stream()
            .filter(s -> s == PRE_COMMIT)
            .count();

        if (preCommitCount > states.size() / 2) {
            return Decision.COMMIT;
        } else {
            return Decision.ABORT;
        }
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81

/**
 * 三阶段提交状态枚举
 */
public enum ThreePCState {
    INIT,           // 初始状态
    CAN_COMMIT,     // 可以提交状态（第一阶段后）
    PRE_COMMIT,     // 预提交状态（第二阶段后）
    COMMITTED,      // 已提交状态
    ABORTED;        // 已中止状态

    /**
     * 检查状态转换是否合法
     */
    public boolean canTransitionTo(ThreePCState newState) {
        switch (this) {
            case INIT:
                return newState == CAN_COMMIT || newState == ABORTED;
            case CAN_COMMIT:
                return newState == PRE_COMMIT || newState == ABORTED;
            case PRE_COMMIT:
                return newState == COMMITTED || newState == ABORTED;
            case COMMITTED:
            case ABORTED:
                return false; // 终态，不能再转换
            default:
                return false;
        }
    }
}

/**
 * 三阶段提交参与者接口
 */
public interface ThreePCParticipant {

    /**
     * 第一阶段：询问是否可以提交
     * @param transactionId 事务ID
     * @return 投票结果
     */
    Vote canCommit(String transactionId);

    /**
     * 第二阶段：预提交
     * @param transactionId 事务ID
     * @return 执行结果
     */
    boolean preCommit(String transactionId);

    /**
     * 第三阶段：执行提交
     * @param transactionId 事务ID
     * @return 执行结果
     */
    boolean doCommit(String transactionId);

    /**
     * 中止事务
     * @param transactionId 事务ID
     * @return 执行结果
     */
    boolean abort(String transactionId);

    /**
     * 获取当前状态
     */
    ThreePCState getCurrentState(String transactionId);

    /**
     * 处理超时情况
     */
    void handleTimeout(String transactionId, ThreePCState expectedState);
}

/**
 * 事务操作接口
 */
@FunctionalInterface
public interface TransactionOperation {
    void execute(ThreePCParticipant participant, String transactionId) throws Exception;
}

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221

/**
 * 三阶段提交协调者实现
 */
public class ThreePCCoordinator {
    private final Logger logger = LoggerFactory.getLogger(ThreePCCoordinator.class);
    private final ExecutorService executorService;
    private final TransactionLogger transactionLogger;
    private final TimeoutConfiguration timeoutConfig;

    // 超时配置
    public static class TimeoutConfiguration {
        public final int canCommitTimeoutMs;
        public final int preCommitTimeoutMs;
        public final int doCommitTimeoutMs;

        public TimeoutConfiguration(int canCommitTimeout, int preCommitTimeout, int doCommitTimeout) {
            this.canCommitTimeoutMs = canCommitTimeout;
            this.preCommitTimeoutMs = preCommitTimeout;
            this.doCommitTimeoutMs = doCommitTimeout;
        }
    }

    public ThreePCCoordinator(TimeoutConfiguration timeoutConfig) {
        this.timeoutConfig = timeoutConfig;
        this.executorService = Executors.newCachedThreadPool();
        this.transactionLogger = new TransactionLogger();
    }

    /**
     * 执行三阶段提交事务
     */
    public TransactionResult executeTransaction(String transactionId,
                                              List<ThreePCParticipant> participants,
                                              TransactionOperation operation) {

        logger.info("Starting 3PC transaction: {}", transactionId);
        transactionLogger.logTransactionStart(transactionId, participants);

        try {
            // 第一阶段：CanCommit
            if (!canCommitPhase(transactionId, participants, operation)) {
                abortTransaction(transactionId, participants);
                return TransactionResult.ABORTED;
            }

            // 第二阶段：PreCommit
            if (!preCommitPhase(transactionId, participants)) {
                abortTransaction(transactionId, participants);
                return TransactionResult.ABORTED;
            }

            // 记录提交决策
            transactionLogger.logDecision(transactionId, "COMMIT");

            // 第三阶段：DoCommit
            if (doCommitPhase(transactionId, participants)) {
                transactionLogger.logTransactionComplete(transactionId, ThreePCState.COMMITTED);
                logger.info("3PC transaction committed successfully: {}", transactionId);
                return TransactionResult.COMMITTED;
            } else {
                logger.warn("DoCommit phase had issues for transaction: {}", transactionId);
                return TransactionResult.COMMIT_FAILED_NEED_RETRY;
            }

        } catch (Exception e) {
            logger.error("3PC transaction failed: {}", transactionId, e);
            abortTransaction(transactionId, participants);
            return TransactionResult.ABORTED;
        }
    }

    /**
     * 第一阶段：CanCommit
     */
    private boolean canCommitPhase(String transactionId,
                                  List<ThreePCParticipant> participants,
                                  TransactionOperation operation) {

        logger.info("Starting CanCommit phase for transaction: {}", transactionId);
        transactionLogger.logPhaseStart(transactionId, "CAN_COMMIT");

        // 并发询问所有参与者
        Map<ThreePCParticipant, Future<Vote>> futures = new HashMap<>();

        for (ThreePCParticipant participant : participants) {
            Future<Vote> future = executorService.submit(() -> {
                try {
                    // 先执行操作检查（但不实际执行）
                    return participant.canCommit(transactionId);
                } catch (Exception e) {
                    logger.error("CanCommit failed for participant: {}", participant, e);
                    return Vote.NO;
                }
            });
            futures.put(participant, future);
        }

        // 收集投票结果
        boolean allCanCommit = true;
        for (Map.Entry<ThreePCParticipant, Future<Vote>> entry : futures.entrySet()) {
            try {
                Vote vote = entry.getValue().get(timeoutConfig.canCommitTimeoutMs, TimeUnit.MILLISECONDS);
                if (vote != Vote.YES) {
                    allCanCommit = false;
                    logger.warn("Participant {} voted NO in CanCommit phase", entry.getKey());
                    break;
                }
            } catch (TimeoutException e) {
                logger.warn("CanCommit timeout for participant: {}", entry.getKey());
                allCanCommit = false;
                break;
            } catch (Exception e) {
                logger.error("CanCommit error for participant: {}", entry.getKey(), e);
                allCanCommit = false;
                break;
            }
        }

        transactionLogger.logPhaseComplete(transactionId, "CAN_COMMIT", allCanCommit);
        return allCanCommit;
    }

    /**
     * 第二阶段：PreCommit
     */
    private boolean preCommitPhase(String transactionId, List<ThreePCParticipant> participants) {
        logger.info("Starting PreCommit phase for transaction: {}", transactionId);
        transactionLogger.logPhaseStart(transactionId, "PRE_COMMIT");

        List<Future<Boolean>> futures = new ArrayList<>();

        for (ThreePCParticipant participant : participants) {
            Future<Boolean> future = executorService.submit(() -> {
                try {
                    return participant.preCommit(transactionId);
                } catch (Exception e) {
                    logger.error("PreCommit failed for participant: {}", participant, e);
                    return false;
                }
            });
            futures.add(future);
        }

        // 收集PreCommit结果
        boolean allPreCommitted = true;
        for (Future<Boolean> future : futures) {
            try {
                boolean result = future.get(timeoutConfig.preCommitTimeoutMs, TimeUnit.MILLISECONDS);
                if (!result) {
                    allPreCommitted = false;
                    break;
                }
            } catch (Exception e) {
                logger.error("PreCommit phase error", e);
                allPreCommitted = false;
                break;
            }
        }

        transactionLogger.logPhaseComplete(transactionId, "PRE_COMMIT", allPreCommitted);
        return allPreCommitted;
    }

    /**
     * 第三阶段：DoCommit
     */
    private boolean doCommitPhase(String transactionId, List<ThreePCParticipant> participants) {
        logger.info("Starting DoCommit phase for transaction: {}", transactionId);
        transactionLogger.logPhaseStart(transactionId, "DO_COMMIT");

        List<Future<Boolean>> futures = new ArrayList<>();

        for (ThreePCParticipant participant : participants) {
            Future<Boolean> future = executorService.submit(() -> {
                try {
                    return participant.doCommit(transactionId);
                } catch (Exception e) {
                    logger.error("DoCommit failed for participant: {}", participant, e);
                    return false;
                }
            });
            futures.add(future);
        }

        // 收集DoCommit结果
        boolean allCommitted = true;
        for (Future<Boolean> future : futures) {
            try {
                boolean result = future.get(timeoutConfig.doCommitTimeoutMs, TimeUnit.MILLISECONDS);
                if (!result) {
                    allCommitted = false;
                    // 注意：这里即使失败也要继续，因为已经决定提交
                    logger.warn("DoCommit failed for a participant, but continuing");
                }
            } catch (Exception e) {
                logger.error("DoCommit phase error", e);
                allCommitted = false;
            }
        }

        transactionLogger.logPhaseComplete(transactionId, "DO_COMMIT", allCommitted);
        return allCommitted;
    }

    /**
     * 中止事务
     */
    private void abortTransaction(String transactionId, List<ThreePCParticipant> participants) {
        logger.info("Aborting transaction: {}", transactionId);

        for (ThreePCParticipant participant : participants) {
            try {
                participant.abort(transactionId);
            } catch (Exception e) {
                logger.error("Abort failed for participant: {}", participant, e);
            }
        }

        transactionLogger.logTransactionComplete(transactionId, ThreePCState.ABORTED);
    }
}

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292

/**
 * 数据库参与者的3PC实现
 */
public class DatabaseThreePCParticipant implements ThreePCParticipant {
    private final Logger logger = LoggerFactory.getLogger(DatabaseThreePCParticipant.class);
    private final String participantId;
    private final DataSource dataSource;
    private final Map<String, TransactionContext> transactions;
    private final ScheduledExecutorService timeoutExecutor;

    // 事务上下文
    private static class TransactionContext {
        Connection connection;
        ThreePCState state;
        long lastUpdateTime;
        ScheduledFuture<?> timeoutTask;

        TransactionContext(Connection conn) {
            this.connection = conn;
            this.state = ThreePCState.INIT;
            this.lastUpdateTime = System.currentTimeMillis();
        }
    }

    public DatabaseThreePCParticipant(String participantId, DataSource dataSource) {
        this.participantId = participantId;
        this.dataSource = dataSource;
        this.transactions = new ConcurrentHashMap<>();
        this.timeoutExecutor = Executors.newScheduledThreadPool(2);
    }

    @Override
    public Vote canCommit(String transactionId) {
        try {
            logger.info("Participant {} checking canCommit for transaction {}",
                       participantId, transactionId);

            TransactionContext ctx = transactions.get(transactionId);
            if (ctx == null) {
                // 创建新的事务上下文
                Connection conn = dataSource.getConnection();
                conn.setAutoCommit(false);
                ctx = new TransactionContext(conn);
                transactions.put(transactionId, ctx);
            }

            // 检查是否可以提交（业务逻辑验证）
            if (canPerformTransaction(ctx.connection, transactionId)) {
                ctx.state = ThreePCState.CAN_COMMIT;
                ctx.lastUpdateTime = System.currentTimeMillis();

                // 设置超时任务
                scheduleTimeoutTask(transactionId, ThreePCState.CAN_COMMIT);

                logger.info("Participant {} voted YES for transaction {}",
                           participantId, transactionId);
                return Vote.YES;
            } else {
                logger.warn("Participant {} voted NO for transaction {}",
                           participantId, transactionId);
                return Vote.NO;
            }

        } catch (Exception e) {
            logger.error("CanCommit failed for transaction {}", transactionId, e);
            return Vote.NO;
        }
    }

    @Override
    public boolean preCommit(String transactionId) {
        try {
            logger.info("Participant {} executing preCommit for transaction {}",
                       participantId, transactionId);

            TransactionContext ctx = transactions.get(transactionId);
            if (ctx == null || ctx.state != ThreePCState.CAN_COMMIT) {
                logger.error("Invalid state for preCommit: {}",
                           ctx != null ? ctx.state : "null");
                return false;
            }

            // 执行实际的事务操作（但不提交）
            executeTransactionOperations(ctx.connection, transactionId);

            // 状态转换
            ctx.state = ThreePCState.PRE_COMMIT;
            ctx.lastUpdateTime = System.currentTimeMillis();

            // 取消之前的超时任务，设置新的超时任务
            cancelTimeoutTask(ctx);
            scheduleTimeoutTask(transactionId, ThreePCState.PRE_COMMIT);

            logger.info("Participant {} successfully preCommitted transaction {}",
                       participantId, transactionId);
            return true;

        } catch (Exception e) {
            logger.error("PreCommit failed for transaction {}", transactionId, e);
            return false;
        }
    }

    @Override
    public boolean doCommit(String transactionId) {
        try {
            logger.info("Participant {} executing doCommit for transaction {}",
                       participantId, transactionId);

            TransactionContext ctx = transactions.get(transactionId);
            if (ctx == null || ctx.state != ThreePCState.PRE_COMMIT) {
                logger.error("Invalid state for doCommit: {}",
                           ctx != null ? ctx.state : "null");
                return false;
            }

            // 提交事务
            ctx.connection.commit();

            // 状态转换
            ctx.state = ThreePCState.COMMITTED;
            ctx.lastUpdateTime = System.currentTimeMillis();

            // 清理资源
            cleanupTransaction(transactionId);

            logger.info("Participant {} successfully committed transaction {}",
                       participantId, transactionId);
            return true;

        } catch (Exception e) {
            logger.error("DoCommit failed for transaction {}", transactionId, e);
            return false;
        }
    }

    @Override
    public boolean abort(String transactionId) {
        try {
            logger.info("Participant {} aborting transaction {}",
                       participantId, transactionId);

            TransactionContext ctx = transactions.get(transactionId);
            if (ctx != null) {
                // 回滚事务
                ctx.connection.rollback();
                ctx.state = ThreePCState.ABORTED;

                // 清理资源
                cleanupTransaction(transactionId);
            }

            logger.info("Participant {} successfully aborted transaction {}",
                       participantId, transactionId);
            return true;

        } catch (Exception e) {
            logger.error("Abort failed for transaction {}", transactionId, e);
            return false;
        }
    }

    @Override
    public ThreePCState getCurrentState(String transactionId) {
        TransactionContext ctx = transactions.get(transactionId);
        return ctx != null ? ctx.state : null;
    }

    @Override
    public void handleTimeout(String transactionId, ThreePCState expectedState) {
        TransactionContext ctx = transactions.get(transactionId);
        if (ctx == null || ctx.state != expectedState) {
            return;
        }

        logger.warn("Timeout occurred for transaction {} in state {}",
                   transactionId, expectedState);

        switch (expectedState) {
            case CAN_COMMIT:
                // CanCommit超时，自动中止
                logger.info("CanCommit timeout, auto aborting transaction {}", transactionId);
                abort(transactionId);
                break;

            case PRE_COMMIT:
                // PreCommit超时，自动提交（3PC的关键特性！）
                logger.info("PreCommit timeout, auto committing transaction {}", transactionId);
                doCommit(transactionId);
                break;

            default:
                logger.warn("Unexpected timeout state: {}", expectedState);
                break;
        }
    }

    /**
     * 设置超时任务
     */
    private void scheduleTimeoutTask(String transactionId, ThreePCState state) {
        TransactionContext ctx = transactions.get(transactionId);
        if (ctx == null) return;

        long timeoutMs = getTimeoutForState(state);

        ScheduledFuture<?> timeoutTask = timeoutExecutor.schedule(() -> {
            handleTimeout(transactionId, state);
        }, timeoutMs, TimeUnit.MILLISECONDS);

        ctx.timeoutTask = timeoutTask;
    }

    /**
     * 取消超时任务
     */
    private void cancelTimeoutTask(TransactionContext ctx) {
        if (ctx.timeoutTask != null && !ctx.timeoutTask.isDone()) {
            ctx.timeoutTask.cancel(false);
        }
    }

    /**
     * 获取状态对应的超时时间
     */
    private long getTimeoutForState(ThreePCState state) {
        switch (state) {
            case CAN_COMMIT:
                return 30000; // 30秒
            case PRE_COMMIT:
                return 60000; // 60秒
            default:
                return 30000;
        }
    }

    /**
     * 检查是否可以执行事务
     */
    private boolean canPerformTransaction(Connection conn, String transactionId) {
        // 实现具体的业务检查逻辑
        try {
            // 检查连接状态
            if (conn.isClosed()) {
                return false;
            }

            // 可以添加更多的业务检查
            // 例如：资源可用性、约束条件验证等

            return true;
        } catch (SQLException e) {
            logger.error("Error checking transaction capability", e);
            return false;
        }
    }

    /**
     * 执行事务操作
     */
    private void executeTransactionOperations(Connection conn, String transactionId) throws SQLException {
        // 实现具体的事务操作
        // 这里只是示例，实际应该根据业务需求实现

        try (PreparedStatement stmt = conn.prepareStatement(
             "UPDATE account SET balance = balance + ? WHERE id = ?")) {
            stmt.setDouble(1, 100.0);
            stmt.setString(2, "test_account");
            stmt.executeUpdate();
        }

        logger.debug("Transaction operations executed for {}", transactionId);
    }

    /**
     * 清理事务资源
     */
    private void cleanupTransaction(String transactionId) {
        TransactionContext ctx = transactions.remove(transactionId);
        if (ctx != null) {
            cancelTimeoutTask(ctx);

            try {
                if (ctx.connection != null && !ctx.connection.isClosed()) {
                    ctx.connection.close();
                }
            } catch (SQLException e) {
                logger.error("Error closing connection for transaction {}", transactionId, e);
            }
        }
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64

/**
 * 3PC使用示例
 */
public class ThreePCExample {

    public static void main(String[] args) {
        // 创建超时配置
        ThreePCCoordinator.TimeoutConfiguration timeoutConfig =
            new ThreePCCoordinator.TimeoutConfiguration(15000, 30000, 45000);

        // 创建协调者
        ThreePCCoordinator coordinator = new ThreePCCoordinator(timeoutConfig);

        // 创建参与者
        ThreePCParticipant db1 = new DatabaseThreePCParticipant("DB1", createDataSource("db1"));
        ThreePCParticipant db2 = new DatabaseThreePCParticipant("DB2", createDataSource("db2"));
        ThreePCParticipant db3 = new DatabaseThreePCParticipant("DB3", createDataSource("db3"));

        List<ThreePCParticipant> participants = Arrays.asList(db1, db2, db3);

        // 执行分布式事务
        String transactionId = "3PC_TXN_" + System.currentTimeMillis();

        // 定义事务操作
        TransactionOperation operation = (participant, txnId) -> {
            if (participant instanceof DatabaseThreePCParticipant) {
                // 这里定义具体的业务操作
                logger.info("Executing business operation for participant {} in transaction {}",
                           participant, txnId);
            }
        };

        // 执行事务
        TransactionResult result = coordinator.executeTransaction(
            transactionId, participants, operation);

        // 处理结果
        handleTransactionResult(result, transactionId);
    }

    private static void handleTransactionResult(TransactionResult result, String transactionId) {
        switch (result) {
            case COMMITTED:
                System.out.println("✅ 3PC Transaction committed successfully: " + transactionId);
                break;
            case ABORTED:
                System.out.println("❌ 3PC Transaction aborted: " + transactionId);
                break;
            case COMMIT_FAILED_NEED_RETRY:
                System.out.println("⚠️ 3PC Transaction commit failed, need retry: " + transactionId);
                // 实现重试逻辑
                break;
        }
    }

    private static DataSource createDataSource(String dbName) {
        HikariConfig config = new HikariConfig();
        config.setJdbcUrl("jdbc:mysql://localhost:3306/" + dbName);
        config.setUsername("username");
        config.setPassword("password");
        config.setMaximumPoolSize(10);
        return new HikariDataSource(config);
    }
}