概述

如何处理消息堆积问题? 这个问题也是经常面试被问到的问题。除此之外当然是有实际应用场景的。笔者在公司就碰到消息堆积问题，笔者公司是做流量监控防网络攻击的，因此需要对路由器所有流量进行处理和风险过滤，你可以想象一个公司的流量数据，公司员工只要在使用电脑就有进出流量，我们就要对每个请求进行处理。除此之外，我见过一些常见错误处理消息堆积问题的方式也分享给大家。

增加分区同时增加消费实例

kafka中一个消费组内的一个消费者线程只能对应一个主题内的一个分区进行消费，也就是说如果你单独增加消费者线程对消息堆积问题是没有任何效果的，只会浪费多余的消费者线程，只有在增加了分区，多余的消费者线程才能进行工作，否则空闲。以下我做了一个案例供大家参考

案例背景：

1.topic 为quickstart-events

2.Partitions 为 3

3.内部有200条消息等待消费

4.消费组为 test-consumer-group-2

案例代码和运行结果：

//代码
@KafkaHandler
    @KafkaListener(topics = "quickstart-events",groupId = "test-consumer-group-2",concurrency = "10")
    public void test4(String msg){
        try {
            Thread.sleep(10);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("接收到消息:"+msg+"--"+System.currentTimeMillis()+"---"+Thread.currentThread().hashCode());
    }
    
//运行结果
接收到消息:hello1(消息)--1623552780784(消费时间)---1714576524（消费者线程hashcode）--最早消费
接收到消息:hello2--1623552780784---1194736805
接收到消息:hello0--1623552780784---1755757420

接收到消息:hello183--1623552781479---1714576524
接收到消息:hello198--1623552781490---1194736805
接收到消息:hello186--1623552781490---1714576524
接收到消息:hello188--1623552781500---1714576524
接收到消息:hello191--1623552781510---1714576524
接收到消息:hello199--1623552781521---1714576524---最后消费

1.每条消息消费时间是10ms,一共消耗2000ms,三个线程同时消费，最后平均耗时737ms
2.如果concurrency = "1"，消耗时间是2000+ms
3.这里我设置concurrency = "10"，明显看到只有1714576524，1194736805，1755757420在使用
4.所以设置concurrency比分区数大是常见的错误
复制代码

单个消费者线程使用异步消费(线程池)

当我们分区固定情况下，我们其实也可以通过单个消费者线程提高消费速度来解决消息堆积问题。那么最常见的方案就是使用线程池，这样其实可以最大限度利用机器的cpu,从而提高消费能力。其实这种方案和增加分区和消费线程实例本质上是一样的，都是提高并发处理能力，只是增加分区更方便实现，而使用异步操作要我们管理好offset,防止消息丢失。以下我做了一个案例供大家参考

案例背景：

1.topic 为quickstart-events

2.Partitions 为 3

3.内部有200条消息等待消费

4.消费组为 test-consumer-group-2

5.关闭自动提交 spring.kafka.consumer.enable-auto-commit=false

6.spring.kafka.listener.ack-mode=manual

@KafkaHandler
    @KafkaListener(topics = "quickstart-events",groupId = "test-consumer-group-2",concurrency = "1")
    public void test5(String msg,Acknowledgment ack){

        executorService.submit(()->{
            try {
                Thread.sleep(10);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("接收到消息:"+msg+"--"+System.currentTimeMillis()+"---"+Thread.currentThread().hashCode());
            //手动提交
            ack.acknowledge();
        });
    }
    
//运行结果
接收到消息:hello5(消息)--1623554862457（消费时间）---996771787（线程池线程hashcode）--最早消费消息
接收到消息:hello1--1623554862457---1477599004
接收到消息:hello3--1623554862457---1621154349
接收到消息:hello14--1623554862473---1621154349

接收到消息:hello188--1623554863146---1621154349
接收到消息:hello191--1623554863148---996771787
接收到消息:hello194--1623554863148---1477599004
接收到消息:hello196--1623554863156---1621154349
接收到消息:hello198--1623554863159---1477599004--最晚消费消息

1.每条消息消费时间是10ms,一共消耗2000ms,三个线程同时消费，最后平均耗时702ms
2.concurrency = "1"，但是内部使用多线程消费
3.为了防止消息丢失，启动了手动提交offset(ack.acknowledge())
4.所以直接使用线程池消费，但是不手动管理offset也是常见的错误
复制代码