消息队列基础与可靠性设计

订单服务 → 库存服务
         → 积分服务
         → 短信服务
         → 通知服务

订单服务 → MQ → 库存服务
              → 积分服务
              → 短信服务
              → 通知服务

// 传统同步调用 - 强耦合
public void createOrder(Order order) {
    orderDao.save(order);           // 保存订单
    inventoryService.deduct(order); // 调用库存服务
    pointsService.grant(order);     // 调用积分服务
    smsService.send(order);         // 调用短信服务
    // 任何一个服务失败,整个订单创建失败
}

// 使用 MQ 解耦
public void createOrder(Order order) {
    orderDao.save(order);
    mqTemplate.send("order.created", new OrderCreatedEvent(order));
    // 只发送消息,不关心下游如何处理
}

// 同步处理
public Result createOrderSync(Order order) {
    long start = System.currentTimeMillis();
    
    orderService.save(order);           // 50ms
    smsService.sendSms(order);          // 200ms
    couponService.grantCoupon(order);   // 100ms
    pointsService.grantPoints(order);   // 100ms
    
    long cost = System.currentTimeMillis() - start;
    log.info("订单创建耗时: {}ms", cost); // 约 450ms
    return Result.success();
}

// 异步处理
public Result createOrderAsync(Order order) {
    long start = System.currentTimeMillis();
    
    orderService.save(order);  // 50ms
    mqTemplate.send("order.topic", order);  // 5ms
    
    long cost = System.currentTimeMillis() - start;
    log.info("订单创建耗时: {}ms", cost); // 约 55ms
    return Result.success();
}

秒杀开始瞬间 → 10万请求/秒
数据库承受能力 → 1000请求/秒

秒杀请求 → MQ(队列) → 消费者 → 数据库
 10万/秒   缓冲      1000/秒   正常处理

// 秒杀接口 - 接收请求
@PostMapping("/seckill")
public Result seckill(@RequestBody SeckillRequest request) {
    // 快速验证
    if (!seckillService.checkToken(request.getToken())) {
        return Result.fail("无效请求");
    }
    
    // 发送到 MQ,立即返回
    mqTemplate.send("seckill.queue", request);
    return Result.success("排队中,请稍后查询结果");
}

// 秒杀消费者 - 限速处理
@RabbitListener(queues = "seckill.queue", concurrency = "1-1")
public void handleSeckill(SeckillRequest request) {
    // 消费者并发度控制为 1,保证数据库不被打垮
    seckillService.processSeckill(request);
}

// 紧耦合的方式
public void onPaymentSuccess(String orderId) {
    orderService.updateStatus(orderId);
    inventoryService.deduct(orderId);
    pointsService.grant(orderId);
    couponService.use(orderId);
    notificationService.send(orderId);
    // 新增系统需要修改此处代码
}

// 发布事件
public void onPaymentSuccess(String orderId) {
    PaymentSuccessEvent event = new PaymentSuccessEvent(orderId);
    eventPublisher.publish("payment.success", event);
    // 不关心谁订阅了这个事件
}

// 各系统独立订阅
@EventListener(topic = "payment.success")
public void onPaymentSuccess(PaymentSuccessEvent event) {
    // 库存系统处理
}

@EventListener(topic = "payment.success")
public void onPaymentSuccess(PaymentSuccessEvent event) {
    // 积分系统处理
}

生产者 → Broker → 队列/主题 → 消费者

@Service
public class OrderProducer {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    
    public void sendOrderCreatedMessage(Order order) {
        // 构建消息
        OrderCreatedEvent event = new OrderCreatedEvent(
            order.getId(),
            order.getUserId(),
            order.getAmount()
        );
        
        // 发送消息
        rabbitTemplate.convertAndSend(
            "order.exchange",    // 交换机
            "order.created",     // 路由键
            event                // 消息内容
        );
        
        log.info("订单创建消息已发送: orderId={}", order.getId());
    }
}

@Service
public class OrderConsumer {
    @Autowired
    private OrderService orderService;
    
