线上Kafka积压300万消息，我是怎么3小时清空的?

线上Kafka消息堆积，尤其是百万级别的那种，说轻松点是个技术问题，说严重点，这玩意儿要是处理不好，真能引发生产事故。我经历过类似的惨案，生产环境直接堆了几百万条消息，监控爆红，业务方一边在群里催命，一边在钉钉上“@全员”...那场景，你懂的，跟打仗没啥区别。

这种问题的核心，其实就一个字：慢。

不管是消息处理慢，还是处理失败后重复堆积，反正最后都会变成“山积如山”的状态。所以解决这个问题，要么让消息处理变快，要么让它能错得优雅点，不要挂住整个队列。至于代码bug，那真的是基础中的基础，先修复是没错的，但我更关注的是怎么从架构和实操角度快速解燃眉之急。

比如说，最常见的锅就是消费者忘了提交偏移量。这看起来是个低级错误，但真有不少人踩坑。我就亲眼见过某次项目上线后，Kafka消费逻辑里consumer.commitSync()直接没写...开发自己本地测试没发现问题，因为数据量小，一次poll一条，测试完就没了；上了线之后，每次重启就从头开始消费，消息翻来覆去地处理，然后慢慢地堆了几十万条，业务直接卡住。

修复方法倒也简单，就是确保处理完业务逻辑之后，及时且可靠地提交offset。不过这里有个点得注意：你不能在处理前提交，也不能处理一半就提交，万一中途异常，那就是数据丢失或脏数据。

一般我推荐用这样的模式：

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    try {
        for (ConsumerRecord<String, String> record : records) {
            handleMessage(record);
        }
        consumer.commitSync();
    } catch (Exception e) {
        log.error("处理Kafka消息出错", e);
        // 异常不能提交offset，下一轮再尝试
    }
}

这段代码的重点是：处理完再提交，中间出错不提交，避免消费状态错乱。这点看起来小，但很多线上事故都是这种“省略细节”引起的。

当然，如果代码逻辑没问题，那很可能就是消费速度跟不上。这个时候，多线程消费+批处理优化，是最直观的提速方案。

我们之前有个订单系统，促销高峰期每分钟Kafka里塞进来几万条订单确认消息，原来单线程消费+同步处理的方式直接跪了，CPU打满，队列一直堆。后来我用了线程池+阻塞队列做了个消费中间件，每次poll到消息后，不直接处理，而是扔到一个线程池里异步处理，甚至还做了消息合并，比如相同用户的多条消息合并处理，减少IO操作。

大概这么搞：

ExecutorService executor = Executors.newFixedThreadPool(10);
BlockingQueue<ConsumerRecord<String, String>> queue = new LinkedBlockingQueue<>();

// 拉消息线程
new Thread(() -> {
    while (true) {
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
        for (ConsumerRecord<String, String> record : records) {
            queue.put(record); // 队列满就阻塞
        }
    }
}).start();

// 消费线程
for (int i = 0; i < 10; i++) {
    executor.submit(() -> {
        while (true) {
            ConsumerRecord<String, String> record = queue.take();
            handleMessage(record);
        }
    });
}

这种做法能显著提升吞吐量，关键是要把handleMessage内部处理逻辑优化好，避免阻塞，比如数据库连接池要足够，IO操作别写死循环等待，Redis别死锁这些基础问题。

但有时候呢，即便你再怎么优化，也追不上积压速度。这时候，扩容就成了最直接有效的手段。

我见过最暴力的做法是——直接新建临时Topic，加上10倍分区，然后并发拉满疯狂处理。

你要知道，Kafka分区数决定了并发数。原来2个partition的topic，不管你部署多少个消费者实例，最多也就两条线程在处理。而你堆了几百万条消息，这速度就跟蜗牛一样。

于是，搞一个temp-high-speed-topic，分区开到20个，用个简单的转发器把原来的数据重定向过来：

while (true) {
    ConsumerRecords<String, String> records = oldConsumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> record : records) {
        producer.send(new ProducerRecord<>("temp-high-speed-topic", record.key(), record.value()));
    }
}

然后新开一堆临时消费者，按分区打散并发消费，几分钟干完平时几个小时的事。

这种做法当然有成本，比如临时加机器，维护多份部署架构，还得考虑消费顺序性和幂等处理（不然出事你都找不到谁写的锅代码）。

但在业务逼着你必须五分钟内解决堆积的时候，这就是救命的操作。

哦对了，别忘了处理完这些数据之后，一定要恢复原先架构，该缩容的缩容，该下掉的下掉。临时Topic不能常驻，否则资源开销会非常大，而且容易留下隐藏问题，比如监控不到位、重复消费等。

还有一点很重要的就是：监控和预警机制要拉满。

很多人是积压几百万才发现问题，这就太晚了。Kafka的Lag指标是可以精细化监控的，我们现在的做法是，任何一个Consumer Lag超过10000，就告警；达到50000，自动触发扩容逻辑。

配合这个机制，我们做了一个小组件叫做Kafka Guardian，会定期扫描每个ConsumerGroup的Lag数据，发现某个topic堆积严重时，会自动通知开发，并提示排查链路是否异常，是否需要手动介入或自动扩容。

实际上，大多数Kafka的“灾难”场景，都是“监控不到+误操作+处理不及时”这三件事叠加的结果。你只要把这三点守住了，再加上合理的限流降级机制，Kafka基本不会背锅。

最后留个问题：如果你的Kafka消费逻辑依赖第三方接口（比如调用外部服务），那你该怎么处理消息超时、失败重试、幂等问题？尤其是在积压已经发生的前提下，怎么避免因为接口限流或报错导致雪上加霜？我后面会分享一次我们处理“接口慢+Kafka堆积”的真实案例，欢迎讨论。