如何为Kafka消费者使用动态主题
config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
config.put(ConsumerConfig.GROUP_ID_CONFIG, "group_id");
config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
return new DefaultKafkaConsumerFactory<>(config);
}
@Bean
public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
ConcurrentKafkaListenerContainerFactory<String, String> factory = new
ConcurrentKafkaListenerContainerFactory();
factory.setConsumerFactory(consumerFactory());
return factory;
}
here I need to change the topic as a dynamic content
@KafkaListener(topics = "mytopic")
private void consume2(String Data) {
String userID = null;
String topic = null;
String message = null;
System.out.println(Data);
}
共有1个答案
你有没有试过
@kafkalistener(topics=“${kafka.topics}”）
并将其设置为环境变量或应用程序属性？
-
目前，我有一个Flink集群，它想通过一个模式消费Kafka主题，通过使用这种方式，我们不需要维护一个硬代码Kafka主题列表。 --update--我需要知道主题信息的原因是我们需要这个主题名称作为参数，在即将到来的Flink sink部分中使用。
-
因为我是新的Kafka，所以我能够从文件中读取记录，并通过生产者将消息发送到Kafka主题，但不能通过消费者消费相同的主题。 注意：您可以从任何文本文件中读取数据，我使用的是Kafka2.11-0.9。0.0版本 这是我的密码： 下面是输出：
-
我的问题与单个消费者从多个话题消费有关。假设所有主题都加载了1M个记录，一个使用者必须处理这些记录。它将按照什么顺序从主题中读取（我的意思是首先读取哪个主题/分区，等等） Kafka内部资料的任何链接会有帮助吗？
-
我有一个主题列表（目前是10个），其大小可以在未来增加。我知道我们可以产生多个线程（每个主题）来消耗每个主题，但在我的例子中，如果主题的数量增加，那么消耗主题的线程数量也会增加，这是我不希望的，因为主题不会太频繁地获取数据，所以线程将是理想的。 有没有办法让单个消费者从所有话题中消费？如果是的话，我们怎样才能做到呢？另外，Kafka将如何维护抵消？请建议答案。
-
本文向大家介绍Kafka 的消费者如何消费数据相关面试题，主要包含被问及Kafka 的消费者如何消费数据时的应答技巧和注意事项，需要的朋友参考一下 消费者每次消费数据的时候，消费者都会记录消费的物理偏移量（offset）的位置 等到下次消费时，他会接着上次位置继续消费
-
我将Kafka用于一个消息传递应用程序。对于这个应用程序，有一个生产者将消息放入一个主题，消费者注册到这个主题，并消费这些消息。这些消费者是Dockerized应用程序。出于自动缩放的目的，每个使用者在创建时都注册为具有唯一ID的使用者。 Consumer2被创建为docker容器，并将自身注册为ID为的使用者 现在，无论出于什么原因，失败，并被替换，后者将自己注册为kafka的使用者，其ID为。