    @RabbitListener(queues = "order.queue")
    public void handleOrderCreated(OrderCreatedEvent event) {
        try {
            // 处理消息
            orderService.processOrderCreated(event);
            
            // 消息处理成功,自动 ACK(RabbitMQ 默认)
            log.info("订单创建消息处理成功: orderId={}", event.getOrderId());
            
        } catch (Exception e) {
            log.error("订单创建消息处理失败: orderId={}", event.getOrderId(), e);
            // 抛出异常,消息重新入队或进入死信队列
            throw new AmqpRejectAndDontRequeueException("处理失败", e);
        }
    }
}

队列: order.queue
消费者A ← 消息1
消费者B ← 消息2
消费者C ← 消息3

主题: order.topic
消费者A ← 消息1, 消息2, 消息3
消费者B ← 消息1, 消息2, 消息3
消费者C ← 消息1, 消息2, 消息3

                    ┌─────────────┐
                    │  生产者问题  │
                    │ - 发送失败   │
                    │ - 消息丢失   │
                    └─────────────┘
                           ↓
┌──────────────┐    ┌─────────────┐    ┌──────────────┐
│ 系统复杂度上升 │←→ │  Broker问题  │ ←→│ 运维成本增加  │
│ - 架构更复杂  │    │ - 宕机丢失   │    │ - 需要监控   │
│ - 排障更困难  │    │ - 磁盘故障   │    │ - 需要备份   │
└──────────────┘    └─────────────┘    └──────────────┘
                           ↓
                    ┌─────────────┐
                    │  消费者问题  │
                    │ - 重复消费   │
                    │ - 消息积压   │
                    └─────────────┘

// 方案1: 同步确认
public void sendWithConfirm(Message message) {
    try {
        rabbitTemplate.convertAndSend("exchange", "routingKey", message);
        // 同步等待确认
        rabbitTemplate.waitForConfirms(5000);
        log.info("消息发送成功");
    } catch (Exception e) {
        log.error("消息发送失败", e);
        // 重试或记录到数据库
        messageStore.saveFailedMessage(message);
    }
}

// 方案2: 异步确认
public void sendWithAsyncConfirm(Message message) {
    rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
        if (ack) {
            log.info("消息确认成功");
        } else {
            log.error("消息确认失败: {}", cause);
            // 重试或补偿
            retryService.retry(message);
        }
    });
    
    rabbitTemplate.convertAndSend("exchange", "routingKey", message);
}

// 方案3: 事务消息(RocketMQ)
public void sendTransactionMessage(Message message) {
    // 发送半消息
    TransactionSendResult result = rocketMQTemplate.sendMessageInTransaction(
        "order.group",
        "order.topic",
        MessageBuilder.withPayload(message).build(),
        null
    );
    
    if (result.getSendStatus() == SendStatus.SEND_OK) {
        log.info("事务消息发送成功");
    }
}

@RocketMQTransactionListener
class OrderTransactionListener implements RocketMQLocalTransactionListener {
    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        // 执行本地事务
        try {
            orderService.createOrder((Order) msg.getPayload());
            return RocketMQLocalTransactionState.COMMIT;
        } catch (Exception e) {
            return RocketMQLocalTransactionState.ROLLBACK;
        }
    }
    
    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message msg) {
        // 检查本地事务状态
        Order order = (Order) msg.getPayload();
        if (orderService.exists(order.getId())) {
            return RocketMQLocalTransactionState.COMMIT;
        }
        return RocketMQLocalTransactionState.ROLLBACK;
    }
}

// RabbitMQ: 开启消息持久化
@Bean
public Queue orderQueue() {
    return QueueBuilder.durable("order.queue")  // 持久化队列
        .withArgument("x-message-ttl", 86400000)  // 消息过期时间
        .build();
}

// 发送持久化消息
public void sendPersistentMessage(Order order) {
    Message message = MessageBuilder
        .withBody(JSON.toJSONString(order).getBytes())
        .setDeliveryMode(MessageDeliveryMode.PERSISTENT)  // 持久化消息
        .build();
    
    rabbitTemplate.send("order.exchange", "order.created", message);
}

