Seata AT 模式落地实录：订单-库存-积分的分布式事务治理与踩坑全记录

支付回调到达
    ├── 订单服务：将订单状态从 PENDING 更新为 PAID
    ├── 库存服务：确认扣减库存（从"预扣"变为"实扣"）
    └── 积分服务：给用户增加活动积分（用于平台内的等级和权益）

                        ┌─────────────┐
                        │ Seata Server│
                        │   (TC)      │
                        └──────┬──────┘
                               │ 协调全局事务
              ┌────────────────┼────────────────┐
              ▼                ▼                ▼
      ┌──────────────┐ ┌──────────────┐ ┌──────────────┐
      │  订单服务     │ │  库存服务     │ │  积分服务     │
      │  (TM + RM)   │ │  (RM)        │ │  (RM)        │
      │  order_db    │ │  stock_db    │ │  points_db   │
      │  undo_log    │ │  undo_log    │ │  undo_log    │
      └──────────────┘ └──────────────┘ └──────────────┘

CREATE TABLE `undo_log` (
  `branch_id` BIGINT NOT NULL COMMENT '分支事务ID',
  `xid` VARCHAR(128) NOT NULL COMMENT '全局事务ID',
  `context` VARCHAR(128) NOT NULL COMMENT '上下文',
  `rollback_info` LONGBLOB NOT NULL COMMENT '回滚数据（前后镜像）',
  `log_status` INT NOT NULL COMMENT '状态',
  `log_created` DATETIME NOT NULL,
  `log_modified` DATETIME NOT NULL,
  UNIQUE KEY `ux_undo_log` (`xid`, `branch_id`)
) ENGINE = InnoDB;

@Service
public class PaymentCallbackService {
​
    @Autowired
    private OrderMapper orderMapper;
    @Autowired
    private StockFeignClient stockClient;
    @Autowired
    private PointsFeignClient pointsClient;
​
    @GlobalTransactional(name = "payment-callback-tx", timeoutMills = 30000, rollbackFor = Exception.class)
    public void handlePaymentSuccess(PaymentCallbackDTO callback) {
        Long orderId = callback.getOrderId();
        Long activityId = callback.getActivityId();
        Long userId = callback.getUserId();
        Integer amount = callback.getAmount();
​
        // 1. 更新订单状态
        int affected = orderMapper.updateStatus(orderId, OrderStatus.PENDING, OrderStatus.PAID);
        if (affected == 0) {
            throw new BusinessException("订单状态更新失败，可能已被处理");
        }
​
        // 2. 确认库存扣减（从预扣变为实扣）
        Result<?> stockResult = stockClient.confirmDeduct(activityId, orderId);
        if (!stockResult.isSuccess()) {
            throw new BusinessException("库存确认失败: " + stockResult.getMessage());
        }
​
        // 3. 增加用户积分
        Result<?> pointsResult = pointsClient.addPoints(userId, activityId, calculatePoints(amount));
        if (!pointsResult.isSuccess()) {
            throw new BusinessException("积分增加失败: " + pointsResult.getMessage());
        }
​
        log.info("支付回调处理成功 orderId={}, activityId={}, userId={}", orderId, activityId, userId);
    }
​
    private int calculatePoints(Integer payAmount) {
        // 每消费1元获得10积分
        return payAmount * 10;
    }
}

@Service
public class StockServiceImpl implements StockService {
​
    @Autowired
    private StockMapper stockMapper;
​
    @Transactional // 普通的本地事务注解即可，Seata会自动代理
    @Override
    public void confirmDeduct(Long activityId, Long orderId) {
        // 将预扣库存标记为已确认
        int affected = stockMapper.confirmDeduct(activityId, orderId);
        if (affected == 0) {
            throw new BusinessException("库存确认失败，预扣记录不存在");
        }
    }
}

# Seata Server 配置
store:
  mode: db
  db:
    datasource: druid
    db-type: mysql
    url: jdbc:mysql://seata-db:3306/seata_server
    user: seata
    password: ****
    min-conn: 10
    max-conn: 30

-- 改造前：所有事务争抢修改同一行
UPDATE activity_stock SET confirmed = confirmed + 1 WHERE activity_id = #{activityId};
​
-- 改造后：每个订单修改自己的预扣记录（不同行，不竞争全局锁）
UPDATE stock_deduct_log
SET status = 'CONFIRMED'
WHERE order_id = #{orderId} AND activity_id = #{activityId};

store:
  mode: redis
  redis:
    host: seata-redis
    port: 6379
    database: 0
    min-conn: 10
    max-conn: 30

@GlobalTransactional(name = "payment-callback-tx", timeoutMills = 30000)
public void handlePaymentSuccess(PaymentCallbackDTO callback) {
    // ✅ 只有这三个操作需要分布式事务保护
    orderMapper.updateStatus(callback.getOrderId(), OrderStatus.PENDING, OrderStatus.PAID);
    stockClient.confirmDeduct(callback.getActivityId(), callback.getOrderId());
    pointsClient.addPoints(callback.getUserId(), callback.getActivityId(), calculatePoints(callback.getAmount()));
}

// ❌ 通知和日志移到事务外面，用事件驱动
@TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
public void afterPaymentSuccess(PaymentSuccessEvent event) {
    notifyService.sendPaymentSuccessNotification(event.getUserId());
    auditLogService.recordPayment(event);
}

@Component
public class CustomFailureHandler extends DefaultFailureHandler {

    @Autowired
    private AlertService alertService;
    @Autowired
    private CompensationService compensationService;

    @Override
    public void onRollbackFailure(GlobalTransaction tx, Throwable originalException) {
        String xid = tx.getXid();

        // 1. 告警：通知值班人员
        alertService.sendCritical(
            "Seata回滚失败告警",
            String.format("全局事务 %s 回滚失败，需要人工介入。原始异常: %s",
                xid, originalException.getMessage())
        );

        // 2. 记录到补偿表，等待人工处理
        compensationService.recordRollbackFailure(xid, originalException);

        // 3. 调用父类的默认处理（持续重试）
        super.onRollbackFailure(tx, originalException);

        log.error("全局事务 {} 回滚失败，已记录补偿任务并触发告警", xid, originalException);
    }
}

CREATE TABLE `seata_compensation` (
    `id` BIGINT PRIMARY KEY AUTO_INCREMENT,
    `xid` VARCHAR(128) NOT NULL COMMENT '全局事务ID',
    `order_id` BIGINT COMMENT '关联订单ID',
    `failure_reason` TEXT COMMENT '失败原因',
    `status` VARCHAR(32) DEFAULT 'PENDING' COMMENT '处理状态',
    `handled_by` VARCHAR(64) COMMENT '处理人',
    `handled_at` DATETIME COMMENT '处理时间',
    `handle_note` TEXT COMMENT '处理备注',
    `created_at` DATETIME DEFAULT CURRENT_TIMESTAMP
);