分布式事务与最终一致性

// XA 事务示例（MySQL + Atomikos）
import com.atomikos.jdbc.AtomikosDataSourceBean;
import javax.transaction.UserTransaction;

public class XATransactionExample {
    
    private AtomikosDataSourceBean orderDataSource;
    private AtomikosDataSourceBean inventoryDataSource;
    
    public void createOrderWithXA(Order order) {
        UserTransaction utx = getUserTransaction();
        try {
            utx.begin();
            
            // 操作订单库
            Connection orderConn = orderDataSource.getConnection();
            PreparedStatement orderStmt = orderConn.prepareStatement(
                "INSERT INTO orders (order_no, amount) VALUES (?, ?)"
            );
            orderStmt.setString(1, order.getOrderNo());
            orderStmt.setBigDecimal(2, order.getAmount());
            orderStmt.executeUpdate();
            
            // 操作库存库
            Connection invConn = inventoryDataSource.getConnection();
            PreparedStatement invStmt = invConn.prepareStatement(
                "UPDATE inventory SET stock = stock - ? WHERE product_id = ?"
            );
            invStmt.setInt(1, order.getQuantity());
            invStmt.setLong(2, order.getProductId());
            invStmt.executeUpdate();
            
            utx.commit();
        } catch (Exception e) {
            utx.rollback();
            throw new RuntimeException("XA transaction failed", e);
        }
    }
}

CREATE TABLE outbox_message (
    id BIGINT PRIMARY KEY AUTO_INCREMENT,
    topic VARCHAR(100) NOT NULL COMMENT '消息主题',
    biz_key VARCHAR(200) NOT NULL COMMENT '业务唯一键',
    payload TEXT NOT NULL COMMENT '消息内容',
    status VARCHAR(20) NOT NULL COMMENT 'NEW-新建, SENT-已发送, FAILED-失败',
    retry_count INT DEFAULT 0 COMMENT '重试次数',
    create_time DATETIME DEFAULT CURRENT_TIMESTAMP,
    update_time DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
    INDEX idx_status_create (status, create_time),
    UNIQUE KEY uk_biz_key (biz_key)
) COMMENT '本地消息表';

@Service
public class OrderService {
    
    @Autowired
    private OrderRepository orderRepository;
    
    @Autowired
    private OutboxMessageRepository outboxRepository;
    
    @Transactional(rollbackFor = Exception.class)
    public Order createOrder(CreateOrderCommand command) {
        // 1. 创建订单
        Order order = Order.create(command);
        orderRepository.save(order);
        
        // 2. 写入本地消息表（同一个事务）
        OutboxMessage message = OutboxMessage.builder()
            .topic("order-created")
            .bizKey(order.getOrderNo())
            .payload(toJson(order))
            .status("NEW")
            .build();
        outboxRepository.save(message);
        
        return order;
    }
    
    private String toJson(Object obj) {
        try {
            return new ObjectMapper().writeValueAsString(obj);
        } catch (JsonProcessingException e) {
            throw new RuntimeException("JSON serialization failed", e);
        }
    }
}

@Component
public class MessagePublisher {
    
    @Autowired
    private OutboxMessageRepository outboxRepository;
    
    @Autowired
    private RocketMQTemplate rocketMQTemplate;
    
    @Scheduled(fixedDelay = 5000) // 每 5 秒执行一次
    public void publishPendingMessages() {
        // 查询待发送的消息（限制数量避免一次处理太多）
        List<OutboxMessage> messages = outboxRepository
            .findTop100ByStatusOrderByCreateTimeAsc("NEW");
        
        for (OutboxMessage msg : messages) {
            try {
                // 发送到 MQ
                SendResult result = rocketMQTemplate.syncSend(
                    msg.getTopic(), 
                    msg.getPayload()
                );
                
                // 发送成功，更新状态
                if (result.getSendStatus() == SendStatus.SEND_OK) {
                    msg.setStatus("SENT");
                    msg.setUpdateTime(LocalDateTime.now());
                    outboxRepository.save(msg);
                }
            } catch (Exception e) {
                // 发送失败，增加重试次数
                msg.setRetryCount(msg.getRetryCount() + 1);
                if (msg.getRetryCount() >= 5) {
                    msg.setStatus("FAILED");
                    // 发送告警
                    alertService.sendAlert("消息发送失败: " + msg.getBizKey());
                }
                outboxRepository.save(msg);
            }
        }
    }
}

