Spring中如何通过多线程@KafkaListener消费单个Kafka Topic？

你说的完全没错！通过ConcurrentKafkaListenerContainerFactory.setConcurrency(3)来配置消费线程数，就是Spring Kafka中实现单个Topic多线程消费的标准且正确的方式。

这个concurrency参数的作用是指定当前监听器容器的并发消费线程数量，每个线程都会对应一个独立的KafkaConsumer实例，它们会共同订阅目标Topic。Kafka的分区分配机制会自动把Topic的分区均匀分配给这些线程（同一个消费者组内），从而实现多线程并行消费。

具体配置示例 #

1. 配置ConcurrentKafkaListenerContainerFactory

@Configuration
public class KafkaConsumerConfig {

    @Value("${spring.kafka.bootstrap-servers}")
    private String bootstrapServers;

    @Value("${spring.kafka.consumer.group-id}")
    private String consumerGroupId;

    @Bean
    public ConsumerFactory<String, String> consumerFactory() {
        Map<String, Object> configProps = new HashMap<>();
        configProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        configProps.put(ConsumerConfig.GROUP_ID_CONFIG, consumerGroupId);
        configProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        configProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        return new DefaultKafkaConsumerFactory<>(configProps);
    }

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, String> factory =
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        // 设置并发消费线程数为3
        factory.setConcurrency(3);
        // 可选：根据业务需求调整ACK模式，比如手动提交偏移量
        // factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL);
        return factory;
    }
}

2. 编写@KafkaListener监听器

@Component
public class SingleTopicConsumer {

    @KafkaListener(topics = "your-target-topic")
    public void consumeMessage(String message) {
        // 这里编写你的消息处理逻辑
        System.out.printf("线程[%s] 消费消息: %s%n", 
                          Thread.currentThread().getName(), message);
    }
}

几个关键注意点 #

• 分区数与并发数的匹配：Kafka的规则是同一个消费者组内，一个分区只能被一个线程消费。如果你的Topic只有2个分区，却设置concurrency=3，那么会有1个线程一直处于空闲状态，永远不会收到消息。所以建议concurrency的值不要超过目标Topic的分区数，这样每个线程都能分配到至少一个分区，充分利用资源。
• 消息顺序性：如果你的业务要求同一个分区内的消息必须严格顺序消费，这种方式完全没问题——因为每个分区只会被一个线程处理。但跨分区的消息，不同线程的处理顺序是不保证的。
• 消费者组的影响：所有这些消费线程都属于同一个消费者组（由group-id指定），所以不要和其他消费者组共享同一个group-id，否则会影响分区分配。

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