-- 訂單服務(wù)：本地消息表（含對賬字段）
CREATE TABLE message_log (
    id BIGINT PRIMARY KEY AUTO_INCREMENT,
    order_id BIGINT NOT NULL COMMENT '訂單ID（關(guān)聯(lián)t_order表，唯一）',
    rocketmq_msg_id VARCHAR(64) DEFAULT NULL COMMENT 'RocketMQ消息唯一ID（關(guān)聯(lián)消息中間件）',
    message_content TEXT NOT NULL COMMENT '消息內(nèi)容（JSON格式，含orderId、skuId、quantity等）',
    business_type VARCHAR(32) NOT NULL COMMENT '業(yè)務(wù)類型：ORDER_CREATE（創(chuàng)建訂單）、INVENTORY_DEDUCT（扣庫存）、NOTICE_SEND（發(fā)通知）',
    msg_status ENUM('INIT','SENT','CONSUMED','FAIL') DEFAULT 'INIT' COMMENT '消息狀態(tài)：INIT=初始，SENT=已投遞，CONSUMED=已消費，F(xiàn)AIL=失敗',
    reconcile_status ENUM('PENDING','SUCCESS','FAIL','RETRY') DEFAULT 'PENDING' COMMENT '對賬狀態(tài)：PENDING=待對賬，SUCCESS=對賬一致，F(xiàn)AIL=對賬不一致，RETRY=待重試',
    retry_count TINYINT DEFAULT 0 COMMENT '對賬重試次數(shù)（最大5次）',
    next_reconcile_time DATETIME NOT NULL COMMENT '下次對賬時間（定時任務(wù)篩選依據(jù)）',
    create_time DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '消息創(chuàng)建時間',
    update_time DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '狀態(tài)更新時間',
    UNIQUE KEY uk_order_id_business_type (order_id, business_type) COMMENT '避免同一訂單同一業(yè)務(wù)類型重復(fù)發(fā)消息'
);
-- 訂單服務(wù)：訂單表（簡化，僅保留核心字段）
CREATE TABLE t_order (
    id BIGINT PRIMARY KEY AUTO_INCREMENT,
    order_id VARCHAR(64) NOT NULL COMMENT '訂單唯一編號',
    user_id BIGINT NOT NULL COMMENT '用戶ID',
    sku_id BIGINT NOT NULL COMMENT '商品SKU ID',
    quantity INT NOT NULL COMMENT '購買數(shù)量',
    order_status ENUM('CREATED','PAID','SHIPPED','FINISHED','CANCELED') DEFAULT 'CREATED' COMMENT '訂單狀態(tài)',
    create_time DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '訂單創(chuàng)建時間',
    UNIQUE KEY uk_order_id (order_id)
);
-- 庫存服務(wù)：庫存扣減記錄表（支撐狀態(tài)查詢與冪等）
CREATE TABLE inventory_deduct_log (
    id BIGINT PRIMARY KEY AUTO_INCREMENT,
    order_id VARCHAR(64) NOT NULL COMMENT '訂單ID（關(guān)聯(lián)訂單服務(wù)）',
    sku_id BIGINT NOT NULL COMMENT '商品SKU ID',
    deduct_quantity INT NOT NULL COMMENT '扣減數(shù)量',
    deduct_status ENUM('SUCCESS','FAIL','PROCESSING') DEFAULT 'PROCESSING' COMMENT '扣減狀態(tài)',
    create_time DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
    UNIQUE KEY uk_order_id (order_id) COMMENT '按訂單ID冪等，避免重復(fù)扣減'
);

