提高kafka消费速度之从源码去了解Spring-kafka的concurrency参数

第一部分、引言

在spring应用中,如果我们需要订阅kafka消息,通常情况下我们不会直接使用kafka-client,而是使用了更方便的一层封装spring-kafka。特别是在springboot微服务中,基于注解和配置的spring-kafka可以给我们带来更简单更便捷的开发方式。不过,它不仅仅只是简单的封装的kafka-client,仔细看他的源码会发现里面大有文章,不得不惊呼大神写这个框架的牛逼。

本文介绍了常见的几种在spring-kafka中提高kafka消费速度的方式,并重点介绍了使用多线程并发消费的方式,通过解读源码真正理解该多线程消费模型。

注意:本文涉及到的case是在partition数大于consumer实例数的情况。比如现在有5台部署了consumer实例机器,要去消费30个partition,是这种情况下的提升消费速度优化。
如果你现在面临的情况是那种partition数很少,又受限于“一个partition只能对应一个消费线程”的限制。比如partition数只有1,这种情况就必须要“把拉取动作和处理动作分开” 可以参考以下模型 —— 消息拉下来之后就丢给线程池handle并自动提交offset,也可以让消息在handler完之后手动提交offset(无论哪种方式都是不安全的,建议参考类似于TCP的滑动窗口控制的方式来控制消费位移的提交)

第二部分:提高消费速度的几种操作

kakfa是我们在项目开发中经常使用的消息中间件。由于它的写性能非常高,因此,经常会碰到Kafka消息队列拥堵的情况。遇到这种情况时,有时我们不能直接清理整个topic,因为还有别的服务正在使用该topic。因此只能额外启动一个或多个相同名称的consumer-group的消费者实例来加快消息消费(如果该topic只有一个partition,实际上再启动一个新的消费者,没有作用)。——这是最原始的提高消费速度的方式。
然后我们介绍一下引入spring-kafka有哪些操作(其实都是可以用代码实现,只是你要自己做这层二次封装)

一、用多线程并发消费

通过设置concurrency参数的方式。先看代码,我们可以使用两种不同的途径设置该参数:
第一种方式,直接在factory里面设置。

我们给ConcurrentKafkaListenerContainerFactory设置了concurrency等于3,也可以通过在application.properties中添加spring.kafka.listener.concurrency=3的方式配置factory
@Bean
KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        factory.setConcurrency(3);
        factory.setBatchListener(true);
        factory.getContainerProperties().setPollTimeout(3000);
        return factory;
 }
本人不推荐使用该方式,因为设置它的意思是给factory里面的每个listener都设置了3个线程,但其实有些listener监听的topic并没有那么多分区。推荐用第二种方式。

第二种方式,在@KafkaListener设置,支持SpEL表达式。

单独给某个listener配置并发线程数,这种方式在逻辑上更为合理。
@KafkaListener(topics = {"${kafka.calculate.topic}"}, concurrency = "3")
    public void listen(ConsumerRecord<String, String> record) {
        // doing
    }

先介绍一下spring-kafka在工作的时候启动的一堆线程:
分为两类线程,一类是Consumer线程,另一类是Listener线程
Consumer线程: 用来直接调用kafka-client的poll()方法获取消息。
Listener线程: 真正调用处理我们代码中标有@KafkaListener注解方法的线程。
如果不使用spring-kafka,而是直接用kafka-client的话,那么正常我们会整一个while循环,在循环里面调用poll(),然后处理消息,这在kafka broker看来就是一个Consumer。如果你想用多个Consumer, 除了多启动几个进程以外,也可以在一个进程使用多个线程执行此while()循环。spring-kafka就是这么干的。

因此,先看结论:
对于concurrency=3这个参数的值的设定来说,它设置的其实是每个@KafkaListener的并发数。spring-kafka在初始化的时候会启动concurrencyConsumer线程来执行@KafkaListener里面的方法。这里如果你是用第一种方式,在factory里面直接设置concurrency,那么每个加了@KafkaListener注解的都会新建concurrency个线程,这样如果listener订阅的topic没那么多分区,new那么多线程只会带来格外的性能开销,这样是我不推荐在factory指定concurrency的原因。

最后一部分将会带大家阅读源码证明以上这段话

另外:
看到这里可能会问,那我说的Listener线程没用到啊。不急,继续往下看,在源码中已经给出了:

