kafka 消费者动态检测添加的主题 [英] kafka consumer to dynamically detect topics added

问题描述

我正在使用 KafkaConsumer 来消费来自 Kafka 服务器的消息(主题).

I'm using KafkaConsumer to consume messages from Kafka server (topics)..

• 它适用于在启动消费者代码之前创建的主题...

但问题是，如果主题是动态创建的(我的意思是说在消费者代码启动之后)，它将不起作用，但是 API 说它将支持动态主题创建..这是供您参考的链接..

But the problem is, it will not work if the topics created dynamically(i mean to say after consumer code started), but the API says it will support dynamic topic creation.. Here is the link for your reference..

使用的Kafka版本:0.9.0.1

Kafka version used : 0.9.0.1

https://kafka.apache.org/090/javadoc/index.html?org/apache/kafka/clients/consumer/KafkaConsumer.html

这里是JAVA代码...

Here is the JAVA code...

    Properties props = new Properties();
    props.put("bootstrap.servers", "localhost:9092");
    props.put("group.id", "test");
    props.put("enable.auto.commit", "false");
    props.put("auto.commit.interval.ms", "1000");
    props.put("session.timeout.ms", "30000");
    props.put("key.deserializer","org.apache.kafka.common.serialization.StringDeserializer");
    props.put("value.deserializer","org.apache.kafka.common.serialization.StringDeserializer");

    KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
    Pattern r = Pattern.compile("siddu(\\d)*");

    consumer.subscribe(r, new HandleRebalance());
    try {
         while(true) {
             ConsumerRecords<String, String> records = consumer.poll(Long.MAX_VALUE);
             for (TopicPartition partition : records.partitions()) {
                 List<ConsumerRecord<String, String>> partitionRecords = records.records(partition);
                 for (ConsumerRecord<String, String> record : partitionRecords) {
                     System.out.println(partition.partition()  + ": "  +record.offset() + ": " + record.value());
                 }
                 long lastOffset = partitionRecords.get(partitionRecords.size() - 1).offset();

                 consumer.commitSync(Collections.singletonMap(partition, new OffsetAndMetadata(lastOffset + 1)));
             }
         }
     } finally {
       consumer.close();
     }

注意:我的主题名称与正则表达式匹配..如果我重新启动消费者，它将开始阅读推送到主题的消息......

NOTE: My topic names are matching the Regular Expression.. And if i restart the consumer then it will start reading messages pushed to topic...

非常感谢任何帮助...

Any help is really appreciated...

推荐答案

在 apache kafka 邮件档案中有一个答案.我在下面复制它:

There was an answer to this in the apache kafka mail archives. I am copying it below:

消费者支持配置选项metadata.max.age.ms"它基本上控制了获取主题元数据的频率.经过默认情况下，这设置得相当高(5 分钟)，这意味着它需要最多 5 分钟发现与您的常规匹配的新主题表达.您可以将其设置得更低以更快地发现主题.

The consumer supports a configuration option "metadata.max.age.ms" which basically controls how often topic metadata is fetched. By default, this is set fairly high (5 minutes), which means it will take up to 5 minutes to discover new topics matching your regular expression. You can set this lower to discover topics quicker.

所以在你的道具中你可以:

So in your props you can:

props.put("metadata.max.age.ms", 5000);

这将使您的消费者每 5 秒发现一次新主题.

This will cause your consumer to find out about new topics every 5 seconds.

这篇关于kafka 消费者动态检测添加的主题的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！