springbootでトピックごとに別々のKafkaリスナーを作成するにはどうすればいいですか？

Question

私は春とカフカが初めてです。私は、ユーザが実行時にカフカのトピックを作成することができるユースケース[SpringBoot-kafkaを使用している]に取り組んでいます。 Springアプリケーションは、実行時にこれらのトピックをプログラマブルにサブスクライブすることが期待されています。私が今まで知っているのは、カフカのリスナーはデザインタイムであるため、起動前にトピックを指定する必要があるということです。 SpringBoot-Kafka統合でkafkaトピックを動的に購読する方法はありますか？ここに述べたように、私は実装する予定です。この https://github.com/spring-projects/spring-kafka/issues/132

現在のアプローチは、[Javaコードを使用]の代わりにカフカの消費者を自分で実装する春カフカの統合を使用していないされて呼ば

spring boot kafka consumer - how to properly consume kafka messages from spring boot

Answer 1

カフカリスナー注釈を使用してそれらを指定する場合は、「デザインタイム」のみです。 Spring-kafkaでは、それらを動的に作成することもできます。KafkaMessageListenerContainerを参照してください。

その場で作成されたカフカリスナーの最も簡単な例は次のようになります。より多くの説明とコードの場合

Map<String, Object> consumerConfig = ImmutableMap.of(
    BOOTSTRAP_SERVERS_CONFIG, "brokerAddress", 
    GROUP_ID_CONFIG, "groupId" 
); 

DefaultKafkaConsumerFactory<String, String> kafkaConsumerFactory = 
     new DefaultKafkaConsumerFactory<>(
       consumerConfig, 
       new StringDeserializer(), 
       new StringDeserializer()); 

ContainerProperties containerProperties = new ContainerProperties("topicName"); 
containerProperties.setMessageListener((MessageListener<String, String>) record -> { 
    //do something with received record 
} 

ConcurrentMessageListenerContainer container = 
     new ConcurrentMessageListenerContainer<>(
       kafkaConsumerFactory, 
       containerProperties); 

container.start(); 

このブログ記事を参照してください。http://www.douevencode.com/articles/2017-12/spring-kafka-without-annotations/