如何在Kafka中使用多线程：基于Java ExecutorService实现消息并行读取

核心思路 #

Kafka Consumer本身的poll()方法是单线程拉取消息的，如果直接在消费线程里处理耗时的业务逻辑，很容易导致消息堆积。我们可以让主线程（消费线程）只负责拉取消息，然后把每条消息提交给ExecutorService线程池，由线程池中的线程异步处理消息，这样消费线程可以持续拉取下一批消息，提升整体处理效率。

完整代码示例 #

首先确保你的项目引入了Kafka客户端依赖（以Maven为例）：

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>3.6.1</version> <!-- 推荐使用最新稳定版本 -->
</dependency>

接下来是实现代码：

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class KafkaMultiThreadedProcessor {
    private static final String TOPIC_NAME = "your-target-topic";
    private static final String BOOTSTRAP_SERVERS = "localhost:9092"; // 替换成你的Kafka地址
    private static final String GROUP_ID = "multi-thread-processing-group";
    private static final int THREAD_POOL_SIZE = 5; // 根据机器性能和消息量调整

    public static void main(String[] args) {
        // 1. 配置Kafka Consumer
        Properties consumerProps = new Properties();
        consumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
        consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, GROUP_ID);
        consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        // 开启手动提交Offset，确保消息处理完成后再提交，避免消息丢失
        consumerProps.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");

        // 2. 创建固定大小的线程池
        ExecutorService executorService = Executors.newFixedThreadPool(THREAD_POOL_SIZE);

        try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(consumerProps)) {
            consumer.subscribe(Collections.singletonList(TOPIC_NAME));

            while (true) {
                // 拉取消息，超时时间设为1秒
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));

                // 遍历拉取到的每条消息，提交给线程池处理
                for (ConsumerRecord<String, String> record : records) {
                    executorService.submit(new MessageHandler(record, consumer));
                }
            }
        } catch (Exception e) {
            System.err.println("Consumer遇到异常: " + e.getMessage());
            e.printStackTrace();
        } finally {
            // 3. 优雅关闭线程池：先停止接受新任务，等待已有任务完成
            executorService.shutdown();
            try {
                if (!executorService.awaitTermination(60, java.util.concurrent.TimeUnit.SECONDS)) {
                    // 超时后强制关闭
                    executorService.shutdownNow();
                }
            } catch (InterruptedException e) {
                executorService.shutdownNow();
            }
        }
    }

    // 消息处理类，实现Runnable接口，负责具体的业务逻辑
    private static class MessageHandler implements Runnable {
        private final ConsumerRecord<String, String> record;
        private final KafkaConsumer<String, String> consumer;

        public MessageHandler(ConsumerRecord<String, String> record, KafkaConsumer<String, String> consumer) {
            this.record = record;
            this.consumer = consumer;
        }

        @Override
        public void run() {
            try {
                // 这里替换成你的实际业务处理逻辑：比如解析消息、存储到数据库、调用外部接口等
                System.out.printf("线程[%s]处理消息：Key=%s, Value=%s, Partition=%d, Offset=%d%n",
                        Thread.currentThread().getName(), record.key(), record.value(), record.partition(), record.offset());
                
                // 模拟耗时处理（比如调用接口、IO操作）
                Thread.sleep(1000);

                // 处理完成后异步提交Offset，避免阻塞当前线程
                consumer.commitAsync((offsets, exception) -> {
                    if (exception != null) {
                        System.err.printf("Offset提交失败，偏移量：%s，异常信息：%s%n", offsets, exception.getMessage());
                    }
                });
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                System.err.println("消息处理被中断：" + e.getMessage());
            } catch (Exception e) {
                System.err.printf("处理消息失败，消息内容：%s，异常信息：%s%n", record.value(), e.getMessage());
                // 这里可以根据业务逻辑选择：重试处理、跳过消息并提交Offset、或者记录错误日志后告警
                // consumer.commitAsync();
            }
        }
    }
}

关键注意点 #

• 线程池选型：
  • 用newFixedThreadPool适合消息量稳定的场景，线程数固定，避免资源耗尽；
  • 如果消息量波动大，可以考虑newCachedThreadPool，但要通过ThreadPoolExecutor自定义最大线程数，防止OOM。
• Offset提交策略：
  • 一定要关闭自动提交，改用手动提交（commitAsync或commitSync），确保消息处理完成后再提交Offset，避免消息丢失；
  • commitAsync是异步提交，不会阻塞处理线程；如果需要严格的消息顺序，可使用commitSync同步提交，但会牺牲部分性能。
• Consumer线程安全：
  • Kafka Consumer不是线程安全的，绝对不能在多个线程中共享同一个Consumer实例！本示例中Consumer只在主线程操作拉取和提交Offset，线程池只处理消息，是安全的。
• 优雅关闭：
  • 程序退出时要先关闭线程池，等待所有正在处理的消息完成，再关闭Consumer，避免消息丢失。

内容的提问来源于stack exchange，提问作者Sanjiv