Kafka：0.11 版中 ConsumerConnector 的新 API 是什么

Question

我正在将 Kafka 客户端从 0.8.2.0 更新为 0.11.0.0。

在我的旧代码中，我使用ConsumerConnector 通过createMessageStreams 方法获取消息流，然后遍历每个主题的流。但是，似乎 ConsumerConnector 在新 API 中已被贬低。

package kafka.consumer

import ...
/**
 *  Main interface for consumer
 */
@deprecated("This trait has been deprecated and will be removed in a future release.", "0.11.0.0")
trait ConsumerConnector {
   ...
   def createMessageStreams[K,V](topicCountMap: Map[String,Int],
                                keyDecoder: Decoder[K],
                                valueDecoder: Decoder[V]): Map[String,List[KafkaStream[K,V]]]
   ...
}

我查找了新的 API 并找到了两个候选：

• Client API 在 org.apache.kafka.clients.consumer
• Stream API 在 org.apache.kafka.streams 中

我应该使用哪一个？而且，如何在新的 Kafka API 中实现同样的功能？

Answer 1

新消费者示例如下：

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "test");
props.put("enable.auto.commit", "true");
props.put("auto.commit.interval.ms", "1000");
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);
consumer.subscribe(Arrays.asList("foo", "bar"));
while (true) {
    ConsumerRecords<String, String> records = consumer.poll(100);
    for (ConsumerRecord<String, String> record : records)
        System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
}

更多详情请参阅https://kafka.apache.org/0110/javadoc/index.html。