protected void pollAndInvoke() {
            if (!this.autoCommit && !this.isRecordAck) {
                processCommits();
            }
            processSeeks();
            checkPaused();
            ConsumerRecords<K, V> records = this.consumer.poll(this.pollTimeout);
            this.lastPoll = System.currentTimeMillis();
            checkResumed();
            debugRecords(records);
            if (records != null && records.count() > 0) {
                if (this.containerProperties.getIdleEventInterval() != null) {
                    this.lastReceive = System.currentTimeMillis();
                }
                // 这里可以看到如果是自动提交offset,会直接把consumer poll下来的消息给到listener执行,
// 即kafka consumer所在线程会直接调用我们的@KafkaListener方法
                invokeListener(records);
            }
            else {
                checkIdle();
            }
        }
如果是手动提交offset,即enable-auto-commit设置为false,则是将消息投放到阻塞队列中,另一边由Listener线程取出执行。

所以,当concurrency=3,自动提交设置为false时,如果你程序里有两个方法标记了@KafkaListener,那么此时会启动 2 * 3 = 6 个Consumer线程,6个Listener线程。
这个信息在排查错误的时候非常重要,但官方文档居然没怎么提线程的事(不够详细),只是在介绍KafkaContainerListener

二、批量消费

涉及到两个参数,

  • 一个是factory.setBatchListener(true) —— 启动批量消费,
  • 一个是propsMap.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 50) —— 设置批量消费每次最多消费多少条记录。

官方文档对max.poll.records的定义是:
这个参数定义了poll()方法最多可以返回多少条消息,默认值为500。注意这里的用词是"最多",也就是说如果在拉取消息的时候新消息不足500条,那有多少返回多少;如果超过500条,就只返回500。这个默认值是比较坑人的,如果你的消息处理逻辑比较重,比如需要查数据库,调用接口,甚至是复杂计算,那么你很难保证能够在max.poll.interval.ms内处理完500条消息,也就是说,如果上游真的突然大爆发生产了成千上万条消息,而平摊到每个消费者身上的消息达到了500的又无法按时消费完成的话,会触发消费者实例的rebalance, 然后这批消息会被分配到另一个消费者中,还是会处理不完,又会触发rebalance, 这样这批消息就永远也处理不完,而且一直在重复处理。

所以在配置的时候要避免出现上述问题,可以提前评估好处理一条消息最长需要多少时间,然后务必覆盖默认的max.poll.records参数。在spring-kafka中这个原生参数对应的参数项是max-poll-records。对于消息处理比较重的操作,建议把这个值改到50以下会保险一些。

配置代码如下:

@Bean
public Map<String, Object> consumerConfigs() {
        Map<String, Object> propsMap = new HashMap<>();
        propsMap.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, propsConfig.getBroker());
        propsMap.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, propsConfig.getEnableAutoCommit());
        propsMap.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "100");
        propsMap.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "15000");
        propsMap.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        propsMap.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        propsMap.put(ConsumerConfig.GROUP_ID_CONFIG, propsConfig.getGroupId());
        propsMap.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, propsConfig.getAutoOffsetReset());
        // 这里设置批量消费的消息个数
    propsMap.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 50);
        return propsMap;
    }
@Bean
public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        factory.setConcurrency(4);
    // 记得要开启批量消费,不然只是单次poll的消息增多而已
        factory.setBatchListener(true);
        factory.getContainerProperties().setPollTimeout(3000);
        return factory;
    }

三、分区消费

这种方式我个人觉得不是很适用分布式部署的情况,所以基本上不会考虑用它。。具体操作代码如下:

