#RocketMQ 事务消息

// 事务监听器
@Service
public class OrderTransactionListener implements TransactionListener {

    @Autowired
    private OrderService orderService;

    @Autowired
    private OrderMapper orderMapper;

    // 执行本地事务（半消息发送后调用）
    @Override
    public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        String body = new String(msg.getBody());
        OrderMessage orderMessage = JSON.parseObject(body, OrderMessage.class);

        try {
            // 执行本地事务
            Order order = new Order();
            order.setUserId(orderMessage.getUserId());
            order.setAmount(orderMessage.getAmount());
            order.setStatus("CREATED");
            orderMapper.insert(order);

            // 本地事务成功，提交消息
            return LocalTransactionState.COMMIT_MESSAGE;

        } catch (Exception e) {
            // 本地事务失败，回滚消息
            log.error("本地事务执行失败", e);
            return LocalTransactionState.ROLLBACK_MESSAGE;
        }
    }

    // 回查（长时间未收到提交/回滚指令时调用）
    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt msg) {
        String body = new String(msg.getBody());
        OrderMessage orderMessage = JSON.parseObject(body, OrderMessage.class);

        // 查询本地事务状态
        Order order = orderMapper.findByUserId(orderMessage.getUserId());
        if (order != null) {
            // 本地事务已成功
            return LocalTransactionState.COMMIT_MESSAGE;
        } else {
            // 本地事务未执行（或失败了）
            return LocalTransactionState.ROLLBACK_MESSAGE;
        }
    }
}

// 发送事务消息
@Service
public class OrderService {

    @Autowired
    private TransactionMQProducer producer;

    public void createOrder(OrderMessage orderMessage) {
        Message msg = new Message("order-topic", JSON.toJSONString(orderMessage).getBytes());

        // 发送事务消息
        // 第二个参数是 arg，会传递给 TransactionListener.executeLocalTransaction
        producer.sendMessageInTransaction(msg, orderMessage);
    }
}