RocketMQ中多消息不同状态回查的设计与优化过程

// 发送消息时绑定业务标识
Message msg = new Message("Topic", "Tag", "order123".getBytes());
msg.putUserProperty("bizId", "order123");
msg.putUserProperty("bizType", "order");
sendResult = producer.sendMessageInTransaction(transactionListener, msg, null);

public class MultiStatusTransactionListener implements TransactionListener {
    @Override
    public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        // 1. 解析消息属性
        String bizId = msg.getUserProperty("bizId");
        String bizType = msg.getUserProperty("bizType");
        
        // 2. 根据业务类型执行不同本地事务
        if ("order".equals(bizType)) {
            return orderService.processOrder(bizId);
        } else if ("payment".equals(bizType)) {
            return paymentService.processPayment(bizId);
        }
        return LocalTransactionState.ROLLBACK_MESSAGE;
    }

    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt msg) {
        // 1. 解析消息属性
        String bizId = msg.getUserProperty("bizId");
        String bizType = msg.getUserProperty("bizType");
        
        // 2. 根据业务类型查询不同状态
        if ("order".equals(bizType)) {
            return orderService.checkOrderStatus(bizId);
        } else if ("payment".equals(bizType)) {
            return paymentService.checkPaymentStatus(bizId);
        }
        return LocalTransactionState.ROLLBACK_MESSAGE;
    }
}

// 自定义回查处理器，支持批量处理
public class BatchCheckProcessor {
    // 缓存待回查消息，按业务类型分组
    private final Map<String, List<String>> pendingCheck = new ConcurrentHashMap<>();
    
    // 注册回查消息
    public void registerMessage(String bizType, String bizId) {
        pendingCheck.computeIfAbsent(bizType, k -> new ArrayList<>()).add(bizId);
        // 达到批量阈值或超时后触发批量查询
        if (pendingCheck.get(bizType).size() >= 100 || needBatchCheck()) {
            batchCheckAndClear(bizType);
        }
    }
    
    // 批量查询与状态更新
    private void batchCheckAndClear(String bizType) {
        List<String> bizIds = pendingCheck.remove(bizType);
        if (bizIds == null || bizIds.isEmpty()) return;
        
        // 根据业务类型调用不同批量查询接口
        if ("order".equals(bizType)) {
            Map<String, OrderStatus> statusMap = orderService.batchQueryStatus(bizIds);
            // 批量更新状态并发送响应
            statusMap.forEach((id, status) -> {
                sendCheckResponse(id, mapToTransactionState(status));
            });
        }
        // 其他业务类型处理...
    }
}

<!-- 在 broker 配置文件中设置不同主题的回查策略 -->
<transactionCheckListener>
    <topicCheckConfig>
        <topic>order_topic</topic>
        <checkInterval>5000</checkInterval> <!-- 订单消息5秒回查一次 -->
        <maxCheckTimes>20</maxCheckTimes>
    </topicCheckConfig>
    <topicCheckConfig>
        <topic>payment_topic</topic>
        <checkInterval>10000</checkInterval> <!-- 支付消息10秒回查一次 -->
        <maxCheckTimes>10</maxCheckTimes>
    </topicCheckConfig>
</transactionCheckListener>

// 初始化多业务线程池
private final Map<String, ExecutorService> threadPools = new HashMap<>();
threadPools.put("order", new ThreadPoolExecutor(
    10, 20, 60, TimeUnit.SECONDS, 
    new LinkedBlockingQueue<>(1000), 
    new ThreadFactoryBuilder().setNameFormat("order-check-%d").build()
));
threadPools.put("payment", ...); // 支付业务线程池

private final Map<String, Semaphore> semaphores = new HashMap<>();
semaphores.put("order", new Semaphore(50)); // 订单业务最多50个并发回查

// 回查前获取分布式锁，避免重复处理
boolean locked = redisTemplate.tryLock("check_lock:" + bizId, 3000);
if (locked) {
    try {
        // 执行回查逻辑
    } finally {
        redisTemplate.unlock("check_lock:" + bizId);
    }
}

// 订单状态回查逻辑
public LocalTransactionState checkOrderStatus(String orderId) {
    OrderStatus status = orderDao.getStatus(orderId);
    if (status == SUCCESS) {
        // 订单成功时，主动检查关联的支付状态
        PaymentStatus payStatus = paymentDao.getStatusByOrder(orderId);
        if (payStatus == SUCCESS) {
            return COMMIT_MESSAGE;
        } else {
            // 支付未完成，延迟回查
            return UNKNOW;
        }
    }
    return mapToTransactionState(status);
}