@Service
@RocketMQMessageListener(
    topic = "order-created",
    consumerGroup = "inventory-consumer"
)
public class InventoryConsumer implements RocketMQListener<String> {
    
    @Autowired
    private InventoryService inventoryService;
    
    @Autowired
    private DeductionRecordRepository deductionRepository;
    
    @Override
    public void onMessage(String message) {
        Order order = parseOrder(message);
        
        // 幂等检查：是否已处理过
        if (deductionRepository.existsByOrderNo(order.getOrderNo())) {
            log.info("订单已处理，跳过: {}", order.getOrderNo());
            return;
        }
        
        // 扣减库存
        inventoryService.deductStock(
            order.getProductId(), 
            order.getQuantity(),
            order.getOrderNo()
        );
    }
}

@Service
public class OrderTransactionProducer {
    
    @Autowired
    private RocketMQTemplate rocketMQTemplate;
    
    public void createOrderWithTransaction(Order order) {
        // 构建消息
        Message<String> message = MessageBuilder
            .withPayload(toJson(order))
            .build();
        
        // 发送事务消息
        rocketMQTemplate.sendMessageInTransaction(
            "order-group",
            "order-created",
            message,
            order // 传递给本地事务执行的参数
        );
    }
}

@RocketMQTransactionListener(rocketMQTemplateBeanName = "rocketMQTemplate")
class OrderTransactionListener implements RocketMQLocalTransactionListener {
    
    @Autowired
    private OrderService orderService;
    
    @Autowired
    private OrderRepository orderRepository;
    
    @Override
    @Transactional(rollbackFor = Exception.class)
    public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        try {
            Order order = (Order) arg;
            // 执行本地事务：创建订单
            orderService.createOrder(order);
            
            // 返回提交状态，消息对消费者可见
            return RocketMQLocalTransactionState.COMMIT;
        } catch (Exception e) {
            // 返回回滚状态，消息被丢弃
            return RocketMQLocalTransactionState.ROLLBACK;
        }
    }
    
    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message msg) {
        // 回查本地事务状态
        String orderNo = extractOrderNo(msg);
        
        if (orderRepository.existsByOrderNo(orderNo)) {
            // 订单存在，提交消息
            return RocketMQLocalTransactionState.COMMIT;
        } else {
            // 订单不存在，回滚消息
            return RocketMQLocalTransactionState.ROLLBACK;
        }
    }
}

@LocalTCC
public interface InventoryTccService {
    
    /**
     * Try: 预占库存
     */
    @TwoPhaseBusinessAction(
        name = "prepareDeduct",
        commitMethod = "commit",
        rollbackMethod = "rollback"
    )
    boolean prepareDeduct(
        @BusinessActionContextParameter(paramName = "productId") Long productId,
        @BusinessActionContextParameter(paramName = "quantity") Integer quantity,
        @BusinessActionContextParameter(paramName = "orderNo") String orderNo
    );
    
    /**
     * Confirm: 确认扣减
     */
    boolean commit(BusinessActionContext context);
    
    /**
     * Cancel: 释放预留库存
     */
    boolean rollback(BusinessActionContext context);
}

@Service
public class InventoryTccServiceImpl implements InventoryTccService {
    
    @Autowired
    private InventoryRepository inventoryRepository;
    
    @Autowired
    private InventoryFreezeRepository freezeRepository;
    
