分布式事务：基于 Saga 补偿的状态机实践

public interface Transition {
    TradeState getCurrent();                    // 当前状态
    TradeState getSuccessState();               // 成功后跳转到
    TradeState getFailState();                  // 失败后跳转到
    Class<? extends StateUnit> getStateUnit();  // 该状态用哪个执行单元
}

@Override
public void setApplicationContext(ApplicationContext ctx) {
    for (Transition t : transitions) {
        StateUnit unit = ctx.getBean(t.getStateUnit());   // Class 引用 → 容器 Bean
        Map<Boolean, Integer> nextMap = new HashMap<>();
        nextMap.put(true,  t.getSuccessState().getId());  // 成功态
        nextMap.put(false, t.getFailState().getId());     // 失败态
        transitionMap.put(t.getCurrent().getId(), new Pair<>(unit, nextMap));
    }
}

public int getNextTradeStatusByResult(TransactionEventDTO event, MachineStateEnum result) {
    int state = event.getTradeState();
    if (result == MachineStateEnum.FAIL) {
        return getTransitionManager().getFail(state);      // 失败 → 补偿链
    } else if (result == MachineStateEnum.SUCCESS) {
        return getTransitionManager().getSuccess(state);   // 成功 → 正向链
    } else if (result == MachineStateEnum.PROCESSING) {
        return state;                                      // 处理中 → 原地不动
    }
    throw new RuntimeException("get next error");
}

public enum PayJumpTransition implements Transition {
    //                当前状态     成功去 →           失败去 →            执行单元
    FREEZE_COUPON(FREEZE_COUPON, FREEZE_LIMIT,   UPDATE_TRADE_FAIL, FreezeCouponUnit.class),
    FREEZE_LIMIT (FREEZE_LIMIT,  CREATE_LOAN,    UNFREEZE_COUPON,   FreezeLimitUnit.class),
    // …… 正向链一路到 终态

    // 补偿链
    UNFREEZE_COUPON(UNFREEZE_COUPON,   UPDATE_TRADE_FAIL, UNFREEZE_COUPON,  UnfreezeCouponUnit.class),
    UPDATE_TRADE_FAIL(UPDATE_TRADE_FAIL, SEND_FAIL_MQ, UPDATE_TRADE_FAIL, UpdateTradeFailUnit.class);
}

CREATE TABLE `main_transaction` (
  `id`             bigint      NOT NULL AUTO_INCREMENT COMMENT '主键ID',
  `instruction_id` bigint      NOT NULL COMMENT '业务指令ID，一笔支用/退款的唯一标识',
  `trade_code`     int         NOT NULL COMMENT '交易类型：支用/退款/预冻结支用，路由到不同状态机',
  `trade_state`    int         NOT NULL COMMENT '当前状态机状态(TradeState.id)，Saga 的游标',
  `fail_state`     int         NOT NULL DEFAULT 0 COMMENT '首次失败发生在哪个状态，用于排查/补偿定位',
  `run_times`      int         NOT NULL DEFAULT 0 COMMENT '状态机累计执行次数，用于限制重试上限',
  `product_id`     varchar(64) NOT NULL COMMENT '产品ID',
  `context`        text        COMMENT '原始请求参数(静态)',
  `biz_data`       text        COMMENT '状态机执行过程中产生的业务数据',
  `create_time`    datetime(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
  `update_time`    datetime(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3),
  PRIMARY KEY (`id`),
  UNIQUE KEY `uk_instruction` (`instruction_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='主交易单';

CREATE TABLE `main_transaction_log` (
  `id`                 bigint NOT NULL AUTO_INCREMENT,
  `instruction_id`     bigint NOT NULL COMMENT '业务指令ID',
  `trade_code`         int    NOT NULL COMMENT '交易类型',
  `before_trade_state` int    NOT NULL COMMENT '跳转前状态',
  `after_trade_state`  int    NOT NULL COMMENT '跳转后状态',
  `trace_id`           varchar(64)  COMMENT '链路追踪ID',
  `operator`           varchar(64)  COMMENT '操作人',
  `create_time`        datetime(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
  PRIMARY KEY (`id`),
  KEY `idx_instruction` (`instruction_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='状态流转日志';

private MachineStateEnum machineExecute(Long instructionId) {
    MachineStateEnum result = MachineStateEnum.PROCESSING;
    TransactionEventDTO event = transactionService.getByInstructionId(instructionId);
    if (isFinal(event)) {
        return MachineStateEnum.SUCCESS;          // 已终态，幂等直接返回
    }

    while (!isFinal(event)) {                     // 只要没到终态，一直往前推
        int stepBefore = event.getTradeState();
        StateUnit stateUnit = getStateUnit(event);   // 查转移表拿执行单元
        try {
            result = stateUnit.execute(event);       // 执行业务
        } catch (Exception e) {
            result = MachineStateEnum.PROCESSING;    // 异常当作可重试
        }
        event.setRunTimes(event.getRunTimes() + 1);

        if (MachineStateEnum.FAIL.equals(result) && event.getFailState() == 0) {
            event.setFailState(stepBefore);          // 记录首次失败点
            transactionService.updateTransaction(event, stepBefore);
        }

        int stepAfter = getNextTradeStatusByResult(event, result);   // 查表算下一步
        if (stepAfter == stepBefore) {               // 推不动了(PROCESSING/自重试)
            transactionService.updateTransaction(event, stepBefore);
            result = MachineStateEnum.PROCESSING;
            break;
        }
        event.setTradeState(stepAfter);              // 推进游标

        if (transactionService.updateTransaction(event, stepBefore) != 1) {  // 乐观锁落库
            throw new RuntimeException(UPDATE_ERROR);     // 被别人改过，放手
        }
    }

    if (!isFinal(event)) {                           // 没推到终态 → 自驱动，延迟后再来一次
        if (event.getRunTimes() > maxRetryTimes) {
            return result;                           // 超自驱动上限，交给外部定时兜底
        }
        scheduledExecutor.schedule(
            () -> machineStateExecutor.execute(() -> run(event.getInstructionId())),
            delayRunSecond, TimeUnit.SECONDS);
    }
    return result;
}

UPDATE main_transaction
SET trade_state = #{stepAfter}, run_times = #{runTimes}, fail_state = #{failState}
WHERE instruction_id = #{instructionId}
  AND trade_state = #{stepBefore};   -- 乐观锁：状态被改过就更新 0 行