当前位置：首页 > article >正文

Kafka 之消息并发消费

article 2024/11/7 9:56:19

前言：

我们知道 Kafka 以其高性能著称，Kafka 的性能极其强大，那在 Java 应用中，我们怎么来利用其 Kafka 的高性能呢？

Kafka 系列文章传送门

Kafka 简介及核心概念讲解

Spring Boot 整合 Kafka 详解

Kafka @KafkaListener 注解的详解及使用

Kafka 客户端工具使用分享【offsetexplorer】

Kafka 之消息同步/异步发送

Kafka 之批量消息发送消费

Kafka 之消息广播消费

Kafka 之消息并发消费

我们使用 @KafkaListener 进行消息消费的时候，默认是单线程消费的，当生产者数据量很大的时候，单线程消费很容器导致消费堆积，浪费了 Kafka 的优异性能，我们在开发 Java 应用程序的时候，如果发现单线程效率低不能满足业务场景的时候，我们自然而然的想到了使用多线程来处理，同样 Kafka 是支持多线程消费的，而且 spring-cloud-starter-stream-kafka 已经帮我们实现好了，我们只需要按需配置即可。

我们在来看看 @KafkaListener 注解，该注解有一个属性叫 concurrency，该属性可以指定并发小消费的现成数量，我们可以进行如下配置：

@KafkaListener(id = "my-kafka-consumer",
            groupId = "my-kafka-consumer-groupId",
            topics = "my-topic",
            concurrency = "3",
            containerFactory = "myContainerFactory")

concurrency = “3”，表示启动三个线程来消费，接下来我们来演示一下 Kafka 的批量消费。

Kafka 并发消费之 Producer 代码

生产者代码还是没有什么特殊，我写了一个 for 循环来发送 100 条消息，如下：

package com.order.service.kafka.producer;

import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

import java.text.SimpleDateFormat;
import java.util.Date;

/**
 * @ClassName： MyKafkaConcurrencyProducer 
 * @Author: Author
 * @Date: 2024/10/22 19:22
 * @Description:
 */
@Slf4j
@Component
public class MyKafkaConcurrencyProducer {

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    public void batchSendMessage() {
        for (int a = 0; a < 100; a++) {
            this.kafkaTemplate.send("my-topic", "第" + a + "条 kafka 消息");
        }
    }

}

Kafka 并发消费之消费者代码

并发消费我们还是使用 @KafkaListener 注解来完成消息消费监听，具体代码如下：

package com.order.service.kafka.consumer;

import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.text.SimpleDateFormat;
import java.util.Date;

/**
 * @ClassName： MyKafkaConcurrencyConsumer
 * @Author: zhangyong
 * @Date: 2024/10/22 19:22
 * @Description:
 */
@Slf4j
@Component
public class MyKafkaConcurrencyConsumer {

    @KafkaListener(id = "my-kafka-concurrency-consumer",
            groupId = "my-kafka-consumer-groupId",
            topics = "my-topic",
            concurrency = "3",
            containerFactory = "myContainerFactory")
    public void listen(String message) {
        log.info("消息消费成功,线程ID:{},消息内容:{}", Thread.currentThread().getId(), message);
    }

}

注意这里相交前面分享的案例，我们在注解的属性上加了 concurrency = “3”，正是这个属性才能放 Kafka 完成并发消费。

Kafka 并发消费之结果验证

我们触发消息发送，控制台部分日志如下：

2024-10-27 18:10:12.089  INFO 7308 --- [-consumer-2-C-1] c.o.s.k.c.MyKafkaConcurrencyConsumer     : 消息消费成功,线程ID:146,消息内容:第12条 kafka 消息
2024-10-27 18:10:12.097  INFO 7308 --- [-consumer-1-C-1] c.o.s.k.c.MyKafkaConcurrencyConsumer     : 消息消费成功,线程ID:144,消息内容:第46条 kafka 消息
2024-10-27 18:10:12.100  INFO 7308 --- [-consumer-2-C-1] c.o.s.k.c.MyKafkaConcurrencyConsumer     : 消息消费成功,线程ID:146,消息内容:第13条 kafka 消息
2024-10-27 18:10:12.123  INFO 7308 --- [-consumer-0-C-1] c.o.s.k.c.MyKafkaConcurrencyConsumer     : 消息消费成功,线程ID:142,消息内容:第68条 kafka 消息
2024-10-27 18:10:12.132  INFO 7308 --- [-consumer-1-C-1] c.o.s.k.c.MyKafkaConcurrencyConsumer     : 消息消费成功,线程ID:144,消息内容:第47条 kafka 消息

通过控制台日志可以看到，线程 ID 分别有 144、146、142，表示确实有多个线程在进行消息监听，结果符合预期。

Kafka 并发消费的时候指定多个个线程合适？

在进行 Kafka 消息并发消费的时候，并不是设定的线程数量越多，效率就越高，线程数的多少跟 Kafka 的 Partition 是有直接关系的，一个 Partition 只能被一个线程消费，因此我们最多将并发线程数量和 Kafka Topic 的 Partition 数量设置一致，或者不要超过 Kafka Topic 的 Partion 的数量，下面我们来分析一下线程数量设置不合理会有什么情况发生。