Kafka 多主题消费

【问题标题】：Kafka multiple topic consumeKafka 多主题消费
【发布时间】：2017-11-24 12:16:59
【问题描述】：

   consumer.subscribe(Pattern.compile(".*"),new ConsumerRebalanceListener() {
            @Override
            public void onPartitionsRevoked(Collection<TopicPartition> clctn) {

            }

            @Override
            public void onPartitionsAssigned(Collection<TopicPartition> clctn) {
            }            
        });

如何在 apache/kafka 中使用正则表达式处理所有主题？上面的代码我试过了，还是不行。

【问题讨论】：

如果您解释了如何它不起作用，将会很有帮助。您预计会发生什么，实际会发生什么？
我想查看所有主题的日志。例如，如果我编写 Pattern.compile("abc_log") 而不是 Pattern.compile(".*")，我可以看到有关 abc_log 的日志。但我想查看与所有主题相关的所有日志。

标签： regex apache-kafka wildcard kafka-consumer-api

【解决方案1】：

对于正则表达式，请使用以下签名

KafkaConsumer.subscribe(Pattern pattern, ConsumerRebalanceListener listener)

例如下面的代码 sn-p 使消费者能够监听所有前缀为my_topics_的主题

ConsumerRebalanceListener listener = new ConsumerRebalanceListener() {

  @Override
  public void onPartitionsRevoked(Collection<TopicPartition> arg0) {
    // Don't need it now.
  }

  @Override
  public void onPartitionsAssigned(Collection<TopicPartition> arg0) {
    // Don't need it now.
  }
};

pattern = Pattern.compile("my_topics_.*");
kafkaConsumer.subscribe(pattern, listener);

【讨论】：

这是否也会在运行时检测到任何新主题的创建？
是的，它会的。只要它与模式匹配。