// Kafka: 配置副本因子
// server.properties
// default.replication.factor=3
// min.insync.replicas=2

// RocketMQ: 配置同步刷盘
// broker.conf
// flushDiskType=SYNC_FLUSH
// brokerRole=SYNC_MASTER

// × 错误: 自动确认后业务失败
@RabbitListener(queues = "order.queue")
public void handleOrder(Order order) {
    // 自动 ACK(RabbitMQ 默认)
    // 如果下面业务失败,消息已经丢失
    orderService.process(order);
}

// √ 正确: 手动确认
@RabbitListener(queues = "order.queue", ackMode = "MANUAL")
public void handleOrder(Order order, Channel channel, 
                        @Header(AmqpHeaders.DELIVERY_TAG) long tag) {
    try {
        // 1. 先执行业务逻辑
        orderService.process(order);
        
        // 2. 业务成功后再 ACK
        channel.basicAck(tag, false);
        log.info("消息处理成功: orderId={}", order.getId());
        
    } catch (Exception e) {
        log.error("消息处理失败: orderId={}", order.getId(), e);
        
        try {
            // 3. 业务失败,NACK 并重新入队
            channel.basicNack(tag, false, true);
        } catch (IOException ex) {
            log.error("NACK 失败", ex);
        }
    }
}

# RabbitMQ 配置
spring:
  rabbitmq:
    publisher-confirm-type: correlated  # 开启确认机制
    publisher-returns: true              # 开启退回机制
    template:
      mandatory: true                    # 路由失败返回生产者
    listener:
      simple:
        acknowledge-mode: manual         # 手动确认
        prefetch: 1                      # 预取数量

生产者                    Broker                   消费者
  │                         │                        │
  ├──── 发送消息 ──────────>│                        │
  │                         ├── 投递消息 ───────────>│
  │                         │                        ├─ 处理成功
  │                         │<─── ACK ───────────────┤
  │                         │                        │
  │<─── 确认超时 ───────────┤                        │
  │                         │                        │
  ├─── 重新发送 ───────────>│                        │
  │                         ├── 投递消息 ───────────>│
  │                         │                        ├─ 再次处理(重复!)
  │                         │                        │

@Service
public class OrderConsumer {
    @Autowired
    private StringRedisTemplate redisTemplate;
    
    @RabbitListener(queues = "order.queue")
    public void handleOrderCreated(OrderCreatedEvent event) {
        // 使用业务唯一键(订单ID)
        String bizKey = "order:created:" + event.getOrderId();
        
        // SETNX: 如果 key 不存在则设置,返回 true;如果已存在,返回 false
        Boolean isFirst = redisTemplate.opsForValue()
            .setIfAbsent(bizKey, "1", Duration.ofHours(24));
        
        if (Boolean.FALSE.equals(isFirst)) {
            log.warn("重复消息,跳过: orderId={}", event.getOrderId());
            return;  // 直接返回,消息已被处理
        }
        
        // 执行业务逻辑
        orderService.processOrderCreated(event);
    }
}

-- 创建订单表,订单号唯一
CREATE TABLE `t_order` (
  `id` bigint NOT NULL AUTO_INCREMENT,
  `order_no` varchar(64) NOT NULL COMMENT '订单号(唯一)',
  `user_id` bigint NOT NULL,
  `amount` decimal(10,2) NOT NULL,
  `status` varchar(20) NOT NULL,
  PRIMARY KEY (`id`),
  UNIQUE KEY `uk_order_no` (`order_no`)  -- 唯一约束
) ENGINE=InnoDB;

@Service
public class OrderService {
    @Transactional
    public void createOrder(OrderCreatedEvent event) {
        Order order = new Order();
        order.setOrderNo(event.getOrderNo());
        order.setUserId(event.getUserId());
        order.setAmount(event.getAmount());
        
        try {
            orderMapper.insert(order);
        } catch (DuplicateKeyException e) {
            // 唯一约束冲突,说明订单已存在,直接返回
            log.warn("订单已存在,跳过: orderNo={}", event.getOrderNo());
            return;
        }
    }
}