import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import javax.annotation.Resource;
import java.nio.charset.StandardCharsets;
import java.util.Date;
import java.util.concurrent.TimeUnit;
@Service
public class OrderServiceImpl implements OrderService {
    // 注入RocketMQ事務(wù)生產(chǎn)者（單例，由Spring容器初始化）
    @Resource
    private TransactionMQProducer transactionMQProducer;
    @Resource
    private OrderMapper orderMapper;
    @Resource
    private MessageLogMapper messageLogMapper;
    /
     * 創(chuàng)建訂單 + 發(fā)送事務(wù)消息
     */
    @Override
    public void createOrder(OrderCreateDTO dto) throws Exception {
        // 1. 構(gòu)造訂單數(shù)據(jù)（生成唯一訂單號）
        String orderId = generateOrderId();
        TOrder order = TOrder.builder()
                .orderId(orderId)
                .userId(dto.getUserId())
                .skuId(dto.getSkuId())
                .quantity(dto.getQuantity())
                .orderStatus("CREATED")
                .build();
        // 2. 構(gòu)造RocketMQ事務(wù)消息（主題：OrderTopic，標(biāo)簽：INVENTORY_DEDUCT+NOTICE_SEND，支持多下游訂閱）
        String msgContent = JSON.toJSONString(dto);
        Message message = new Message(
                "OrderTopic",  // 主題：下游服務(wù)訂閱此主題
                "INVENTORY_DEDUCT||NOTICE_SEND",  // 標(biāo)簽：區(qū)分業(yè)務(wù)類型，下游可按標(biāo)簽過濾
                orderId.getBytes(StandardCharsets.UTF_8),  // 消息Key：訂單ID，便于定位
                msgContent.getBytes(StandardCharsets.UTF_8)
        );
        // 3. 發(fā)送事務(wù)消息（將訂單數(shù)據(jù)作為參數(shù)透傳給事務(wù)監(jiān)聽器）
        transactionMQProducer.sendMessageInTransaction(message, order);
    }
    /
     * 事務(wù)監(jiān)聽器：執(zhí)行本地事務(wù) + 事務(wù)回查
/
    @Resource
    private TransactionListener orderTransactionListener;
    // 初始化事務(wù)生產(chǎn)者時綁定監(jiān)聽器（Spring Bean初始化方法）
    @PostConstruct
    public void initProducer() {
        transactionMQProducer.setTransactionListener(orderTransactionListener);
    }
    /
     * 本地事務(wù)邏輯（由監(jiān)聽器調(diào)用，確保訂單與消息表同成功/同失?。?/
    @Transactional(rollbackFor = Exception.class)
    public LocalTransactionState executeLocalTransaction(TOrder order, Message message) {
        try {
            // 步驟1：保存訂單到訂單表
            orderMapper.insert(order);
            // 步驟2：記錄本地消息表（對賬狀態(tài)初始為PENDING，下次對賬時間5分鐘后）
            Date nextReconcileTime = new Date(System.currentTimeMillis() + TimeUnit.MINUTES.toMillis(5));
            MessageLog log = MessageLog.builder()
                    .orderId(order.getOrderId())
                    .rocketmqMsgId(message.getMsgId())
                    .messageContent(new String(message.getBody()))
                    .businessType("ORDER_CREATE")
                    .msgStatus("INIT")
                    .reconcileStatus("PENDING")
                    .retryCount(0)
                    .nextReconcileTime(nextReconcileTime)
                    .build();
            messageLogMapper.insert(log);
            // 步驟3：提交本地事務(wù)，返回COMMIT（通知RocketMQ投遞消息）
            return LocalTransactionState.COMMIT_MESSAGE;
        } catch (Exception e) {
            // 本地事務(wù)失敗，回滾，返回ROLLBACK（通知RocketMQ丟棄消息）
            return LocalTransactionState.ROLLBACK_MESSAGE;
        }
    }
    /*
     * 事務(wù)回查邏輯（Broker未收到Commit/Rollback時觸發(fā)）
/
    public LocalTransactionState checkLocalTransaction(String orderId) {
        // 查本地消息表，按訂單ID判斷本地事務(wù)狀態(tài)
        MessageLog log = messageLogMapper.selectByOrderId(orderId);
        if (log == null) {
            return LocalTransactionState.ROLLBACK_MESSAGE; // 本地?zé)o記錄，回滾
        }
        // 本地消息已記錄，說明本地事務(wù)成功，返回COMMIT
        if ("INIT".equals(log.getMsgStatus()) || "PENDING".equals(log.getReconcileStatus())) {
            return LocalTransactionState.COMMIT_MESSAGE;
        }
        return LocalTransactionState.ROLLBACK_MESSAGE;
    }
    // 工具方法：生成唯一訂單號
    private String generateOrderId() {
        return "ORDER_" + System.currentTimeMillis() + RandomUtils.nextInt(1000, 9999);
    }
}

import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;
import javax.annotation.Resource;
@Component
// 訂閱訂單主題，僅消費“扣庫存”標(biāo)簽的消息
@RocketMQMessageListener(topic = "OrderTopic", selectorExpression = "INVENTORY_DEDUCT", consumerGroup = "inventory_consumer_group")
public class InventoryConsumer implements RocketMQListener<MessageExt> {
    @Resource
    private InventoryMapper inventoryMapper;
    @Resource
    private InventoryDeductLogMapper deductLogMapper;
    @Resource
    private MessageLogFeignClient messageLogFeignClient; // 調(diào)用訂單服務(wù)的消息表接口
    @Override
    @Transactional(rollbackFor = Exception.class)
    public void onMessage(MessageExt messageExt) {
        // 1. 解析消息（獲取訂單ID、商品ID、扣減數(shù)量）
        String msgContent = new String(messageExt.getBody());
        OrderCreateDTO dto = JSON.parseObject(msgContent, OrderCreateDTO.class);
        String orderId = dto.getOrderId();
        Long skuId = dto.getSkuId();
        Integer quantity = dto.getQuantity();
        // 2. 冪等控制：查詢是否已扣減（按訂單ID）
        InventoryDeductLog existLog = deductLogMapper.selectByOrderId(orderId);
        if (existLog != null && "SUCCESS".equals(existLog.getDeductStatus())) {
            // 已成功扣減，直接返回成功
            messageLogFeignClient.updateMsgStatus(orderId, "CONSUMED"); // 通知訂單服務(wù)更新消息狀態(tài)
            return;
        }
        try {
            // 3. 執(zhí)行庫存扣減（先查庫存是否充足）
            Inventory inventory = inventoryMapper.selectBySkuId(skuId);
            if (inventory == null || inventory.getStock() < quantity) {
                // 庫存不足，記錄失敗狀態(tài)，返回失敗（不重試，避免無效循環(huán)）
                deductLogMapper.insert(InventoryDeductLog.builder()
                        .orderId(orderId)
                        .skuId(skuId)
                        .deductQuantity(quantity)
                        .deductStatus("FAIL")
                        .build());
                messageLogFeignClient.updateMsgStatus(orderId, "FAIL"); // 通知訂單服務(wù)更新消息狀態(tài)
                throw new RuntimeException("庫存不足，扣減失敗");
            }
            // 4. 扣減庫存并記錄日志
            inventory.setStock(inventory.getStock() - quantity);
            inventoryMapper.updateById(inventory);
            deductLogMapper.insert(InventoryDeductLog.builder()
                    .orderId(orderId)
                    .skuId(skuId)
                    .deductQuantity(quantity)
                    .deductStatus("SUCCESS")
                    .build());
            // 5. 通知訂單服務(wù)更新消息狀態(tài)為“已消費”
            messageLogFeignClient.updateMsgStatus(orderId, "CONSUMED");
            // 返回消費成功
        } catch (Exception e) {
            // 消費失敗，記錄“處理中”狀態(tài)，返回重試
            deductLogMapper.insertOrUpdate(InventoryDeductLog.builder()
                    .orderId(orderId)
                    .skuId(skuId)
                    .deductQuantity(quantity)
                    .deductStatus("PROCESSING")
                    .build());
            return ConsumeConcurrentlyStatus.RECONSUME_LATER;
        }
    }
    /
     * 對外提供庫存扣減狀態(tài)查詢接口（供XXL-Job對賬調(diào)用）
     */
    public String queryDeductStatus(String orderId) {
        InventoryDeductLog log = deductLogMapper.selectByOrderId(orderId);
        if (log == null) {
            return "NOT_PROCESSED"; // 未處理
        }
        return log.getDeductStatus(); // SUCCESS/FAIL/PROCESSING
    }
    /
     * 對外提供庫存扣減重試接口（供XXL-Job對賬修復(fù)調(diào)用）
     */
    public boolean retryDeduct(String orderId) {
        // 邏輯同onMessage，僅針對“PROCESSING/FAIL”狀態(tài)的記錄重試，此處省略
        return true;
    }
}