@KafkaListener(id = "id0", topicPartitions = { @TopicPartition(topic = TPOIC, partitions = { "0" }) })
    public void listenPartition0(List<ConsumerRecord<?, ?>> records) {
        log.info("Id0 Listener, Thread ID: " + Thread.currentThread().getId());
        log.info("Id0 records size " +  records.size());

        for (ConsumerRecord<?, ?> record : records) {
           // doing
        }
    }

    @KafkaListener(id = "id1", topicPartitions = { @TopicPartition(topic = TPOIC, partitions = { "1" }) })
    public void listenPartition1(List<ConsumerRecord<?, ?>> records) {
        log.info("Id1 Listener, Thread ID: " + Thread.currentThread().getId());
        log.info("Id1 records size " +  records.size());

        for (ConsumerRecord<?, ?> record : records) {
            // doing
        }

第三部分:从源码去了解Spring-kafka的concurrency参数

直接找到我们的ConcurrentKafkaListenerContainerFactory类中,concurrency的使用:

@Override
protected void initializeContainer(ConcurrentMessageListenerContainer<K, V> instance, KafkaListenerEndpoint endpoint) {
        super.initializeContainer(instance, endpoint);
        if (endpoint.getConcurrency() != null) {
            instance.setConcurrency(endpoint.getConcurrency());
        }
        else if (this.concurrency != null) {
            instance.setConcurrency(this.concurrency);
        }
    }

继续在ConcurrentMessageListenerContainer中我们找到了concurrency的使用:

@Override
protected void doStart() {
        if (!isRunning()) {
            checkTopics();
            ContainerProperties containerProperties = getContainerProperties();
            TopicPartitionInitialOffset[] topicPartitions = containerProperties.getTopicPartitions();
            // 这里你会发现如果你设置的concurrency 大于分区数,spring-kafka也会让它等于分区的个数,肯定不会超过分区数。
            // 可这是在一个JVM里面的concurrency,如果你是多个消费者实例部署在多个服务器上(实际生产都是这么做),**那么你的concurrency 值记得还要除以机器数**。
            if (topicPartitions != null && this.concurrency > topicPartitions.length) {
                this.logger.warn("When specific partitions are provided, the concurrency must be less than or "
                        + "equal to the number of partitions; reduced from " + this.concurrency + " to "
                        + topicPartitions.length);
                this.concurrency = topicPartitions.length;
            }
            setRunning(true);
            // 这里可以很清晰的看到,concurrency是指KafkaMessageListenerContainer的个数,即concurrency表明的是spring会创建多个KafkaMessageListenerContainer。
            // 那么KafkaMessageListenerContainer类又干了什么?这里是for循环顺序创建了concurrency个KafkaMessageListenerContainer,那么spring又是如何并发的?这里会猜想是new了线程去处理
            for (int i = 0; i < this.concurrency; i++) {
                KafkaMessageListenerContainer<K, V> container;
                if (topicPartitions == null) {
                    container = new KafkaMessageListenerContainer<>(this, this.consumerFactory, containerProperties);
                }
                else {
                    container = new KafkaMessageListenerContainer<>(this, this.consumerFactory,
                            containerProperties, partitionSubset(containerProperties, i));
                }
                String beanName = getBeanName();
                container.setBeanName((beanName != null ? beanName : "consumer") + "-" + i);
                if (getApplicationEventPublisher() != null) {
                    container.setApplicationEventPublisher(getApplicationEventPublisher());
                }
                container.setClientIdSuffix("-" + i);
                container.setGenericErrorHandler(getGenericErrorHandler());
                container.setAfterRollbackProcessor(getAfterRollbackProcessor());
                container.setEmergencyStop(() -> {
                    stop(() -> {
                        // NOSONAR
                    });
                    publishContainerStoppedEvent();
                });
                container.start();
                this.containers.add(container);
            }
        }
    }

继续往下看KafkaMessageListenerContainer干了什么?
进入41行的start方法:这是父类AbstractMessageListenerContainer的start方法

@Override
public final void start() {
        checkGroupId();
        synchronized (this.lifecycleMonitor) {
            if (!isRunning()) {
                Assert.isTrue(this.containerProperties.getMessageListener() instanceof GenericMessageListener,
                        () -> "A " + GenericMessageListener.class.getName() + " implementation must be provided");
                doStart();
            }
        }
    }

再进入第8行的doStart方法:
发现doStart是一个抽象方法,于是我们需要回到AbstractMessageListenerContainer的子类KafkaMessageListenerContainer中,查看doStart方法具体做了什么:

@Override
protected void doStart() {
        if (isRunning()) {
            return;
        }
        if (this.clientIdSuffix == null) { // stand-alone container
            checkTopics();
        }
        ContainerProperties containerProperties = getContainerProperties();
        // 这里是做一些参数校验
        checkAckMode(containerProperties);

        Object messageListener = containerProperties.getMessageListener();
        Assert.state(messageListener != null, "A MessageListener is required");
        if (containerProperties.getConsumerTaskExecutor() == null) {
            SimpleAsyncTaskExecutor consumerExecutor = new SimpleAsyncTaskExecutor(
                    (getBeanName() == null ? "" : getBeanName()) + "-C-");
            containerProperties.setConsumerTaskExecutor(consumerExecutor);
        }
        Assert.state(messageListener instanceof GenericMessageListener, "Listener must be a GenericListener");
        // 设置消息来到之后,需要回调的listener和listenerType类型
        GenericMessageListener<?> listener = (GenericMessageListener<?>) messageListener;
        ListenerType listenerType = deteremineListenerType(listener);
        // 这里创建了一个ListenerConsumer,看到这里我们也明白了为什么顺序创建了concurrency个KafkaMessageListenerContainer,却可以做到并发?
        // 理由很简单:因为在KafkaMessageListenerContainer类内部中,对于ListenerConsumer的处理逻辑,是新起线程执行,所以才可以做到并发。
        // 这里又一次说明了设置好concurrency值的重要性,如果处理不好,会new过多闲置的ListenerConsumer,因为一个partition只对应一个消费线程。
        this.listenerConsumer = new ListenerConsumer(listener, listenerType);
        setRunning(true);
        this.listenerConsumerFuture = containerProperties
                .getConsumerTaskExecutor()
                .submitListenable(this.listenerConsumer);
    }

继续,我们进入ListenerConsumer的构造函数里面:

ListenerConsumer(GenericMessageListener<?> listener, ListenerType listenerType) {
            Assert.state(!this.isAnyManualAck || !this.autoCommit,
                    () -> "Consumer cannot be configured for auto commit for ackMode "
                            + this.containerProperties.getAckMode());
            // 这里可以看到每个ListenerConsumer首先创建了一个Kafka定义的Consumer,然后,设置了consumer的订阅的topic、分区偏移量信息以及重新分配分区的监听对象。
            this.consumer = KafkaMessageListenerContainer.this.consumerFactory.createConsumer(
                            this.consumerGroupId,
                            this.containerProperties.getClientId(),
                            KafkaMessageListenerContainer.this.clientIdSuffix,
                            this.containerProperties.getConsumerProperties());

            if (this.transactionManager != null) {
                this.transactionTemplate = new TransactionTemplate(this.transactionManager);
            }
            else {
                this.transactionTemplate = null;
            }
            subscribeOrAssignTopics(this.consumer);
            GenericErrorHandler<?> errHandler = KafkaMessageListenerContainer.this.getGenericErrorHandler();
            this.genericListener = listener;
            if (listener instanceof BatchMessageListener) {
                this.listener = null;
                this.batchListener = (BatchMessageListener<K, V>) listener;
                this.isBatchListener = true;
                this.wantsFullRecords = this.batchListener.wantsPollResult();
            }
            else if (listener instanceof MessageListener) {
                this.listener = (MessageListener<K, V>) listener;
                this.batchListener = null;
                this.isBatchListener = false;
                this.wantsFullRecords = false;
            }
            else {
                throw new IllegalArgumentException("Listener must be one of 'MessageListener', "
                        + "'BatchMessageListener', or the variants that are consumer aware and/or "
                        + "Acknowledging"
                        + " not " + listener.getClass().getName());
            }
            this.listenerType = listenerType;
            this.isConsumerAwareListener = listenerType.equals(ListenerType.ACKNOWLEDGING_CONSUMER_AWARE)
                    || listenerType.equals(ListenerType.CONSUMER_AWARE);
            if (this.isBatchListener) {
                validateErrorHandler(true);
                this.errorHandler = new LoggingErrorHandler();
                this.batchErrorHandler = determineBatchErrorHandler(errHandler);
            }
            else {
                validateErrorHandler(false);
                this.errorHandler = determineErrorHandler(errHandler);
                this.batchErrorHandler = new BatchLoggingErrorHandler();
            }
            Assert.state(!this.isBatchListener || !this.isRecordAck,
                    "Cannot use AckMode.RECORD with a batch listener");
            if (this.containerProperties.getScheduler() != null) {
                this.taskScheduler = this.containerProperties.getScheduler();
                this.taskSchedulerExplicitlySet = true;
            }
            else {
                ThreadPoolTaskScheduler threadPoolTaskScheduler = new ThreadPoolTaskScheduler();
                threadPoolTaskScheduler.initialize();
                this.taskScheduler = threadPoolTaskScheduler;
            }
            this.monitorTask = this.taskScheduler.scheduleAtFixedRate(this::checkConsumer,
                    this.containerProperties.getMonitorInterval() * 1000); // NOSONAR magic #
            if (this.containerProperties.isLogContainerConfig()) {
                this.logger.info(this);
            }
            Map<String, Object> props =
                    KafkaMessageListenerContainer.this.consumerFactory.getConfigurationProperties();
            this.checkNullKeyForExceptions = checkDeserializer(findDeserializerClass(props, false));
            this.checkNullValueForExceptions = checkDeserializer(findDeserializerClass(props, true));
        }

看到这里差不多真相大白了,在梳理一遍:

ConcurrentKafkaListenerContainerFactory类中的concurrency设置了KafkaMessageListenerContainer对象创建的个数,每个KafkaMessageListenerContainer对象创建了一个ListenerConsumer对象,ListenerConsumer对象有封装了一个kafkaConsumer对象。所以concurrency最终设置的是kafkaConsumer对象的个数。这个也和ConcurrentKafkaListenerContainerFactory类中setConcurrency方法的注释是一致的。

总结

调大concurrency的值,真的会提高系统消息消费的能力吗?

此时问题可以理解为:增加kafkaConsumer可以提高系统消息消费能力吗?答案很明显:不一定。因为我们知道,kafka中一个consumer最多消费一个分区,在topic分区数量一定的情况下,如果consumer的数量小于分区数量,那么增加consumer是可以提高系统消息消费能力的。如果consumer的数量大于等于分区数量,那么此时一味提高consumer数量,系统的消息消费能力是不会提高的,反而还有可能下降。这是因为每次rebalance操作,所有的consumer都会参与,当consumer数量大时,rebalance的耗时会增加,因此系统的性能会有所下降。

©著作权归作者所有,转载或内容合作请联系作者
  • 序言:七十年代末,一起剥皮案震惊了整个滨河市,随后出现的几起案子,更是在滨河造成了极大的恐慌,老刑警刘岩,带你破解...
    沈念sama阅读 198,082评论 5 464
  • 序言:滨河连续发生了三起死亡事件,死亡现场离奇诡异,居然都是意外死亡,警方通过查阅死者的电脑和手机,发现死者居然都...
    沈念sama阅读 83,231评论 2 375
  • 文/潘晓璐 我一进店门,熙熙楼的掌柜王于贵愁眉苦脸地迎上来,“玉大人,你说我怎么就摊上这事。” “怎么了?”我有些...
    开封第一讲书人阅读 145,047评论 0 327
  • 文/不坏的土叔 我叫张陵,是天一观的道长。 经常有香客问我,道长,这世上最难降的妖魔是什么? 我笑而不...
    开封第一讲书人阅读 52,977评论 1 268
  • 正文 为了忘掉前任,我火速办了婚礼,结果婚礼上,老公的妹妹穿的比我还像新娘。我一直安慰自己,他们只是感情好,可当我...
    茶点故事阅读 61,893评论 5 359
  • 文/花漫 我一把揭开白布。 她就那样静静地躺着,像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上,一...
    开封第一讲书人阅读 47,014评论 1 275
  • 那天,我揣着相机与录音,去河边找鬼。 笑死,一个胖子当着我的面吹牛,可吹牛的内容都是我干的。 我是一名探鬼主播,决...
    沈念sama阅读 36,976评论 3 388
  • 文/苍兰香墨 我猛地睁开眼,长吁一口气:“原来是场噩梦啊……” “哼!你这毒妇竟也来了?” 一声冷哼从身侧响起,我...
    开封第一讲书人阅读 35,605评论 0 254
  • 序言:老挝万荣一对情侣失踪,失踪者是张志新(化名)和其女友刘颖,没想到半个月后,有当地人在树林里发现了一具尸体,经...
    沈念sama阅读 39,888评论 1 293
  • 正文 独居荒郊野岭守林人离奇死亡,尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 34,906评论 2 314
  • 正文 我和宋清朗相恋三年,在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 36,732评论 1 328
  • 序言:一个原本活蹦乱跳的男人离奇死亡,死状恐怖,灵堂内的尸体忽然破棺而出,到底是诈尸还是另有隐情,我是刑警宁泽,带...
    沈念sama阅读 32,513评论 3 316
  • 正文 年R本政府宣布,位于F岛的核电站,受9级特大地震影响,放射性物质发生泄漏。R本人自食恶果不足惜,却给世界环境...
    茶点故事阅读 37,980评论 3 301
  • 文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹,春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 29,132评论 0 19
  • 文/苍兰香墨 我抬头看了看天上的太阳。三九已至,却和暖如春,着一层夹袄步出监牢的瞬间,已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 30,447评论 1 255
  • 我被黑心中介骗来泰国打工, 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留,地道东北人。 一个月前我还...
    沈念sama阅读 42,027评论 2 343
  • 正文 我出身青楼,却偏偏与公主长得像,于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子,可洞房花烛夜当晚...
    茶点故事阅读 41,232评论 2 339

推荐阅读更多精彩内容