public interface TransactionListener {
    // 执行本地事务
    LocalTransactionState executeLocalTransaction(Message msg, Object arg);
    
    // 事务回查
    LocalTransactionState checkLocalTransaction(MessageExt msg);
}

@Component
@RocketMQTransactionListener(txProducerGroup = "order-transaction-check-group")
public class OrderTransactionListener implements TransactionListener {
    
    @Autowired
    private OrderService orderService;
    
    @Autowired
    private InventoryService inventoryService;
    
    /**
     * 执行本地事务
     * @param message 消息对象
     * @param o 业务参数
     * @return 事务状态
     */
    @Override
    public LocalTransactionState executeLocalTransaction(Message message, Object o) {
        try {
            // 解析消息体
            String body = new String(message.getBody(), StandardCharsets.UTF_8);
            OrderDTO orderDTO = JSON.parseObject(body, OrderDTO.class);
            
            // 开启本地事务
            TransactionStatus status = orderService.createOrderWithTransaction(orderDTO);
            
            // 根据事务状态返回
            if (status.isCompleted() && status.isSuccess()) {
                return LocalTransactionState.COMMIT_MESSAGE;
            } else {
                return LocalTransactionState.ROLLBACK_MESSAGE;
            }
        } catch (Exception e) {
            log.error("执行本地事务失败", e);
            return LocalTransactionState.ROLLBACK_MESSAGE;
        }
    }
    
    /**
     * 事务回查
     * @param messageExt 消息对象
     * @return 事务状态
     */
    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {
        try {
            String body = new String(messageExt.getBody(), StandardCharsets.UTF_8);
            OrderDTO orderDTO = JSON.parseObject(body, OrderDTO.class);
            
            // 查询本地事务状态
            boolean isOrderCreated = orderService.checkOrderExists(orderDTO.getOrderId());
            
            if (isOrderCreated) {
                log.info("事务回查：订单已创建，提交消息");
                return LocalTransactionState.COMMIT_MESSAGE;
            } else {
                log.warn("事务回查：订单不存在，回滚消息");
                return LocalTransactionState.ROLLBACK_MESSAGE;
            }
        } catch (Exception e) {
            log.error("事务回查失败", e);
            return LocalTransactionState.ROLLBACK_MESSAGE;
        }
    }
}

@Service
public class OrderServiceImpl implements OrderService {
    
    @Autowired
    private RocketMQTemplate rocketMQTemplate;
    
    @Autowired
    private OrderMapper orderMapper;
    
    @Autowired
    private InventoryService inventoryService;
    
    /**
     * 创建订单并发送事务消息
     */
    @Transactional(rollbackFor = Exception.class)
    public TransactionStatus createOrderWithTransaction(OrderDTO orderDTO) {
        try {
            // 1. 创建订单
            OrderPO order = convertToPO(orderDTO);
            orderMapper.insert(order);
            
            // 2. 扣减库存（本地事务内）
            inventoryService.decreaseStock(orderDTO.getProductId(), orderDTO.getCount());
            
            // 3. 构建消息
            MessageBuilder<OrderDTO> builder = MessageBuilder
                .withPayload(orderDTO)
                .setHeader(MessageConst.PROPERTY_TRANSACTION_ID, generateTransactionId());
            
            // 4. 发送事务消息
            rocketMQTemplate.sendMessageInTransaction(
                "order-topic:order-tag",  // Topic 和 Tag
                builder.build(),           // 消息对象
                orderDTO                   // 业务参数（传递给监听器）
            );
            
            return TransactionStatus.success();
            
        } catch (Exception e) {
            log.error("创建订单失败", e);
            return TransactionStatus.failure(e.getMessage());
        }
    }
    
    /**
     * 检查订单是否存在（用于事务回查）
     */
    public boolean checkOrderExists(Long orderId) {
        OrderPO order = orderMapper.selectById(orderId);
        return order != null;
    }
    
    private String generateTransactionId() {
        return UUID.randomUUID().toString().replace("-", "");
    }
}

@Component
@RocketMQMessageListener(
    topic = "order-topic",
    consumerGroup = "inventory-consumer-group",
    messageModel = MessageModel.CLUSTERING
)
public class InventoryConsumer implements RocketMQListener<OrderDTO> {
    
    @Autowired
    private InventoryService inventoryService;
    
    @Autowired
    private PointsService pointsService;
    