@Service
public class PaymentConsumer {
    @Autowired
    private StringRedisTemplate redisTemplate;
    @Autowired
    private PaymentService paymentService;
    
    public void handlePayment(PaymentEvent event) {
        String key = "payment:idempotent:" + event.getPaymentId();
        String requestId = UUID.randomUUID().toString();
        
        // 1. 尝试获取幂等标记
        Boolean success = redisTemplate.opsForValue()
            .setIfAbsent(key, requestId, Duration.ofDays(7));
        
        if (Boolean.FALSE.equals(success)) {
            log.warn("重复支付请求: paymentId={}", event.getPaymentId());
            return;
        }
        
        try {
            // 2. 执行业务逻辑
            paymentService.processPayment(event);
            
        } catch (Exception e) {
            // 3. 业务失败,删除幂等标记,允许重试
            redisTemplate.delete(key);
            throw e;
        }
    }
}

@Service
public class OrderService {
    // 订单状态流转: CREATED → PAID → SHIPPED → COMPLETED
    public boolean markPaid(Long orderId) {
        Order order = orderMapper.selectById(orderId);
        
        // 只有 CREATED 状态才能变为 PAID
        if (order.getStatus() != OrderStatus.CREATED) {
            log.warn("订单状态不允许支付: orderId={}, status={}", 
                     orderId, order.getStatus());
            return false;  // 状态已流转,幂等返回
        }
        
        // 更新状态(使用乐观锁)
        int rows = orderMapper.updateStatus(
            orderId, 
            OrderStatus.PAID, 
            OrderStatus.CREATED  // 期望原状态
        );
        
        return rows > 0;
    }
}

// Mapper
@Update("UPDATE t_order SET status = #{newStatus} " +
        "WHERE id = #{orderId} AND status = #{oldStatus}")
int updateStatus(@Param("orderId") Long orderId,
                 @Param("newStatus") OrderStatus newStatus,
                 @Param("oldStatus") OrderStatus oldStatus);

// 组合方案: Redis 快速去重 + 数据库兜底
@Service
public class OrderConsumer {
    @Autowired
    private StringRedisTemplate redisTemplate;
    @Autowired
    private OrderService orderService;
    
    @RabbitListener(queues = "order.queue")
    public void handleOrderCreated(OrderCreatedEvent event) {
        // 1. Redis 快速去重(第一道防线)
        String key = "order:dedup:" + event.getOrderId();
        Boolean isFirst = redisTemplate.opsForValue()
            .setIfAbsent(key, "1", Duration.ofDays(7));
        
        if (Boolean.FALSE.equals(isFirst)) {
            log.info("Redis 去重命中: orderId={}", event.getOrderId());
            return;
        }
        
        try {
            // 2. 数据库唯一约束兜底(第二道防线)
            orderService.createOrder(event);
            
        } catch (DuplicateKeyException e) {
            // 数据库层面也已存在,确认是重复消息
            log.warn("数据库去重命中: orderId={}", event.getOrderId());
        } catch (Exception e) {
            // 其他异常,删除 Redis 标记,允许重试
            redisTemplate.delete(key);
            throw e;
        }
    }
}

生产速度: 10000 消息/秒
消费速度: 1000 消息/秒
积压速度: 9000 消息/秒

// 监控示例
@Component
public class MqMonitor {
    @Scheduled(fixedRate = 60000)  // 每分钟监控一次
    public void monitorLag() {
        // 1. 队列积压量
        long messageCount = rabbitTemplate.execute(channel -> {
            AMQP.Queue.DeclareOk declareOk = channel.queueDeclarePassive("order.queue");
            return declareOk.getMessageCount();
        });
        
        // 2. 消费延迟
        long lagTime = System.currentTimeMillis() - getLastMessageTime();
        
        // 3. 消费速率
        double consumeRate = getConsumeRate();
        
        // 4. 告警
        if (messageCount > 10000) {
            alertService.sendAlert("消息积压告警", 
                String.format("队列: order.queue, 积压: %d, 建议: 扩容消费者", messageCount));
        }
    }
}

