2PC 两阶段提交#

public class TwoPhaseCommitCoordinator {

    private final List<Participant> participants;
    private final Map<String, TransactionState> transactions;

    public static void main(String[] args) {
        TwoPhaseCommitCoordinator coordinator = new TwoPhaseCommitCoordinator();
        coordinator.executeDistributedTransaction();
    }

    public boolean executeDistributedTransaction() {
        String transactionId = UUID.randomUUID().toString();
        System.out.println("开始分布式事务: " + transactionId);

        // 第一阶段：Prepare
        boolean allPrepared = phaseOnePrepare(transactionId);
        if (!allPrepared) {
            System.out.println("Prepare 阶段失败，执行回滚");
            phaseTwoAbort(transactionId);
            return false;
        }

        // 第二阶段：Commit
        System.out.println("所有参与者 Prepare 成功，执行提交");
        phaseTwoCommit(transactionId);
        return true;
    }

    private boolean phaseOnePrepare(String transactionId) {
        // 向所有参与者发送 Prepare
        List<Future<Boolean>> results = new ArrayList<>();
        for (Participant participant : participants) {
            Future<Boolean> future = CompletableFuture.supplyAsync(() ->
                participant.prepare(transactionId)
            );
            results.add(future);
        }

        // 等待所有响应
        for (Future<Boolean> future : results) {
            try {
                Boolean vote = future.get(30, TimeUnit.SECONDS);
                if (!vote) {
                    return false; // 任何一个说 No，全部回滚
                }
            } catch (TimeoutException e) {
                System.out.println("参与者 Prepare 超时");
                return false;
            } catch (InterruptedException | ExecutionException e) {
                return false;
            }
        }
        return true;
    }

    private void phaseTwoCommit(String transactionId) {
        for (Participant participant : participants) {
            try {
                participant.commit(transactionId);
            } catch (Exception e) {
                // 记录日志，后续补偿
                System.out.println("提交失败，需要人工介入: " + participant);
            }
        }
    }

    private void phaseTwoAbort(String transactionId) {
        for (Participant participant : participants) {
            participant.rollback(transactionId);
        }
    }
}

interface Participant {
    boolean prepare(String transactionId);
    void commit(String transactionId);
    void rollback(String transactionId);
}