每个主题可以有一个 Kafka 消费者线程吗？

Question

我们有使用 Spring-Kafka (2.1.7) 的 Springboot 应用程序。我们启用了并发，因此每个分区可以有一个消费者线程。所以目前，如果我们有 3 个主题，每个主题有 2 个分区，就会有 2 个消费者线程，如下所示：

ConsumerThread1 - [topic1-0, topic2-0, topic3-0]
ConsumerThread2 - [topic1-1, topic2-1, topic3-1]

然而，不是每个分区一个 KafkaListener（或消费者线程），我们希望每个主题有一个消费者线程。例如：

ConsumerThread1 - [topic1-0, topic1-1]
ConsumerThread2 - [topic2-0, topic2-1]
ConsumerThread3 - [topic3-0, topic3-1]

如果这是不可能的，即使以下设置也可以：

ConsumerThread1 - [topic1-0]
ConsumerThread2 - [topic1-1]
ConsumerThread3 - [topic2-0]
ConsumerThread4 - [topic2-1]
ConsumerThread5 - [topic3-0]
ConsumerThread6 - [topic3-1]

问题是我们事先不知道完整的主题列表（我们使用通配符主题模式）。可以随时添加新主题，并且应该在运行时为这个新主题动态创建一个新的消费者线程（或多个线程）。

有什么办法可以实现吗？

Answer 1

您可以从spring-kafka:2.2为每个主题创建单独的容器并将并发设置为 1，这样每个容器都会从每个主题中消费

从版本 2.2 开始，您可以使用相同的工厂来创建任何 ConcurrentMessageListenerContainer。如果您想创建多个具有相似属性的容器，或者您希望使用一些外部配置的工厂，例如 Spring Boot 自动配置提供的工厂，这可能很有用。创建容器后，您可以进一步修改其属性，其中许多属性是使用 container.getContainerProperties() 设置的。以下示例配置 ConcurrentMessageListenerContainer：

@Bean
public ConcurrentMessageListenerContainer<String, String>(
    ConcurrentKafkaListenerContainerFactory<String, String> factory) {

ConcurrentMessageListenerContainer<String, String> container =
    factory.createContainer("topic1", "topic2");
container.setMessageListener(m -> { ... } );
return container;
}

注意：以这种方式创建的容器不会添加到端点注册表中。它们应该被创建为@Bean 定义，以便它们在应用程序上下文中注册。

Answer 2

感谢@Gary Russel的建议，我能够提出以下解决方案，为@KafkaListener每个 Kafka 主题创建一个 bean 实例（或消费者线程）。这样，如果属于特定主题的消息出现问题，就不会影响其他主题的处理。

注意- 以下代码InstanceAlreadyExistsException在启动期间引发异常。不过，这似乎并不影响功能。使用日志输出，我能够验证每个主题是否有一个 bean 实例（或线程），并且它们能够处理消息。

@SpringBootApplication
@EnableScheduling
@Slf4j
public class KafkaConsumerApp {

    public static void main(String[] args) {
        log.info("Starting spring boot KafkaConsumerApp..");
        SpringApplication.run(KafkaConsumerApp.class, args);
    }

}


@EnableKafka
@Configuration
public class KafkaConfiguration {

    private final KafkaProperties kafkaProperties;

    @Value("${kafka.brokers:localhost:9092}")
    private String bootstrapServer;

    @Value("${kafka.consumerClientId}")
    private String consumerClientId;

    @Value("${kafka.consumerGroupId}")
    private String consumerGroupId;

    @Value("${kafka.topicMonitorClientId}")
    private String topicMonitorClientId;

    @Value("${kafka.topicMonitorGroupId}")
    private String topicMonitorGroupId;

    @Autowired
    private ConfigurableApplicationContext context;