# RabbitMQ: 增加消费者并发
# application.yml
spring:
  rabbitmq:
    listener:
      simple:
        concurrency: 10     # 最小并发数(原: 5)
        max-concurrency: 20 # 最大并发数(原: 10)

// Kafka: 增加分区数
// 注意: 分区数只能增加,不能减少
kafka-topics.sh --alter --topic order.topic \
  --partitions 10 \
  --bootstrap-server localhost:9092

// 增加消费者实例
// Kubernetes: 扩容 Pod
kubectl scale deployment order-consumer --replicas=5

// × 优化前: 每条消息查数据库
@RabbitListener(queues = "order.queue")
public void handleOrder(OrderEvent event) {
    Order order = orderMapper.selectById(event.getOrderId());  // 查 DB
    User user = userMapper.selectById(order.getUserId());      // 查 DB
    productService.getProduct(order.getProductId());           // 查 DB
    // 单条消息耗时: 50ms+
}

// √ 优化后: 批量查询 + 缓存
@RabbitListener(queues = "order.queue", containerFactory = "batchFactory")
public void handleOrders(List<OrderEvent> events) {
    // 批量查询
    List<Long> orderIds = events.stream()
        .map(OrderEvent::getOrderId)
        .collect(Collectors.toList());
    Map<Long, Order> orderMap = orderMapper.selectBatchIds(orderIds)
        .stream()
        .collect(Collectors.toMap(Order::getId, Function.identity()));
    
    // 批量处理
    events.forEach(event -> {
        Order order = orderMap.get(event.getOrderId());
        // 处理逻辑
    });
    
    // 单条消息耗时: 5ms
}

@Service
public class OrderConsumer {
    @Autowired
    private DeadLetterService deadLetterService;
    
    private static final int MAX_RETRY = 3;
    
    @RabbitListener(queues = "order.queue")
    public void handleOrder(OrderEvent event, 
                           @Header("retry_count", required = false) Integer retryCount) {
        
        // 检查重试次数
        if (retryCount != null && retryCount >= MAX_RETRY) {
            // 超过最大重试次数,转入死信队列
            deadLetterService.sendToDeadLetter(event, "重试次数超限");
            return;
        }
        
        try {
            orderService.process(event);
            
        } catch (BusinessException e) {
            // 业务异常,不应该重试
            deadLetterService.sendToDeadLetter(event, e.getMessage());
            
        } catch (Exception e) {
            // 系统异常,可以重试
            throw new AmqpRejectAndDontRequeueException("重试", e);
        }
    }
}

@Service
public class OrderConsumer {
    @Autowired
    private RateLimiter rateLimiter;  // Guava RateLimiter
    
    @RabbitListener(queues = "order.queue")
    public void handleOrder(OrderEvent event) {
        // 限流: 每秒最多处理 100 条
        if (!rateLimiter.tryAcquire()) {
            log.warn("触发限流,消息延迟处理");
            throw new AmqpRejectAndDontRequeueException("限流中");
        }
        
        // 降级: 非核心业务跳过
        if (event.isNonCritical()) {
            log.info("非核心业务,降级处理");
            return;
        }
        
        orderService.process(event);
    }
}

发现积压
    ↓
确认原因
    ├─ 生产突增 → 临时扩容
    ├─ 消费变慢 → 优化逻辑
    ├─ 异常阻塞 → 隔离毒消息
    └─ 下游故障 → 限流降级
    ↓
执行治理
    ↓
持续监控
    ↓
恢复后复盘

