酒类商城，防止超卖的尝试系列（第四次）

CREATE TABLE `local_message_consume_log` (
    `id` BIGINT NOT NULL AUTO_INCREMENT,
    `message_id` VARCHAR(64) NOT NULL COMMENT '消息唯一标识',
    `biz_id` VARCHAR(64) NOT NULL COMMENT '业务ID（如订单号）',
    `create_time` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
    PRIMARY KEY (`id`),
    UNIQUE KEY `uk_message_id` (`message_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='消息消费日志表（幂等表）';

/**
*
* 生产者
* 这里只负责插入消息表，不直接发送 MQ。发送逻辑交给后台 Job 统一处理，保证"发送"这个动作也是可重试、可追溯的。
*/
@Service
@Slf4j
public class OrderService {

    @Autowired
    private RedisService redisService;

    @Autowired
    private OrderMapper orderMapper;

    @Autowired
    private LocalMessageMapper localMessageMapper;

    public CreateOrderResult createOrder(Long productId, Integer buyCount, Long userId) {

        // ========== 第1步：Redis 预扣减 ==========
        boolean deducted = redisService.deductStock(productId, buyCount);
        if (!deducted) {
            throw new InsufficientStockException("库存不足");
        }

        // ========== 第2步：本地事务（创建订单 + 插消息表）==========
        // ⚠️ 注意：这里不扣 DB 库存！DB 库存由 MQ 消费者异步扣减
        String messageId;
        Long orderId;
        try {
            orderId = saveOrderAndMessage(productId, buyCount, userId);
            messageId = UUID.randomUUID().toString();
            log.info("本地事务提交成功，orderId={}", orderId);
        } catch (Exception e) {
            // 本地事务失败 → 回滚 Redis
            redisService.rollbackStock(productId, buyCount);
            throw new RuntimeException("系统繁忙，请重试");
        }

        // ========== 第3步：返回用户成功 ==========
        // ⚠️ 此时 DB 库存尚未扣减，订单状态为"待确认"
        // 用户能看到订单，但支付前需要等待库存确认

        // ========== 第4步：尝试发送 MQ（异步，不阻塞主流程）==========
        try {
            OrderMessage msg = buildOrderMessage(orderId, productId, buyCount, userId);
            mqProducer.send("order-topic", msg);
            localMessageMapper.updateStatus(messageId, MessageStatus.SENT.getCode());
            log.info("MQ 发送成功，messageId={}", messageId);
        } catch (Exception e) {
            // 发送失败 → 消息表保持"待发送"，后台 Job 补偿
            log.warn("MQ 发送失败，等待 Job 补偿，messageId={}", messageId, e);
        }

        return CreateOrderResult.success(orderId);
    }

    @Transactional(rollbackFor = Exception.class)
    public Long saveOrderAndMessage(Long productId, Integer buyCount, Long userId) {
        // 1. 创建订单（状态：待确认，库存尚未扣减）
        Order order = new Order();
        order.setProductId(productId);
        order.setBuyCount(buyCount);
        order.setUserId(userId);
        order.setStatus(OrderStatus.PENDING_CONFIRM.getCode()); // 待确认
        order.setRedisDeducted(true); // 标记 Redis 已扣
        orderMapper.insert(order);

        // 2. 插入本地消息表（状态：待发送）
        LocalMessage message = new LocalMessage();
        message.setMessageId(UUID.randomUUID().toString());
        message.setBizId(String.valueOf(order.getId()));
        message.setTopic("order-topic");
        message.setBody(JSON.toJSONString(buildOrderMessage(order.getId(), productId, buyCount, userId)));
        message.setStatus(MessageStatus.PENDING.getCode());
        message.setRetryCount(0);
        message.setMaxRetryCount(5);
        message.setNextRetryTime(new Date());
        localMessageMapper.insert(message);

        return order.getId();
    }
}


/**
*
* 后台补偿job
*/
@Component
@Slf4j
public class LocalMessageJob {

    @Autowired
    private LocalMessageMapper localMessageMapper;

    @Autowired
    private MqProducer mqProducer;

    /**
     * 定时扫描待发送消息（每 5 秒执行一次）
     */
    @Scheduled(fixedDelay = 5000)
    public void processPendingMessages() {
        // 查询状态为"待发送"或"发送失败"，且到达重试时间的消息
        // 每次取 100 条，避免单次处理太多
        List<LocalMessage> messages = localMessageMapper.selectPendingMessages(
            MessageStatus.PENDING.getCode(),
            MessageStatus.FAILED.getCode(),
            new Date(),
            100
        );

        if (messages.isEmpty()) {
            return;
        }

        log.info("扫描到 {} 条待发送消息", messages.size());

        for (LocalMessage message : messages) {
            processSingleMessage(message);
        }
    }

    /**
     * 处理单条消息
     */
    private void processSingleMessage(LocalMessage message) {
        try {
            // 1. 发送 MQ
            mqProducer.send(
                message.getTopic(),
                message.getTag(),
                message.getBody()
            );

            // 2. 发送成功 → 更新状态为"已发送"
            localMessageMapper.updateStatus(
                message.getId(),
                MessageStatus.SENT.getCode(),
                message.getRetryCount() + 1
            );
            log.info("消息发送成功，messageId={}", message.getMessageId());

        } catch (Exception e) {
            log.error("消息发送失败，messageId={}, 当前重试次数={}",
                message.getMessageId(), message.getRetryCount(), e);

            // 3. 发送失败 → 更新重试次数 + 计算下次重试时间（指数退避）
            int newRetryCount = message.getRetryCount() + 1;
            int maxRetryCount = message.getMaxRetryCount();

            if (newRetryCount >= maxRetryCount) {
                // 超过最大重试次数 → 状态改为"最终失败"，触发告警
                localMessageMapper.updateStatus(
                    message.getId(),
                    MessageStatus.FINAL_FAILED.getCode(),
                    newRetryCount
                );
                // 发送告警通知（钉钉/邮件/电话）
                sendAlert(message);
                log.error("消息重试次数超过上限，进入最终失败状态，messageId={}", message.getMessageId());
            } else {
                // 计算下次重试时间：指数退避 2^n 秒
                long delaySeconds = (long) Math.pow(2, newRetryCount);
                Date nextRetryTime = new Date(System.currentTimeMillis() + delaySeconds * 1000);
                localMessageMapper.updateStatusAndRetryTime(
                    message.getId(),
                    MessageStatus.FAILED.getCode(),
                    newRetryCount,
                    nextRetryTime
                );
                log.warn("消息发送失败，将在 {} 秒后重试，messageId={}", delaySeconds, message.getMessageId());
            }
        }
    }
}


/**
*
* 消费者端
* 幂等处理
* 在消息队列场景中，重复消费是常态（网络抖动、消费者重启、ACK 超时等都可能触发重发）
* 因此，消费者必须实现幂等。
*/
@Component
@RocketMQMessageListener(topic = "order-topic", consumerGroup = "order-consumer-group")
@Slf4j
public class OrderConsumer implements RocketMQListener<OrderMessage> {

    @Autowired
    private ProductMapper productMapper;

    @Autowired
    private OrderMapper orderMapper;

    @Autowired
    private LocalMessageConsumeLogMapper consumeLogMapper;

    @Override
    public void onMessage(OrderMessage msg) {
        String messageId = msg.getMessageId();
        Long orderId = msg.getOrderId();

        // 1. 幂等判断
        if (consumeLogMapper.existsByMessageId(messageId)) {
            log.info("消息已消费过，幂等跳过，messageId={}", messageId);
            return;
        }

        try {
            // 2. DB 扣减库存（乐观锁兜底）—— 这才是真正的库存扣减！
            int rows = productMapper.decreaseStock(msg.getProductId(), msg.getBuyCount());
            if (rows == 0) {
                // DB 库存不足 → Redis 和 DB 数据不一致
                // 重试无法解决，直接告警 + 人工介入
                log.error("DB 库存不足！productId={}, orderId={}", msg.getProductId(), orderId);
                // 不抛异常（抛了也解决不了），让消息进入死信队列
                return;
            }

            // 3. 更新订单状态（待确认 → 已确认/待发货）
            orderMapper.updateStatus(orderId, OrderStatus.CONFIRMED.getCode());

            // 4. 记录消费日志（幂等）
            consumeLogMapper.insert(messageId, orderId);

            log.info("DB 库存扣减成功，orderId={}", orderId);

        } catch (Exception e) {
            // 可恢复异常（DB 连接超时等）→ 抛出异常，触发 MQ 重试
            log.warn("消费者处理失败，将自动重试，orderId={}", orderId, e);
            throw e;
        }
    }
}