#RocketMQ 事务消息原理

// 事务消息监听器
public class OrderTransactionListener implements TransactionListener {

    // 阶段2：执行本地事务（这个方法在发送半消息后由Broker回调）
    @Override
    public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        String orderId = msg.getKeys();
        Order order = (Order) arg;

        try {
            // 1. 创建订单（本地事务）
            orderService.createOrder(order);

            // 2. 扣减库存
            inventoryService.deduct(order.getItems());

            // 3. 扣减余额
            accountService.deduct(order.getUserId(), order.getAmount());

            // 全部成功，提交事务消息
            return LocalTransactionState.COMMIT_MESSAGE;

        } catch (InsufficientBalanceException e) {
            // 余额不足，回滚订单和库存
            orderService.cancelOrder(orderId);
            inventoryService.restore(order.getItems());
            return LocalTransactionState.ROLLBACK_MESSAGE;

        } catch (Exception e) {
            // 其他异常，先返回UNKNOWN，让MQ回查
            // 如果是临时故障（如数据库连接超时），回查后可能成功
            return LocalTransactionState.UNKNOW;
        }
    }

    // 阶段3：回查本地事务状态（MQ主动调用）
    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt msg) {
        String orderId = msg.getKeys();

        try {
            // 查询本地数据库，确认订单状态
            Order order = orderService.getOrderById(orderId);

            if (order == null) {
                // 订单不存在，说明本地事务失败了（被回滚了）
                return LocalTransactionState.ROLLBACK_MESSAGE;
            }

            if ("PAID".equals(order.getStatus())) {
                // 订单已支付，说明本地事务成功了
                return LocalTransactionState.COMMIT_MESSAGE;
            }

            if ("PENDING".equals(order.getStatus())) {
                // 订单还在处理中，可能是临时状态，让MQ继续等待
                return LocalTransactionState.UNKNOW;
            }

            // 其他状态，保守处理，回滚
            return LocalTransactionState.ROLLBACK_MESSAGE;

        } catch (Exception e) {
            // 查询失败，返回UNKNOWN，下次再查
            return LocalTransactionState.UNKNOW;
        }
    }
}

// 创建事务消息Producer
TransactionMQProducer producer = new TransactionMQProducer("order-tx-producer");
producer.setNamesrvAddr("name1:9876;name2:9876");
producer.setTransactionListener(new OrderTransactionListener());
producer.start();

// 发送事务消息
Order order = new Order();
order.setId(UUID.randomUUID().toString());
order.setUserId("user123");
order.setAmount(100.0);

Message msg = new Message("order-topic", "order-tag", order.getId().getBytes(),
    new ObjectMapper().writeValueAsBytes(order));

// 发送时会先发半消息，然后回调executeLocalTransaction
SendResult sendResult = producer.sendMessageInTransaction(msg, order);

if (sendResult.getSendStatus() == SendStatus.SEND_OK) {
    System.out.println("事务消息发送成功，状态：" + sendResult.getTransactionId());
} else {
    System.out.println("半消息发送失败");
}

// 回查时，如果订单正在处理中，返回COMMIT
@Override
public LocalTransactionState checkLocalTransaction(MessageExt msg) {
    Order order = orderService.getOrderById(msg.getKeys());

    if ("PROCESSING".equals(order.getStatus())) {
        // 还在处理中，说明本地事务最终会成功
        // 返回COMMIT，让消息投递给下游
        // 下游消费时再做最终校验
        return LocalTransactionState.COMMIT_MESSAGE;
    }

    if ("SUCCESS".equals(order.getStatus())) {
        return LocalTransactionState.COMMIT_MESSAGE;
    }

    return LocalTransactionState.ROLLBACK_MESSAGE;
}