@Service
@Slf4j
public class OrderProducer {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    
    @PostConstruct
    public void init() {
        // 开启确认机制
        rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
            if (!ack) {
                log.error("消息发送失败: {}", cause);
                // 发送失败,记录到数据库或重试
                if (correlationData != null) {
                    failedMessageStore.save(correlationData);
                }
            }
        });
        
        // 开启退回机制
        rabbitTemplate.setReturnsCallback(returned -> {
            log.error("消息路由失败: {}", returned.getReplyText());
            // 路由失败,告警
            alertService.sendAlert("消息路由失败", returned.toString());
        });
    }
    
    public void sendOrderCreated(Order order) {
        OrderCreatedEvent event = new OrderCreatedEvent(order);
        
        // 设置消息 ID,用于确认回调
        CorrelationData correlationData = new CorrelationData(
            UUID.randomUUID().toString()
        );
        
        // 发送消息
        rabbitTemplate.convertAndSend(
            "order.exchange",
            "order.created",
            event,
            message -> {
                // 设置消息持久化
                message.getMessageProperties().setDeliveryMode(
                    MessageDeliveryMode.PERSISTENT
                );
                // 设置过期时间
                message.getMessageProperties().setExpiration("86400000");
                return message;
            },
            correlationData
        );
    }
}

@Service
@Slf4j
public class OrderConsumer {
    @Autowired
    private StringRedisTemplate redisTemplate;
    @Autowired
    private OrderService orderService;
    @Autowired
    private DeadLetterService deadLetterService;
    
    @RabbitListener(
        queues = "order.queue",
        ackMode = "MANUAL",
        concurrency = "5-10"  // 并发数
    )
    public void handleOrderCreated(
            OrderCreatedEvent event,
            Channel channel,
            @Header(AmqpHeaders.DELIVERY_TAG) long tag) {
        
        String orderId = event.getOrderId();
        String dedupKey = "order:dedup:" + orderId;
        
        try {
            // 1. 幂等性检查
            Boolean isFirst = redisTemplate.opsForValue()
                .setIfAbsent(dedupKey, "1", Duration.ofDays(7));
            
            if (Boolean.FALSE.equals(isFirst)) {
                log.info("重复消息,跳过: orderId={}", orderId);
                // 重复消息,直接 ACK,不重新入队
                channel.basicAck(tag, false);
                return;
            }
            
            // 2. 执行业务逻辑
            orderService.processOrderCreated(event);
            
            // 3. 业务成功,ACK
            channel.basicAck(tag, false);
            log.info("订单创建处理成功: orderId={}", orderId);
            
        } catch (BusinessException e) {
            // 4. 业务异常,不应该重试,转入死信
            log.error("业务异常,转入死信: orderId={}", orderId, e);
            deadLetterService.sendToDeadLetter(event, e.getMessage());
            channel.basicAck(tag, false);
            
        } catch (Exception e) {
            // 5. 系统异常,可以重试
            log.error("系统异常,准备重试: orderId={}", orderId, e);
            redisTemplate.delete(dedupKey);  // 删除幂等标记,允许重试
            
            try {
                // NACK,重新入队
                channel.basicNack(tag, false, true);
            } catch (IOException ex) {
                log.error("NACK 失败", ex);
            }
        }
    }
}

@Component
@Slf4j
public class MqMonitor {
    @Autowired
    private RabbitAdmin rabbitAdmin;
    @Autowired
    private AlertService alertService;
    
    @Scheduled(fixedRate = 60000)
    public void monitor() {
        // 获取队列信息
        Properties props = rabbitAdmin.getQueueProperties("order.queue");
        if (props == null) {
            log.error("队列不存在: order.queue");
            return;
        }
        
        // 队列深度
        Integer messageCount = (Integer) props.get("messageCount");
        // 消费者数量
        Integer consumerCount = (Integer) props.get("consumerCount");
        
        log.info("队列监控: messageCount={}, consumerCount={}", 
                 messageCount, consumerCount);
        
        // 告警规则
        if (messageCount > 10000) {
            alertService.sendAlert(
                "消息积压告警",
                String.format("队列: order.queue, 积压: %d, 消费者: %d", 
                              messageCount, consumerCount)
            );
        }
        
        if (consumerCount == 0) {
            alertService.sendAlert(
                "消费者离线告警",
                "队列: order.queue 没有消费者"
            );
        }
    }
}

