KafkaListener 指定 ConcurrentKafkaListenerContainerFactory

KafkaListener 注解可以指定使用哪个 ConcurrentKafkaListenerContainerFactory，例如：

@KafkaListener(
  topics = "myTopic",
  containerFactory = "myCustomKafkaListenerContainerFactory")
public void listen(String message) {
  // 处理消息逻辑
}

其中，`myCustomKafkaListenerContainerFactory` 是一个由 `ConcurrentKafkaListenerContainerFactory` 实现类创建的实例，用于控制 KafkaListener 的并发处理方式。通过自定义这个工厂类，我们可以配置 KafkaListener 的并发度、消费者组、批量处理等属性。

相关问题

基于 ConcurrentKafkaListenerContainerFactory 创建 KafkaListener

基于 ConcurrentKafkaListenerContainerFactory 创建 KafkaListener 的步骤如下：

1. 创建一个 KafkaListenerContainerFactory 实例，可以使用 ConcurrentKafkaListenerContainerFactory。

2. 设置 KafkaListenerContainerFactory 的属性，例如：bootstrap.servers、group.id、auto.offset.reset 等。

3. 创建一个 KafkaListener 方法，并使用 @KafkaListener 注解来指定监听的 topic。

4. 在方法中处理接收到的消息。

示例代码如下：

@Configuration
@EnableKafka
public class KafkaListenerConfig {

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        factory.setConcurrency(3);
        factory.getContainerProperties().setPollTimeout(3000);
        return factory;
    }

    @Bean
    public ConsumerFactory<String, String> consumerFactory() {
        return new DefaultKafkaConsumerFactory<>(consumerConfigs());
    }

    @Bean
    public Map<String, Object> consumerConfigs() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "group-id");
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        return props;
    }

    @KafkaListener(topics = "test-topic")
    public void listen(ConsumerRecord<String, String> record) {
        System.out.println("Received message: " + record.value());
    }
}

在以上示例代码中，我们创建了一个 ConcurrentKafkaListenerContainerFactory 实例，并设置了其属性。

然后，我们创建了一个 KafkaListener 方法，使用 @KafkaListener 注解来指定监听的 topic，当接收到消息时，会执行该方法。

最后，在 KafkaListener 方法中，我们处理接收到的消息。

@KafkaListener中的containerFactory、groupId参数

`containerFactory`是Spring Kafka提供的用于创建Kafka消息监听容器的工厂类。通过使用不同的工厂类，我们可以创建不同类型的消息监听容器。例如，我们可以使用`ConcurrentKafkaListenerContainerFactory`创建一个支持多线程并发消费的监听容器。

`groupId`是消费者组的标识符。Kafka中的消费者可以组成消费者组，每个消费者组内的消费者共同消费一个或多个主题的消息，并且每个消息只会被同一个消费者组内的一个消费者消费一次。`groupId`参数用于指定当前消费者所属的消费者组。如果不指定`groupId`，则默认使用空字符串作为组标识符，这样所有的消费者会被认为是同一个组的成员，从而导致重复消费的问题。