Блог

У меня есть две разные темы для Кафки. Тип данных и структура данных (Json) одинаковы в обеих темах. Могу ли я использовать один и тот же контейнер прослушивателя для обеих тем? Я попробовал использовать, и мне удалось получить результаты, полученные из обеих тем.

Ниже приведены мои конфигурации.

 @Bean
public ConcurrentKafkaListenerContainerFactory<Object, Object> kafkaListenerContainerFactory(
    ConcurrentKafkaListenerContainerFactoryConfigurer factoryConfigure,
    ConsumerFactory<Object, Object> kafkaConsumerFactory) {
    ConcurrentKafkaListenerContainerFactory<Object, Object> factory = new ConcurrentKafkaListenerContainerFactory<>();
    factoryConfigure.configure(factory, kafkaConsumerFactory);
    factory.setBatchListener(true);
    return factory;
}
 

Это мой слушатель.

 
@KafkaListener(id = "#{'${spring.kafka.listener.id}'}", topics = "#{'${spring.kafka.consumer.topic}'}", groupId = "#{'${spring.kafka.consumer.group-id}'}")
public void getTopics( List<Request> model) {

    System.out.println("I am from topic1");
    System.out.println(model);

}

@KafkaListener(id = "#{'${spring.kafka.listener.id2}'}", topics = "#{'${spring.kafka.consumer.topic2}'}", groupId = "#{'${spring.kafka.consumer.group-id2}'}")
public void getTopics( List<Request> model) {

System.out.println("I am from topic2");
System.out.println(model);

}
 

Я явно не использую никаких аннотаций для указания имени контейнера.
Может ли это привести меня к какому-либо вопросу?

Неясно, о чем вы спрашиваете; фабрика контейнеров создаст отдельный контейнер для каждого слушателя.

topics Свойство представляет собой массив, поэтому вы можете использовать

 @KafkaListener(id = "#{'${spring.kafka.listener.id}'}", 
    topics = { "${spring.kafka.consumer.topic}",
               "${spring.kafka.consumer.topic2}" },
    groupId = "#{'${spring.kafka.consumer.group-id}'}")
public void getTopics( List<Request> model) {

    System.out.println("I am from both topics");
    System.out.println(model);

}
 

Если вам нужно знать, из какой темы взята каждая запись, вы можете использовать

 public void getTopics( List<Request> model,
     @Header(KafkaHeaders.RECEIVED_TOPIC) List<String> topics) {
 

Тема в индексе n предназначена для записи в индексе модели n.