// × 问题: 同步调用,响应慢
public Result createOrder(Order order) {
    orderService.save(order);           // 50ms
    smsService.sendSms(order);          // 200ms
    emailService.sendEmail(order);      // 300ms
    notificationService.notify(order);  // 100ms
    return Result.success();            // 总耗时: 650ms
}

// √ 改进: 异步处理,快速响应
@Service
public class OrderService {
    @Autowired
    private OrderMapper orderMapper;
    @Autowired
    private RabbitTemplate rabbitTemplate;
    
    @Transactional
    public Result createOrder(Order order) {
        // 1. 保存订单
        orderMapper.insert(order);
        
        // 2. 发送订单创建事件
        OrderCreatedEvent event = new OrderCreatedEvent(
            order.getId(),
            order.getUserId(),
            order.getAmount()
        );
        rabbitTemplate.convertAndSend("order.exchange", "order.created", event);
        
        return Result.success();
        // 总耗时: 50ms(订单落库) + 5ms(发送消息) = 55ms
    }
}

// 短信消费者
@Service
public class SmsConsumer {
    @RabbitListener(queues = "sms.queue")
    public void handleOrderCreated(OrderCreatedEvent event) {
        smsService.sendSms(event.getOrderId());
    }
}

// 邮件消费者
@Service
public class EmailConsumer {
    @RabbitListener(queues = "email.queue")
    public void handleOrderCreated(OrderCreatedEvent event) {
        emailService.sendEmail(event.getOrderId());
    }
}

// 站内通知消费者
@Service
public class NotificationConsumer {
    @RabbitListener(queues = "notification.queue")
    public void handleOrderCreated(OrderCreatedEvent event) {
        notificationService.notify(event.getOrderId());
    }
}

// 支付服务发布事件
@Service
public class PaymentService {
    @Autowired
    private PaymentMapper paymentMapper;
    @Autowired
    private RocketMQTemplate rocketMQTemplate;
    
    @Transactional
    public void paymentSuccess(String paymentId) {
        // 1. 更新支付状态
        Payment payment = paymentMapper.selectById(paymentId);
        payment.setStatus(PaymentStatus.SUCCESS);
        paymentMapper.updateById(payment);
        
        // 2. 发送支付成功事件
        PaymentSuccessEvent event = new PaymentSuccessEvent(
            payment.getOrderId(),
            payment.getUserId(),
            payment.getAmount()
        );
        
        // 使用事务消息,保证支付状态和消息发送的一致性
        rocketMQTemplate.sendMessageInTransaction(
            "payment-group",
            "payment-success-topic",
            MessageBuilder.withPayload(event).build(),
            null
        );
    }
}

// 订单服务订阅
@Service
public class OrderConsumer {
    @RocketMQMessageListener(
        topic = "payment-success-topic",
        consumerGroup = "order-group"
    )
    public void onMessage(PaymentSuccessEvent event) {
        orderService.markPaid(event.getOrderId());
    }
}

// 库存服务订阅
@Service
public class InventoryConsumer {
    @RocketMQMessageListener(
        topic = "payment-success-topic",
        consumerGroup = "inventory-group"
    )
    public void onMessage(PaymentSuccessEvent event) {
        inventoryService.deduct(event.getOrderId());
    }
}

// 优惠券服务订阅
@Service
public class CouponConsumer {
    @RocketMQMessageListener(
        topic = "payment-success-topic",
        consumerGroup = "coupon-group"
    )
    public void onMessage(PaymentSuccessEvent event) {
        couponService.grant(event.getUserId());
    }
}

// 秒杀服务
@Service
public class SeckillService {
    @Autowired
    private RedisTemplate redisTemplate;
    @Autowired
    private RabbitTemplate rabbitTemplate;
    
    // 秒杀接口
    public Result seckill(Long userId, Long productId) {
        // 1. 快速校验
        String token = (String) redisTemplate.opsForValue()
            .get("seckill:token:" + productId);
        if (token == null) {
            return Result.fail("秒杀已结束");
        }
        
        // 2. Redis 预扣库存
        Long stock = redisTemplate.opsForValue().decrement("seckill:stock:" + productId);
        if (stock < 0) {
            redisTemplate.opsForValue().increment("seckill:stock:" + productId);
            return Result.fail("库存不足");
        }
        
        // 3. 发送到 MQ
        SeckillRequest request = new SeckillRequest(userId, productId);
        rabbitTemplate.convertAndSend("seckill.exchange", "seckill.request", request);
        
        return Result.success("排队中,请稍后查询结果");
    }
}

