Kafka配置中的线程数如何设置

Kafka配置中的线程数主要涉及服务端和客户端，具体设置如下：

一、服务端线程配置（server.properties）

1. 网络请求处理线程

   • num.network.threads：处理网络请求的线程数，建议设置为CPU核数的2-3倍。
   • 示例：num.network.threads=8（8核服务器可设为8-12）。

2. 磁盘IO处理线程

   • num.io.threads：处理磁盘读写的线程数，建议设置为CPU核数的50%，不超过总核数。
   • 示例：num.io.threads=4（8核服务器可设为4-6）。

3. 分区恢复线程

   • num.recovery.threads.per.data.dir：每个数据目录的恢复线程数，通常设置为1-3，用于日志恢复和清理。
   • 示例：num.recovery.threads.per.data.dir=2。

二、客户端线程配置

1. 生产者多线程

   • 通过线程池控制发送线程数，无直接配置参数，需自行实现多线程逻辑，例如：

     ExecutorService producerExecutor = Executors.newFixedThreadPool(10); // 10个生产者线程  
     producerExecutor.submit(() -> {  
         // 生产者发送逻辑  
     });  
     

2. 消费者多线程

   • 旧版API：通过createMessageStreams的topicCountMap设置每个主题的消费线程数，需确保线程数≤分区数。

     Map<String, Integer> topicCountMap = new HashMap<>();  
     topicCountMap.put("topic1", 3); // 3个线程消费topic1  
     

   • 新版API（推荐）：使用ConcurrentKafkaListenerContainerFactory的concurrency参数，需与分区数匹配。

     @Bean  
     public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {  
         ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();  
         factory.setConcurrency(3); // 每个消费者实例的线程数（不超过分区数）  
         return factory;  
     }  
     

三、关键原则

• 服务端：线程数需与硬件资源（CPU/磁盘）匹配，避免过度配置导致资源竞争。
• 客户端：消费者线程数需≤分区数，确保并行消费效率；生产者线程数根据吞吐量需求调整，需注意缓冲区大小（buffer.memory）。
• 动态调整：可通过监控（如JMX）观察线程使用情况，动态优化参数。

参数来源：