    @Override
    @Transactional(rollbackFor = Exception.class)
    public boolean prepareDeduct(Long productId, Integer quantity, String orderNo) {
        // 1. 检查库存是否充足
        Inventory inventory = inventoryRepository.findByProductId(productId);
        if (inventory.getAvailableStock() < quantity) {
            throw new RuntimeException("库存不足");
        }
        
        // 2. 冻结库存（预占）
        inventory.setAvailableStock(inventory.getAvailableStock() - quantity);
        inventory.setFrozenStock(inventory.getFrozenStock() + quantity);
        inventoryRepository.save(inventory);
        
        // 3. 记录冻结明细（用于回滚）
        InventoryFreeze freeze = InventoryFreeze.builder()
            .orderNo(orderNo)
            .productId(productId)
            .quantity(quantity)
            .status("FROZEN")
            .createTime(LocalDateTime.now())
            .build();
        freezeRepository.save(freeze);
        
        return true;
    }
    
    @Override
    @Transactional(rollbackFor = Exception.class)
    public boolean commit(BusinessActionContext context) {
        String orderNo = context.getActionContext("orderNo", String.class);
        Integer quantity = context.getActionContext("quantity", Integer.class);
        Long productId = context.getActionContext("productId", Long.class);
        
        // 1. 删除冻结记录（幂等处理）
        InventoryFreeze freeze = freezeRepository.findByOrderNo(orderNo);
        if (freeze == null) {
            // 已处理过，幂等返回
            return true;
        }
        freezeRepository.delete(freeze);
        
        // 2. 真正扣减冻结库存
        Inventory inventory = inventoryRepository.findByProductId(productId);
        inventory.setFrozenStock(inventory.getFrozenStock() - quantity);
        inventoryRepository.save(inventory);
        
        return true;
    }
    
    @Override
    @Transactional(rollbackFor = Exception.class)
    public boolean rollback(BusinessActionContext context) {
        String orderNo = context.getActionContext("orderNo", String.class);
        Integer quantity = context.getActionContext("quantity", Integer.class);
        Long productId = context.getActionContext("productId", Long.class);
        
        // 1. 查询冻结记录（幂等处理）
        InventoryFreeze freeze = freezeRepository.findByOrderNo(orderNo);
        if (freeze == null) {
            // 已回滚过，幂等返回
            return true;
        }
        
        // 2. 恢复可用库存
        Inventory inventory = inventoryRepository.findByProductId(productId);
        inventory.setAvailableStock(inventory.getAvailableStock() + quantity);
        inventory.setFrozenStock(inventory.getFrozenStock() - quantity);
        inventoryRepository.save(inventory);
        
        // 3. 删除冻结记录
        freezeRepository.delete(freeze);
        
        return true;
    }
}

@Override
public boolean commit(BusinessActionContext context) {
    String orderNo = context.getActionContext("orderNo", String.class);
    
    // 幂等检查：是否已处理
    InventoryFreeze freeze = freezeRepository.findByOrderNo(orderNo);
    if (freeze == null) {
        log.info("订单已提交，幂等返回: {}", orderNo);
        return true;
    }
    
    // 执行提交逻辑...
}

@Override
public boolean rollback(BusinessActionContext context) {
    String orderNo = context.getActionContext("orderNo", String.class);
    
    // 空回滚检查：Try 是否执行过
    InventoryFreeze freeze = freezeRepository.findByOrderNo(orderNo);
    if (freeze == null) {
        log.warn("Try 未执行，空回滚: {}", orderNo);
        // 记录空回滚标记，防止悬挂
        recordEmptyRollback(orderNo);
        return true;
    }
    
    // 执行回滚逻辑...
}

@Override
public boolean prepareDeduct(Long productId, Integer quantity, String orderNo) {
    // 悬挂检查：是否已回滚
    if (hasEmptyRollbackRecord(orderNo)) {
        log.warn("已回滚，拒绝 Try: {}", orderNo);
        return false;
    }
    
    // 执行 Try 逻辑...
}

@Service
public class OrderSagaService {
    
    @Autowired
    private OrderService orderService;
    
