📚 课程目录第4章消息队列削峰填谷

4、消息队列削峰填谷：RabbitMQ/Kafka在发证场景下的异步解耦实战

发证系统在高峰期会遇到什么情况？我见过最夸张的一次，每秒涌进来 8000 多个发证请求。数据库连接池瞬间被打满，CPU 直接飙到 95%。用户那边呢？页面一直转圈，最后超时报错。

说白了，这就是典型的「瞬时流量尖峰」问题。发证请求来得又快又猛，但后端处理能力是有限的。硬扛肯定不行，加机器也不是万能的。这时候就需要一个缓冲层——消息队列。

  核心思路： 请求先入队，后端按能力消费。削峰填谷，把尖峰流量拉平，系统才能稳得住。

4.1 为什么发证场景必须用消息队列？

你想想看，发证流程其实挺重的。要校验身份、查数据库、生成证书、写存储、记录日志……这一套下来，少说几百毫秒。如果每个请求都同步等结果，用户等不起，系统也扛不住。

我习惯把发证拆成两个阶段：

前端阶段： 只做参数校验和入队，快速返回「受理成功」。
后端阶段： 从队列拉取消息，异步完成真正的发证逻辑。

这样做的好处很明显：

解耦： 前端不用关心后端怎么处理，后端也不受前端流量波动影响。
削峰： 流量再大，队列就是缓冲区，不会直接打到数据库。
填谷： 低谷期队列积压的消息慢慢消费，资源利用率更均衡。

我的经验： 曾经有个项目没上队列，发证接口直接调数据库。双十一那天，数据库连接数飙到 2000+，最后 DBA 直接 kill 了进程。后来上了 RabbitMQ，同样流量下，数据库连接稳定在 50 以内。

4.2 RabbitMQ vs Kafka：发证场景怎么选？

很多人纠结这个问题。我直接说结论：

对比维度	RabbitMQ	Kafka
消息模型	队列 + 交换机，灵活路由	分区 + 偏移量，顺序消费
吞吐量	万级/秒	百万级/秒
消息可靠性	强，支持确认和事务	强，支持副本和 ACK
延迟	微秒级	毫秒级
适用场景	低延迟、复杂路由、任务分发	高吞吐、日志收集、流处理

发证场景怎么选？我个人的建议是：

日均发证量在 10 万以内，对延迟敏感： 选 RabbitMQ。它路由灵活，能轻松实现「按证书类型分发到不同消费者」。
日均发证量在百万级以上，需要海量堆积： 选 Kafka。它的吞吐量和持久化能力更强，适合大规模批量发证。

注意： 我曾经见过有人用 Kafka 做实时发证，结果因为 Kafka 的批量发送特性，导致消息延迟了 2-3 秒。用户那边以为没提交成功，又点了一次……最后重复发证了。所以，延迟敏感的场景慎用 Kafka。

4.3 RabbitMQ 实战：发证请求的异步解耦

先看一个典型的 RabbitMQ 发证架构：

代码实现其实不复杂。我贴一个核心的生产者代码：

// 生产者：发证请求入队
public void sendCertRequest(CertRequest request) {
    // 1. 构建消息
    String message = JSON.toJSONString(request);
    
    // 2. 设置消息属性
    AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
        .deliveryMode(2)  // 持久化
        .priority(request.getPriority())  // 优先级
        .build();
    
    // 3. 发送到交换机
    channel.basicPublish(
        "cert.exchange",    // 交换机
        "cert.routing.key", // 路由键
        props,
        message.getBytes()
    );
    
    // 4. 记录日志
    log.info("发证请求已入队: {}", request.getCertId());
}

消费者这边，我习惯用 Spring AMQP 的 @RabbitListener：

@RabbitListener(queues = "cert.queue")
public void handleCertRequest(String message) {
    CertRequest request = JSON.parseObject(message, CertRequest.class);
    
    try {
        // 1. 校验参数
        validate(request);
        
        // 2. 生成证书
        Cert cert = certService.generate(request);
        
        // 3. 存储证书
        certRepository.save(cert);
        
        // 4. 手动 ACK（默认自动确认）
        // 这里用自动确认模式，抛出异常会自动重试
    } catch (Exception e) {
        log.error("发证失败: {}", request.getCertId(), e);
        // 抛出异常，消息会重回队列或进入死信队列
        throw new AmqpRejectAndDontRequeueException(e);
    }
}