// 秒杀消费者(限速处理)
@Service
public class SeckillConsumer {
    @Autowired
    private SeckillService seckillService;
    
    @RabbitListener(
        queues = "seckill.queue",
        concurrency = "1-1"  // 并发度为 1,限速处理
    )
    public void handleSeckill(SeckillRequest request) {
        try {
            seckillService.createOrder(request.getUserId(), request.getProductId());
            log.info("秒杀成功: userId={}, productId={}", 
                     request.getUserId(), request.getProductId());
        } catch (Exception e) {
            // 失败时恢复库存
            redisTemplate.opsForValue().increment("seckill:stock:" + request.getProductId());
            log.error("秒杀失败", e);
        }
    }
}

// × 没有幂等保护
@RabbitListener(queues = "order.queue")
public void handleOrder(OrderEvent event) {
    // 直接处理,可能导致重复发货、重复扣款
    orderService.ship(event.getOrderId());
}

// √ 幂等处理
@RabbitListener(queues = "order.queue")
public void handleOrder(OrderEvent event) {
    String key = "order:ship:" + event.getOrderId();
    if (!redisTemplate.opsForValue().setIfAbsent(key, "1", Duration.ofDays(1))) {
        return;  // 已处理
    }
    orderService.ship(event.getOrderId());
}

// × 不适合用 MQ 的场景: 需要立即返回结果
public Result queryOrder(Long orderId) {
    // 查询操作,不应该用 MQ
    rabbitTemplate.convertAndSend("query.exchange", "query.order", orderId);
    // 无法立即返回结果
}

// 监控队列积压
@Scheduled(fixedRate = 60000)
public void monitor() {
    Properties props = rabbitAdmin.getQueueProperties("order.queue");
    Integer messageCount = (Integer) props.get("messageCount");
    
    if (messageCount > 10000) {
        alertService.sendAlert("队列积压", messageCount);
    }
}

// × 消息投递成功,但业务失败
@RabbitListener(queues = "order.queue")
public void handleOrder(OrderEvent event) {
    try {
        orderService.process(event);
    } catch (Exception e) {
        log.error("处理失败", e);
        // 吞掉异常,消息被 ACK,但业务未成功
    }
}

// √ 业务失败,消息重试或转死信
@RabbitListener(queues = "order.queue", ackMode = "MANUAL")
public void handleOrder(OrderEvent event, Channel channel, 
                        @Header(AmqpHeaders.DELIVERY_TAG) long tag) {
    try {
        orderService.process(event);
        channel.basicAck(tag, false);
    } catch (Exception e) {
        log.error("处理失败", e);
        // 业务失败,NACK,让消息重新入队或转死信
        channel.basicNack(tag, false, true);
    }
}

用户下单
    ↓
订单服务 → MQ(order.created)
              ↓
         ┌────┴────┬────────┬────────┐
         ↓         ↓        ↓        ↓
      库存服务  优惠券服务  积分服务  通知服务

关键设计:
1. 库存扣减: 使用事务消息,保证订单和库存一致性
2. 幂等性: 所有消费者都需要幂等处理
3. 监控: 监控队列积压、消费延迟、失败率
4. 降级: 通知服务失败不影响订单创建

秒杀请求
    ↓
秒杀服务 → Redis 预扣库存 → MQ(seckill.request)
                                  ↓
                            秒杀消费者(限速 1000 TPS)
                                  ↓
                             创建订单

关键设计:
1. Redis 预扣库存: 快速过滤无效请求
2. MQ 削峰: 缓冲请求,平滑处理
3. 限速消费: 并发度=1,保证数据库不被打垮
4. 失败恢复: 库存回滚