    @Autowired
    public KafkaConfiguration( KafkaProperties kafkaProperties ) {
        this.kafkaProperties = kafkaProperties;
    }

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory( consumerFactory( consumerClientId, consumerGroupId ) );
        factory.getContainerProperties().setAckMode( ContainerProperties.AckMode.MANUAL );
        return factory;
    }

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, String> topicMonitorContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory( consumerFactory( topicMonitorClientId, topicMonitorGroupId ) );
        factory.getContainerProperties().setAckMode( ContainerProperties.AckMode.MANUAL );
        factory.getContainerProperties().setConsumerRebalanceListener( new KafkaRebalanceListener( context ) );
        return factory;
    }

    private ConsumerFactory<String, String> consumerFactory( String clientId, String groupId ) {
        Map<String, Object> config = new HashMap<>();
        config.putAll( kafkaProperties.buildConsumerProperties() );
        config.put( ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServer );
        config.put( ConsumerConfig.CLIENT_ID_CONFIG, clientId );
        config.put( ConsumerConfig.GROUP_ID_CONFIG, groupId );
        config.put( ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false ); // needs to be turned off for rebalancing during topic addition and deletion
                                                                    // check -> /sf/ask/3938527701/?noredirect=1#comment99401765_56274988
        return new DefaultKafkaConsumerFactory<>( config, new StringDeserializer(), new StringDeserializer() );
    }
}


@Configuration
public class KafkaListenerConfiguration {

    @Bean
    @Scope("prototype")
    public KafkaMessageListener kafkaMessageListener() {
        return new KafkaMessageListener();
    }

}


@Slf4j
public class KafkaMessageListener {

    /*
     * This is the actual message listener that will process messages. It will be instantiated per topic.
     */
    @KafkaListener( topics = "${topic}", containerFactory = "kafkaListenerContainerFactory" )
    public void receiveHyperscalerMessage( ConsumerRecord<String, String> record, Acknowledgment acknowledgment, Consumer<String, String> consumer ) {

        log.debug("Kafka message - ThreadName={}, Hashcode={}, Partition={}, Topic={}, Value={}", 
                Thread.currentThread().getName(), Thread.currentThread().hashCode(), record.partition(), record.topic(), record.value() );

        // do processing

        // this is just a sample acknowledgment. it can be optimized to acknowledge after processing a batch of messages. 
        acknowledgment.acknowledge();
    }

}


@Service
public class KafkaTopicMonitor {

    /*
     * The main purpose of this listener is to detect the rebalance events on our topic pattern, so that 
     * we can create a listener bean instance (consumer thread) per topic. 
     *
     * Note that we use the wildcard topic pattern here.
     */
    @KafkaListener( topicPattern = ".*abc.def.ghi", containerFactory = "topicMonitorContainerFactory" )
    public void monitorTopics( ConsumerRecord<String, String> record ) {
        // do nothing
    }

}


@Slf4j
public class KafkaRebalanceListener implements ConsumerAwareRebalanceListener {

    private static final ConcurrentMap<String, KafkaMessageListener> listenerMap = new ConcurrentHashMap<>();
    private final ConfigurableApplicationContext context;

    public KafkaRebalanceListener( ConfigurableApplicationContext context ) {
        this.context = context;
    }

    public void onPartitionsRevokedBeforeCommit(Consumer<?, ?> consumer, Collection<TopicPartition> partitions) {
        // do nothing
    }

    public void onPartitionsRevokedAfterCommit(Consumer<?, ?> consumer, Collection<TopicPartition> partitions) {
        // do nothing
    }

    public void onPartitionsAssigned(Consumer<?, ?> consumer, Collection<TopicPartition> partitions) {

        log.info("OnPartitionsAssigned - partitions={} - {}", partitions.size(), partitions);
        Properties props = new Properties();
        context.getEnvironment().getPropertySources().addLast( new PropertiesPropertySource("topics", props) );

        for( TopicPartition tp: partitions ) {

            listenerMap.computeIfAbsent( tp.topic(), key -> {
                log.info("Creating messageListener bean instance for topic - {}", key );
                props.put( "topic", key );
                // create new KafkaMessageListener bean instance
                return context.getBean( "kafkaMessageListener", KafkaMessageListener.class );
            });
        }
    }
}