kafka new api 0.10 不提供每个主题的流和消费者对象列表答案

【问题标题】：kafka new api 0.10 doesn't provide a list of stream and consumer objects per topickafka new api 0.10 不提供每个主题的流和消费者对象列表
【发布时间】：2017-02-09 11:46:42
【问题描述】：

以前我一直在使用 0.8 API。当您将主题列表传递给它时，它会返回一个流映射（每个主题一个条目）。这允许我生成一个单独的线程并将每个主题的流分配给它。每个主题中的数据太多，产生一个单独的线程有助于多任务处理。

//0.8 code sample
Map<String, List<KafkaStream<byte[], byte[]>>> consumerMap =  
consumer.createMessageStreams(topicCountMap);

我想升级到 0.10。我检查了KafkaStreams 和KafkaConsumer 类。 KafkaConsumer 对象接受配置属性并提供接受主题列表的订阅方法，其返回类型为 void。我找不到可以处理每个主题的方法。

KafkaConsumer consumer = new KafkaConsumer(props);
consumer.subscribe(topicsList);
conusmer.poll(long ms)

KafkaStreams 另一方面似乎有同样的问题。

KStreamBuilder builder = new KStreamBuilder();
String [] topics = new String[] {"topic1", "topic2"};
KStream<byte[], byte[]> source = builder.stream(stringSerde, stringSerde, topics);
KafkaStreams streams = new KafkaStreams(builder, props);
streams.start();

有source.foreach() 方法可用，但它是所有主题的流。任何人，有什么想法吗？

【问题讨论】：

标签： apache-kafka kafka-consumer-api apache-kafka-streams

【解决方案1】：

首先，使用多线程消费者很棘手，因此您在0.8 中使用的模式希望设计得很好:)

最佳实践是使用单线程消费者，因此，如果单个消费者同时订阅主题列表，则“无需”分隔不同的主题。然而，在使用记录时，记录对象会提供有关它来自哪个主题的信息（它携带此元数据）。因此，理论上您可以根据主题将记录分派到不同的线程进行实际处理（即使不推荐这样做！）。

Kafka 通过 partitions 横向扩展，因此，如果单线程消费者无法处理负载，您应该启动多个消费者（作为一个消费者组）扩展您的消费者处理能力。

一个更普遍的问题：如果您想处理每个主题的数据，为什么不使用多个消费者，每个消费者都订阅一个主题？

最后但同样重要的是，在 Apache Kafka 0.10+ 中，Kafka Streams API 是一个新引入的流处理库——尽管它不能与 0.8 KafkaStream 类混淆（提示，有没有“s”）。两者完全不相关。

【讨论】：

回答您的问题，是的，我将创建多个消费者，每个消费者每个主题。与从返回流映射的旧 api 调用 createMessageStreams(topicMapCount) 相比，每次为新主题创建连接感觉不正确。无论如何，感谢您的详细回答。