import com.xxl.job.core.context.XxlJobHelper;
import com.xxl.job.core.handler.annotation.XxlJob;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;
import java.util.Date;
import java.util.List;
import java.util.concurrent.TimeUnit;
@Component
public class OrderReconcileJob {
    @Resource
    private MessageLogMapper messageLogMapper;
    @Resource
    private OrderService orderService;
    @Resource
    private InventoryFeignClient inventoryFeignClient; // 調(diào)用庫存服務(wù)接口
    @Resource
    private NoticeFeignClient noticeFeignClient; // 調(diào)用通知服務(wù)接口
    @Resource
    private ReconcileWorkOrderMapper workOrderMapper; // 對賬工單表
    /*
     * XXL-Job 對賬任務(wù)（分片執(zhí)行）
/
    @XxlJob("orderReconcileJob")
    public void execute() throws Exception {
        // 1. 獲取分片參數(shù)（當(dāng)前分片號、總分片數(shù)）
        int shardIndex = XxlJobHelper.getShardIndex();
        int shardTotal = XxlJobHelper.getShardTotal();
        // 2. 篩選待對賬數(shù)據(jù)：
        // - 對賬狀態(tài)為PENDING/RETRY
        // - 下次對賬時間 <= 當(dāng)前時間
        // - 重試次數(shù) < 5次
        // - 按order_id尾號分片查詢（避免重復(fù)）
        List<MessageLog> pendingLogs = messageLogMapper.selectPendingReconcile(
                shardIndex, shardTotal, 5, new Date()
        );
        if (pendingLogs.isEmpty()) {
            XxlJobHelper.log("當(dāng)前分片無待對賬數(shù)據(jù)，分片號：{}", shardIndex);
            return;
        }
        // 3. 遍歷待對賬記錄，逐一對賬
        for (MessageLog log : pendingLogs) {
            String orderId = log.getOrderId();
            try {
                XxlJobHelper.log("開始對賬訂單：{}", orderId);
                // 步驟1：查詢各服務(wù)狀態(tài)
                // - 訂單服務(wù)狀態(tài)：是否已創(chuàng)建（CREATED）
                String orderStatus = orderService.queryOrderStatus(orderId);
                // - 庫存服務(wù)狀態(tài)：是否已扣減（SUCCESS/FAIL/PROCESSING）
                String inventoryStatus = inventoryFeignClient.queryDeductStatus(orderId);
                // - 通知服務(wù)狀態(tài)：是否已發(fā)送（SUCCESS/FAIL/PROCESSING）
                String noticeStatus = noticeFeignClient.queryNoticeStatus(orderId);
                // 步驟2：狀態(tài)一致性判斷與處理
                // 場景1：訂單已創(chuàng)建，庫存/通知均成功 → 對賬一致
                if ("CREATED".equals(orderStatus) 
                        && "SUCCESS".equals(inventoryStatus) 
                        && "SUCCESS".equals(noticeStatus)) {
                    messageLogMapper.updateReconcileStatus(orderId, "SUCCESS");
                    XxlJobHelper.log("訂單{}對賬一致，狀態(tài)更新為SUCCESS", orderId);
                    continue;
                }
                // 場景2：訂單已創(chuàng)建，庫存/通知存在PROCESSING → 待重試（下次對賬再查）
                if ("CREATED".equals(orderStatus) 
                        && ("PROCESSING".equals(inventoryStatus) || "PROCESSING".equals(noticeStatus))) {
                    // 更新下次對賬時間（10分鐘后）和重試次數(shù)
                    Date nextTime = new Date(System.currentTimeMillis() + TimeUnit.MINUTES.toMillis(10));
                    messageLogMapper.updateRetryInfo(orderId, log.getRetryCount() + 1, nextTime, "RETRY");
                    XxlJobHelper.log("訂單{}存在處理中狀態(tài)，下次對賬時間：{}", orderId, nextTime);
                    continue;
                }
                // 場景3：訂單已創(chuàng)建，庫存/通知存在FAIL → 觸發(fā)自動重試
                if ("CREATED".equals(orderStatus) 
                        && ("FAIL".equals(inventoryStatus) || "FAIL".equals(noticeStatus))) {
                    // 重試次數(shù)未超5次，調(diào)用下游重試接口
                    if (log.getRetryCount() < 5) {
                        boolean inventoryRetry = "FAIL".equals(inventoryStatus) 
                                ? inventoryFeignClient.retryDeduct(orderId) : true;
                        boolean noticeRetry = "FAIL".equals(noticeStatus) 
                                ? noticeFeignClient.retrySend(orderId) : true;
                        if (inventoryRetry && noticeRetry) {
                            // 重試成功，更新下次對賬時間（5分鐘后）
                            Date nextTime = new Date(System.currentTimeMillis() + TimeUnit.MINUTES.toMillis(5));
                            messageLogMapper.updateRetryInfo(orderId, log.getRetryCount() + 1, nextTime, "RETRY");
                            XxlJobHelper.log("訂單{}重試下游服務(wù)成功，下次對賬時間：{}", orderId, nextTime);
                        } else {
                            // 重試失敗，更新重試次數(shù)，下次繼續(xù)
                            Date nextTime = new Date(System.currentTimeMillis() + TimeUnit.MINUTES.toMillis(30));
                            messageLogMapper.updateRetryInfo(orderId, log.getRetryCount() + 1, nextTime, "RETRY");
                            XxlJobHelper.log("訂單{}重試下游服務(wù)失敗，下次對賬時間：{}", orderId, nextTime);
                        }
                        continue;
                    }
                    // 重試次數(shù)超5次，生成人工工單
                    workOrderMapper.insert(ReconcileWorkOrder.builder()
                            .orderId(orderId)
                            .workOrderNo("RECONCILE_" + System.currentTimeMillis())
                            .faultDesc("訂單" + orderId + "：庫存狀態(tài)=" + inventoryStatus + "，通知狀態(tài)=" + noticeStatus + "，重試5次失敗")
                            .workOrderStatus("PENDING")
                            .createTime(new Date())
                            .build());
                    // 更新對賬狀態(tài)為FAIL
                    messageLogMapper.updateReconcileStatus(orderId, "FAIL");
                    XxlJobHelper.log("訂單{}對賬失敗，生成人工工單", orderId);
                    continue;
                }
                // 場景4：訂單狀態(tài)異常（如CANCELED）→ 對賬一致（無需下游處理）
                if ("CANCELED".equals(orderStatus)) {
                    messageLogMapper.updateReconcileStatus(orderId, "SUCCESS");
                    XxlJobHelper.log("訂單{}已取消，對賬狀態(tài)更新為SUCCESS", orderId);
                    continue;
                }
            } catch (Exception e) {
                // 對賬過程異常（如接口超時），標(biāo)記為RETRY，下次再查
                Date nextTime = new Date(System.currentTimeMillis() + TimeUnit.MINUTES.toMillis(10));
                messageLogMapper.updateRetryInfo(orderId, log.getRetryCount() + 1, nextTime, "RETRY");
                XxlJobHelper.log("訂單{}對賬異常，原因：{}，下次對賬時間：{}", orderId, e.getMessage(), nextTime);
            }
        }
        XxlJobHelper.handleSuccess("當(dāng)前分片對賬完成，分片號：{}，處理記錄數(shù)：{}", shardIndex, pendingLogs.size());
    }
}