阅读全文

⭐⭐⭐ Spring Boot 项目实战	⭐⭐⭐ Spring Cloud 项目实战
《Dubbo 实现原理与源码解析 —— 精品合集》	《Netty 实现原理与源码解析 —— 精品合集》
《Spring 实现原理与源码解析 —— 精品合集》	《MyBatis 实现原理与源码解析 —— 精品合集》
《Spring MVC 实现原理与源码解析 —— 精品合集》	《数据库实体设计合集》
《Spring Boot 实现原理与源码解析 —— 精品合集》	《Java 面试题 + Java 学习指南》

摘要: 原创出处 捡田螺的小男孩 「捡田螺的小男孩」欢迎转载，保留摘要，谢谢！

• 前言
• 1. 先排查是不是bug，如果是，要快速修复
• 2. 优化消费者代码逻辑
• 3. 临时紧急扩容，新建临时topic
• 最后

---

🙂🙂🙂关注微信公众号：【芋道源码】有福利：

1. RocketMQ / MyCAT / Sharding-JDBC 所有源码分析文章列表
2. RocketMQ / MyCAT / Sharding-JDBC 中文注释源码 GitHub 地址
3. 您对于源码的疑问每条留言都将得到认真回复。甚至不知道如何读源码也可以请教噢。
4. 新的源码解析文章实时收到通知。每周更新一篇左右。
5. 认真的源码交流微信群。

---

前言

大家在日常开发中，是否处理过大批量消息积压的问题呢？

它一般由于代码bug（比如消费逻辑处理有误）、或者生产者的生产速度大于消费者的消费速度（如大促、抢购等活动期间导致消息数量激增，或者消费者处理速度极慢），就可能导致生产环境出现百万、甚至千万的消息积压。

那么，假设发生kafka百万消息堆积，如何解决呢？

• 先排查是不是bug，如果是，要快速修复
• 优化消费者代码逻辑
• 临时紧急扩容，新建临时topic

1. 先排查是不是bug，如果是，要快速修复

遇到消息积压问题时，我们需要先排查，是不是有bug产生了,比如消费者未正确提交偏移量（Offset）。

消费者在处理完消息后未提交偏移量，导致重复消费或消费停滞。进而形成大量消息积压。

给个伪代码反例：

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> record : records) {
        process(record);
        // 未提交偏移量
    }
}

在处理完消息后，要正确提交偏移量。

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> record : records) {
        process(record);
    }
    //提交偏移量
    consumer.commitSync();
}

2. 优化消费者代码逻辑

如果不是bug，那就可能是消费者速度不给力，导致的消息积压。我们可以优化一下消费者代码逻辑。

可以使用多线程处理，可以减少每条消息的处理时间（比如减少不必要的计算），从而提高消息处理速度。

假设消费者有两台机器，消费者代码优化前是，1秒处理100条消息。代码优化后，l秒可以处理消息500条。

一个小时，可以处理消息：2 500 3600 = 3600 000

可以发现，如果累积了3百多万消息的话，处理完也要一个小时。如果是生产环境，一些比较敏感或者特殊的业务，是不允许很长的时间延迟的。

3. 临时紧急扩容，新建临时topic

业务紧急的话，我们可以临时紧急扩容，新建临时topic。

比如原来的topic 只有两个partition分区，因为消费者处理很耗时等操作，导致了百万消息积压，这时候需要紧急快速处理。

这时候，消费者的代码，我们可以做一些调整，就是不再处理其他业务操作。而是新建临时的topic，把消息转发到临时的topic，并且partition 分区增加到原来的 10倍

然后我们原来消费者业务逻辑处理的代码，放在新的临时消息那里处理。

等快速消费完积压数据之后，得恢复原先部署的架构，下掉临时消费者，重新用原先的 consumer 机器来消费消息。

最后

对于线上kafka 消息大量积压的问题，我总结了这几点：

• 我们要做好监控和告警，当消息积压到一定程度的时候，就要告警，通知负责人，提前处理。
• 不要上来就新建临时topic，去快速处理大量积压问题。应该先排查是不是bug，优化消费者的代码。
• 如果消息设置了超时时间，因为百万消息积压，没来得及处理就过期清理，可以设置定时任务拉起来重发一下。