    @Autowired
    private InventoryService inventoryService;
    
    @Autowired
    private PaymentService paymentService;
    
    @Autowired
    private PointService pointService;
    
    public void createOrderSaga(Order order) {
        List<Runnable> compensations = new ArrayList<>();
        
        try {
            // 步骤 1：创建订单
            orderService.createOrder(order);
            compensations.add(() -> orderService.cancelOrder(order.getOrderNo()));
            
            // 步骤 2：扣减库存
            inventoryService.deductStock(order.getProductId(), order.getQuantity());
            compensations.add(() -> inventoryService.restoreStock(
                order.getProductId(), 
                order.getQuantity()
            ));
            
            // 步骤 3：扣款
            paymentService.charge(order.getUserId(), order.getAmount());
            compensations.add(() -> paymentService.refund(
                order.getUserId(), 
                order.getAmount()
            ));
            
            // 步骤 4：发放积分（允许失败，不补偿）
            try {
                pointService.awardPoints(order.getUserId(), order.getPoints());
            } catch (Exception e) {
                log.warn("积分发放失败，不影响主流程", e);
            }
            
        } catch (Exception e) {
            // 执行补偿（按相反顺序）
            Collections.reverse(compensations);
            for (Runnable compensation : compensations) {
                try {
                    compensation.run();
                } catch (Exception ex) {
                    log.error("补偿失败", ex);
                }
            }
            throw new RuntimeException("Saga 执行失败", e);
        }
    }
}

订单状态：CREATED → PAID → COMPLETED
库存状态：已扣减
积分状态：待发放 → 已发放

失败场景：
- 库存扣减失败 → 订单取消
- 积分发放失败 → 定时补偿重试

@Service
public class PaymentCallbackService {
    
    @Transactional(rollbackFor = Exception.class)
    public void handlePaymentCallback(PaymentCallback callback) {
        // 1. 幂等检查
        if (paymentRecordRepository.existsByPayNo(callback.getPayNo())) {
            log.info("支付回调已处理: {}", callback.getPayNo());
            return;
        }
        
        // 2. 查询订单
        Order order = orderRepository.findByOrderNo(callback.getOrderNo());
        
        // 3. 状态机检查
        if (!order.canTransitionTo(OrderStatus.PAID)) {
            throw new IllegalStateException(
                "订单状态不允许支付: " + order.getStatus()
            );
        }
        
        // 4. 更新订单状态
        order.setStatus(OrderStatus.PAID);
        order.setPayTime(LocalDateTime.now());
        orderRepository.save(order);
        
        // 5. 记录支付流水
        PaymentRecord record = PaymentRecord.builder()
            .payNo(callback.getPayNo())
            .orderNo(order.getOrderNo())
            .amount(callback.getAmount())
            .payTime(callback.getPayTime())
            .build();
        paymentRecordRepository.save(record);
        
        // 6. 发送下游事件（异步）
        eventPublisher.publish("order-paid", order);
    }
}

是否允许短暂不一致？
├─ 否 → 考虑 2PC/XA（强一致，性能低）
└─ 是
    └─ 业务是否可拆分为 Try-Confirm-Cancel？
        ├─ 是 → 考虑 TCC（高一致性，实现复杂）
        └─ 否
            └─ 是否需要保证消息发送与本地事务的原子性？
                ├─ 是 → 事务消息（推荐 RocketMQ）
                └─ 否 → 本地消息表 + 异步补偿

-- 查询长时间未完成的事务
SELECT * FROM outbox_message 
WHERE status = 'NEW' 
  AND create_time < DATE_SUB(NOW(), INTERVAL 10 MINUTE)
ORDER BY create_time;

-- 查询失败的消息
SELECT * FROM outbox_message 
WHERE status = 'FAILED' 
ORDER BY create_time DESC;

-- 查询重试次数过多的消息
SELECT * FROM outbox_message 
WHERE retry_count > 3 
ORDER BY retry_count DESC;