// 發(fā)送消息時綁定業(yè)務(wù)標(biāo)識
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. 根據(jù)業(yè)務(wù)類型執(zhí)行不同本地事務(wù)
        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. 根據(jù)業(yè)務(wù)類型查詢不同狀態(tài)
        if ("order".equals(bizType)) {
            return orderService.checkOrderStatus(bizId);
        } else if ("payment".equals(bizType)) {
            return paymentService.checkPaymentStatus(bizId);
        }
        return LocalTransactionState.ROLLBACK_MESSAGE;
    }
}

// 自定義回查處理器，支持批量處理
public class BatchCheckProcessor {
    // 緩存待回查消息，按業(yè)務(wù)類型分組
    private final Map<String, List<String>> pendingCheck = new ConcurrentHashMap<>();
    
    // 注冊回查消息
    public void registerMessage(String bizType, String bizId) {
        pendingCheck.computeIfAbsent(bizType, k -> new ArrayList<>()).add(bizId);
        // 達到批量閾值或超時后觸發(fā)批量查詢
        if (pendingCheck.get(bizType).size() >= 100 || needBatchCheck()) {
            batchCheckAndClear(bizType);
        }
    }
    
    // 批量查詢與狀態(tài)更新
    private void batchCheckAndClear(String bizType) {
        List<String> bizIds = pendingCheck.remove(bizType);
        if (bizIds == null || bizIds.isEmpty()) return;
        
        // 根據(jù)業(yè)務(wù)類型調(diào)用不同批量查詢接口
        if ("order".equals(bizType)) {
            Map<String, OrderStatus> statusMap = orderService.batchQueryStatus(bizIds);
            // 批量更新狀態(tài)并發(fā)送響應(yīng)
            statusMap.forEach((id, status) -> {
                sendCheckResponse(id, mapToTransactionState(status));
            });
        }
        // 其他業(yè)務(wù)類型處理...
    }
}

<!-- 在 broker 配置文件中設(shè)置不同主題的回查策略 -->
<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>

// 初始化多業(yè)務(wù)線程池
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", ...); // 支付業(yè)務(wù)線程池

private final Map<String, Semaphore> semaphores = new HashMap<>();
semaphores.put("order", new Semaphore(50)); // 訂單業(yè)務(wù)最多50個并發(fā)回查

// 回查前獲取分布式鎖，避免重復(fù)處理
boolean locked = redisTemplate.tryLock("check_lock:" + bizId, 3000);
if (locked) {
    try {
        // 執(zhí)行回查邏輯
    } finally {
        redisTemplate.unlock("check_lock:" + bizId);
    }
}

// 訂單狀態(tài)回查邏輯
public LocalTransactionState checkOrderStatus(String orderId) {
    OrderStatus status = orderDao.getStatus(orderId);
    if (status == SUCCESS) {
        // 訂單成功時，主動檢查關(guān)聯(lián)的支付狀態(tài)
        PaymentStatus payStatus = paymentDao.getStatusByOrder(orderId);
        if (payStatus == SUCCESS) {
            return COMMIT_MESSAGE;
        } else {
            // 支付未完成，延遲回查
            return UNKNOW;
        }
    }
    return mapToTransactionState(status);
}