MQ（消息队列）核心应用场景与实操指南

// 1. 配置 RabbitMQ 连接（Spring Boot 简化版）
@Configuration
public class RabbitConfig {
    @Bean
    public Queue orderQueue() {
        return new Queue("order.create.queue", true); // 持久化队列
    }
}

// 2. 生产者发送消息
@Service
public class OrderService {
    @Autowired
    private RabbitTemplate rabbitTemplate;

    public String createOrder(Order order) {
        // 1. 保存订单（核心逻辑）
        orderMapper.insert(order);
        // 2. 发送异步消息（无需等待消费端响应）
        rabbitTemplate.convertAndSend("order.create.queue", order.getId());
        // 3. 立即返回结果
        return "订单创建成功，订单号：" + order.getId();
    }
}

@Service
public class StockConsumer {
    // 监听订单创建队列，异步扣减库存
    @RabbitListener(queues = "order.create.queue")
    public void deductStock(String orderId) {
        // 1. 查询订单商品信息
        OrderItem item = orderItemMapper.selectByOrderId(orderId);
        // 2. 扣减库存（异步逻辑，不影响订单创建）
        stockMapper.deduct(item.getProductId(), item.getNum());
        // 3. 记录日志/异常处理（消费失败可重试）
        log.info("订单{}库存扣减完成", orderId);
    }
}

@Service
public class SeckillProducer {
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    public String seckill(String userId, String productId) {
        // 1. 简单校验（如用户是否已参与）
        if (seckillMapper.checkUser(userId, productId)) {
            return "您已参与秒杀";
        }
        // 2. 发送秒杀请求到 Kafka（立即返回，不阻塞）
        String message = userId + "," + productId;
        kafkaTemplate.send("seckill.topic", message);
        return "秒杀请求已提交，请等待结果";
    }
}

@Service
public class SeckillConsumer {
    // 配置消费线程数和速率（Spring Kafka 配置）
    @Value("${kafka.consumer.threads:5}")
    private int consumerThreads;
    @Value("${kafka.consumer.rate:100}") // 每秒消费 100 条
    private int consumeRate;

    @KafkaListener(topics = "seckill.topic", concurrency = "${kafka.consumer.threads}")
    public void consume(String message) throws InterruptedException {
        // 1. 按速率消费（控制每秒处理条数）
        Thread.sleep(1000 / consumeRate);
        // 2. 处理秒杀逻辑（扣库存、创建订单）
        String[] params = message.split(",");
        seckillService.process(params[0], params[1]);
    }
}

@Service
public class PayProducer {
    @Autowired
    private RabbitTemplate rabbitTemplate;

    public void sendPayMessage(PayDTO payDTO) {
        // 1. 构建通用消息格式
        Map<String, Object> msg = new HashMap<>();
        msg.put("msgId", UUID.randomUUID().toString());
        msg.put("msgType", "ORDER_PAY");
        msg.put("data", payDTO);
        msg.put("timestamp", System.currentTimeMillis());
        // 2. 发送到交换机（路由到多个消费端）
        rabbitTemplate.convertAndSend("order.exchange", "order.pay", msg);
    }
}

@Service
public class TransactionProducer {
    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    // 1. 执行本地事务 + 发送半消息
    public void transfer(String fromAccount, String toAccount, BigDecimal amount) {
        // 构建事务消息
        Message<String> message = MessageBuilder.withPayload(JSON.toJSONString(new TransferDTO(fromAccount, toAccount, amount)))
                .setHeader(RocketMQHeaders.TRANSACTION_ID, UUID.randomUUID().toString())
                .build();
        // 发送半消息，触发本地事务
        rocketMQTemplate.sendMessageInTransaction("transfer.topic", message, null);
    }

    // 2. 本地事务执行逻辑
    @RocketMQTransactionListener
    class TransferTransactionListener implements RocketMQLocalTransactionListener {
        @Override
        public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg) {
            TransferDTO dto = JSON.parseObject(msg.getPayload().toString(), TransferDTO.class);
            try {
                // 本地事务：扣减A账户余额
                accountMapper.deduct(dto.getFromAccount(), dto.getAmount());
                return RocketMQLocalTransactionState.COMMIT; // 提交消息，允许消费
            } catch (Exception e) {
                return RocketMQLocalTransactionState.ROLLBACK; // 回滚消息，不发送
            }
        }

        // 3. 事务回查（解决生产者宕机导致的状态未知）
        @Override
        public RocketMQLocalTransactionState checkLocalTransaction(Message msg) {
            TransferDTO dto = JSON.parseObject(msg.getPayload().toString(), TransferDTO.class);
            // 检查A账户扣款是否成功
            boolean success = accountMapper.checkDeduct(dto.getFromAccount(), dto.getAmount());
            return success ? RocketMQLocalTransactionState.COMMIT : RocketMQLocalTransactionState.ROLLBACK;
        }
    }
}

@Service
public class TransferConsumer {
    @RocketMQMessageListener(topic = "transfer.topic", consumerGroup = "transfer-group")
    public class TransferListener implements RocketMQListener<String> {
        @Override
        public void onMessage(String msg) {
            TransferDTO dto = JSON.parseObject(msg, TransferDTO.class);
            // 重试机制：失败则重试，保证最终加款成功
            int retryCount = 3;
            while (retryCount > 0) {
                try {
                    accountMapper.add(dto.getToAccount(), dto.getAmount());
                    log.info("B账户{}加款{}元成功", dto.getToAccount(), dto.getAmount());
                    return;
                } catch (Exception e) {
                    retryCount--;
                    log.error("加款失败，剩余重试次数{}", retryCount);
                    try { Thread.sleep(1000); } catch (InterruptedException ex) {}
                }
            }
            // 重试失败，记录死信队列，人工介入
            log.error("B账户{}加款{}元失败，进入死信队列", dto.getToAccount(), dto.getAmount());
        }
    }
}