    @Override
    public void onMessage(OrderDTO orderDTO) {
        try {
            log.info("收到订单消息，开始处理：{}", orderDTO.getOrderId());
            
            // 1. 增加积分
            pointsService.addPoints(orderDTO.getUserId(), orderDTO.getAmount());
            
            // 2. 发送优惠券
            // ...
            
            // 3. 通知物流系统
            // ...
            
            log.info("订单消息处理成功：{}", orderDTO.getOrderId());
            
        } catch (Exception e) {
            log.error("订单消息处理失败：{}", orderDTO.getOrderId(), e);
            // RocketMQ 会自动重试（默认 16 次）
            throw new RuntimeException("处理失败", e);
        }
    }
}

@Service
public class SimpleOrderService {
    
    @Autowired
    private RocketMQTemplate rocketMQTemplate;
    
    public void createOrder(OrderDTO orderDTO) {
        // 1. 执行本地事务
        orderMapper.insert(convertToPO(orderDTO));
        
        // 2. 发送事务消息
        rocketMQTemplate.sendMessageInTransaction(
            "order-topic",
            MessageBuilder.withPayload(orderDTO).build(),
            (msg, args) -> {
                // 本地事务已提交，返回 COMMIT
                return LocalTransactionState.COMMIT_MESSAGE;
            },
            null
        );
    }
}

@SpringBootTest
public class OrderTransactionTest {
    
    @Autowired
    private OrderService orderService;
    
    @Autowired
    private OrderMapper orderMapper;
    
    @Test
    public void testCreateOrderSuccess() {
        // 准备数据
        OrderDTO orderDTO = new OrderDTO();
        orderDTO.setOrderId(10001L);
        orderDTO.setUserId(888L);
        orderDTO.setAmount(new BigDecimal("199.00"));
        
        // 执行
        orderService.createOrderWithTransaction(orderDTO);
        
        // 验证订单已创建
        OrderPO order = orderMapper.selectById(10001L);
        assertNotNull(order);
        
        // 等待消息投递（异步）
        Thread.sleep(2000);
        
        // 验证库存已扣减、积分已增加等
        // ...
    }
}

// 1. 保持监听器轻量，委托给 Service 处理
@Override
public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
    return orderTransactionService.executeAndReturnState((OrderDTO) arg);
}

// 2. 回查方法必须幂等
@Override
public LocalTransactionState checkLocalTransaction(MessageExt msg) {
    // 查询数据库，不要执行写操作
    return orderTransactionService.checkState(msg);
}

// 3. 记录详细日志，便于排查
log.info("事务执行：orderId={}, status={}", orderDTO.getOrderId(), state);

// 1. 在监听器中执行复杂业务逻辑
@Override
public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
    // 错误：调用远程 RPC、执行复杂计算
    remoteService.call();  // ❌
}

// 2. 回查时修改业务数据
@Override
public LocalTransactionState checkLocalTransaction(MessageExt msg) {
    // 错误：回查时执行业务操作
    inventoryService.decreaseStock();  // ❌
}

// 3. 吞掉异常不处理
try {
    orderService.create();
} catch (Exception e) {
    // 错误：只打印日志不返回 ROLLBACK
    log.error("失败", e);  // ❌
}

// 收集一定数量的订单后批量发送
List<OrderDTO> batch = new ArrayList<>();
for (OrderDTO order : orders) {
    orderService.createOrder(order);
    batch.add(order);
    
    if (batch.size() >= 50) {
        // 批量发送事务消息（需自定义实现）
        transactionSender.sendBatch(batch);
        batch.clear();
    }
}

// 使用线程池异步发送
executorService.submit(() -> {
    rocketMQTemplate.sendMessageInTransaction(...);
});

@RocketMQMessageListener(
    topic = "order-topic",
    consumerGroup = "inventory-consumer-group",
    maxReconsumeTimes = 5  // 最大重试 5 次（默认 16 次）
)

@Override
public void onMessage(OrderDTO orderDTO) {
    try {
        pointsService.addPoints(orderDTO.getUserId(), orderDTO.getAmount());
    } catch (Exception e) {
        // 重试多次后仍然失败，记录到死信表
        if (message.getReconsumeTimes() >= 5) {
            deadLetterTable.save(orderDTO, e.getMessage());
            // 触发人工处理
            alertService.send("积分发放失败，请人工处理");
            return; // 不再重试
        }
        throw new RuntimeException("处理失败", e);
    }
}

// 使用唯一事件 ID 去重
public void onMessage(OrderDTO orderDTO) {
    String eventId = orderDTO.getEventId();
    
    // 检查是否已处理
    if (processedTable.exists(eventId)) {
        log.warn("消息已处理：{}", eventId);
        return;
    }
    
    // 处理业务
    process(orderDTO);
    
    // 记录已处理
    processedTable.insert(eventId);
}