ContainerFactory

Kafka 消费者可以使用不同的消息监听器容器，例如 ConcurrentKafkaListenerContainerFactory、KafkaMessageListenerContainer 等。每个容器都提供了不同的功能和配置选项，可以根据实际需求进行选择和配置。 如果你需要自定义 Kafka 消费者的配置选项，可以通过在 Spring Boot 配置文件中设置属性来实现。另外一种方法是通过创建 KafkaListenerContainerFactory bean 并配置其属性来实现。

如何创建一个ConcurrentKafkaListenerContainerFactory并使用

▼
java
复制代码
@Configuration
// 在springboot项目中可以去除，但在spring项目中文件中启用了Kafka功能就必须加上
@EnableKafka
public class KafkaConfig {
    // 消费者config，此处为非springboot项目，所以需要单独构建
    @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, "my-group");
        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;
    }
    // 消费者工厂
    @Bean
    public ConsumerFactory<String, String> consumerFactory() {
        return new DefaultKafkaConsumerFactory<>(consumerConfigs());
    }
    
    
    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        factory.setConcurrency(3);
        factory.getContainerProperties().setPollTimeout(3000);
         //也可以设置为批量消费，每个批次数量在Kafka配置参数中设置ConsumerConfig.MAX_POLL_RECORDS_CONFIG
        //factory.setBatchListener(true);

        return factory;
    }
 
}

在上面的示例中，我们创建了一个 ConcurrentKafkaListenerContainerFactory 的 bean，并配置了一些属性，例如：

• 使用 DefaultKafkaConsumerFactory 作为消费者工厂；
• 设置并发消费者数量为 3；
• 设置轮询超时时间为 3000 毫秒。

想要使用上述创建的Kafka消费者工厂，只需要为其指定containerFactory即可

▼
java
复制代码
// Kafka开启事务的标识
@Transactional(rollbackFor = Exception.class)
@GetMapping("/hello1")
public String hello1() {
    // 发送消息
    kafkaTemplate.send("topic3", "3");
    return "hello1";
}
// 使用containerFactory
// 这里的congtainerFactory已经设置好了groupid，如果containerFactory中没有设置groupid这里记得设置
@KafkaListener(topics = "topic3",containerFactory = "kafkaListenerContainerFactory")
public void processMessage(String message) {
    // 处理消息
    log.info("hello1 >>>>> {}", message);
}

我们可以通过containerFactory应用于不同的消费者方法和主题，以满足不同的需求。

创建多个消费者工厂

常见的使用场景如下：

▼
java
复制代码
@Configuration
public class KafkaConfig {
    
    // 场景1：需要回复的消息（Request-Reply）
    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, String> replyFactory(
            ConsumerFactory<String, String> consumerFactory,
            @Lazy KafkaTemplate<String, String> kafkaTemplate) {
        var factory = new ConcurrentKafkaListenerContainerFactory<String, String>();
        factory.setConsumerFactory(consumerFactory);
        // 如果要用到SendTo注解，就必须set这个属性
        factory.setReplyTemplate(kafkaTemplate);  // 设置回复模板
        return factory;
    }
    
    // 场景2：需要手动提交偏移量
    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, String> manualFactory(
            ConsumerFactory<String, String> consumerFactory) {
        var factory = new ConcurrentKafkaListenerContainerFactory<String, String>();
        factory.setConsumerFactory(consumerFactory);
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL);
        return factory;
    }
    
    // 场景3：需要并发消费
    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, String> concurrentFactory(
            ConsumerFactory<String, String> consumerFactory) {
        var factory = new ConcurrentKafkaListenerContainerFactory<String, String>();
        factory.setConsumerFactory(consumerFactory);
        factory.setConcurrency(3);  // 3个并发消费者
        return factory;
    }
    
    // 4. repliesContainer 被 ReplyingKafkaTemplate 内部使用
    // 用于接收 Request-Reply 模式中的回复消息
    @Bean
    public ConcurrentMessageListenerContainer<String,String> repliesContainer(
            ConcurrentKafkaListenerContainerFactory<String, String> containerFactory) {

        ConcurrentMessageListenerContainer<String, String> replies =
                containerFactory.createContainer( "hello3_replies");
        replies.getContainerProperties().setGroupId("hello3_replies_group");
        replies.setAutoStartup(Boolean.FALSE);
        return replies;
    }
    
    
}

优先级规则

▼
text
复制代码
1. @KafkaListener 指定了 containerFactory 属性
   ↓ 使用指定的 factory
   
2. 没有指定 containerFactory
   ↓ 查找名为 "kafkaListenerContainerFactory" 的 Bean
   
3. 没有自定义的 kafkaListenerContainerFactory
   ↓ 使用 Spring Boot 自动配置的默认 factory

参考：https://mp.weixin.qq.com/s?__biz=MzU4NDcxOTM3OQ==&mid=2247485514&idx=1&sn=546b5e6a5fcc5c0052bb63238257516b&chksm=fd94c536cae34c207bf7abd5a514aff299122d6b68e671fdbb52693cded409b6c8a2b21c1f14&cur_album_id=2829784472496734210&scene=189#wechat_redirect