避坑指南： 我曾经犯过一个错误——消费者没做幂等处理。结果消息重试时，同一个请求生成了两张证书。后来我在发证前先查一次数据库，如果证书已存在就直接返回。记住：消息队列只能保证至少一次投递，不能保证恰好一次。

4.4 Kafka 实战：高吞吐发证场景

如果发证量特别大，比如每天几百万张，RabbitMQ 就有点吃力了。这时候我倾向用 Kafka。

Kafka 的核心优势是：

顺序写入： 磁盘顺序写，吞吐量极高。
分区并行： 一个主题多个分区，消费者可以并行消费。
消息堆积： 积压几亿条消息都不会影响性能。

发证场景下，我一般这样设计：

// Kafka 生产者：批量发证
public void batchSendCert(List<CertRequest> requests) {
    // 1. 按证书类型分区
    Map<String, List<CertRequest>> grouped = requests.stream()
        .collect(Collectors.groupingBy(CertRequest::getCertType));
    
    // 2. 每个分区发送
    grouped.forEach((type, list) -> {
        for (CertRequest request : list) {
            ProducerRecord<String, String> record = new ProducerRecord<>(
                "cert.topic",          // 主题
                type,                  // 分区键
                JSON.toJSONString(request)
            );
            
            // 3. 异步发送，带回调
            producer.send(record, (metadata, exception) -> {
                if (exception != null) {
                    log.error("发送失败: {}", request.getCertId(), exception);
                } else {
                    log.info("发送成功: partition={}, offset={}", 
                        metadata.partition(), metadata.offset());
                }
            });
        }
    });
}

消费者这边，我习惯用 @KafkaListener 配合手动提交：

@KafkaListener(topics = "cert.topic", groupId = "cert-group")
public void consumeCertRequest(ConsumerRecord<String, String> record, 
                                Acknowledgment ack) {
    CertRequest request = JSON.parseObject(record.value(), CertRequest.class);
    
    try {
        // 1. 幂等校验
        if (certRepository.existsById(request.getCertId())) {
            log.warn("证书已存在，跳过: {}", request.getCertId());
            ack.acknowledge();
            return;
        }
        
        // 2. 生成证书
        Cert cert = certService.generate(request);
        certRepository.save(cert);
        
        // 3. 手动提交偏移量
        ack.acknowledge();
        
    } catch (Exception e) {
        log.error("消费失败: {}", request.getCertId(), e);
        // 不提交偏移量，下次重新消费
        // 但要注意：如果一直失败，会形成死循环
        // 我一般会记录到死信表，人工处理
    }
}

注意： Kafka 的消费者默认是自动提交偏移量，每 5 秒提交一次。如果处理时间超过 5 秒，消费者挂了，重启后会重复消费。我建议改成手动提交，处理完一条提交一条。虽然性能会降一点，但可靠性高很多。

4.5 削峰填谷的实战效果

最后看一组真实数据。这是我之前优化过的一个发证系统：

指标	优化前（同步）	优化后（RabbitMQ）	优化后（Kafka）
峰值 TPS	500	3000	15000
平均响应时间	2.3s	50ms（入队）	30ms（入队）
数据库连接数	200	30	20
系统可用性	99.2%	99.95%	99.99%

你看，用了消息队列之后，峰值吞吐量提升了 6-30 倍，响应时间反而降到了毫秒级。数据库连接数从 200 降到了 20-30，系统稳得很。

我的心得： 削峰填谷不是银弹。队列积压太多，消费跟不上，一样会出问题。我一般会监控队列长度，设置告警。如果积压超过阈值，就临时扩容消费者。另外，消息的 TTL 也要设置，避免过期消息堆积。

嗯，消息队列这块就聊这么多。说白了，发证场景下，RabbitMQ 和 Kafka 都能用，关键看你的量级和延迟要求。选对了，系统稳如老狗；选错了，后面有的坑要填。