如何让消费者在 Kafka 0.8 API 中工作

Question

我即将编写一个用于发布和使用 kafka 消息的原型。 我们确实已经设置了 Cloudera 基础架构（动物园管理员、代理等），并且我已经成功地使用了 Kafka 命令行工具来生成和使用消息。

我使用[org.apache.kafka/kafka_2.10 "0.8.2.1"]作为依赖，并且已经能够使用客户端API设置KafkaProducer，它发布带有纯字符串内容的消息，并且可以被命令行消费者成功读取另一边。

我的问题是：internets 上是否有一个代码示例来显示如何初始化KafkaConsumer，并在另一侧阅读该消息，因为我已经搜索了几天，但没有一个code examples 似乎在工作：

• 它们使用的类或方法甚至在 API 本身中都不存在（例如，它们看似将属性映射传递给 org.apache.kafka.clients.consumer.ConsumerConfig 的构造函数，但不存在这样的构造函数；
• 在 kafka.consumer.Consumer 类上调用 createJavaConsumerConnector 静态方法...这些东西存在于哪个宇宙中？）。

通常每个示例看起来都非常复杂。我希望消息传递框架需要几行配置来连接到代理，以及一些功能来放入/从队列或主题中取出。为 Kafka 设置 Producer 并不是一件非常复杂的事情，我希望 Consumer 也类似。

看来我也是not alone。

Answer 1

首先我想提一下，Kafka 0.8.0、0.8.1 和 0.8.2 之间有一些 API 更改（0.9.0 和 0.10.0 发生了市长重写和简化）——因此，您的问题有点开放，只是要求0.8。

要为 0.8.2.2 编写 Java 使用者，您需要包含依赖项：

这是针对 Scala 2.11 的——还有其他可用的 Scala 版本。

<dependency>
  <groupId>org.apache.kafka</groupId>
  <artifactId>kafka_2.11</artifactId>
  <version>0.8.2.2</version>
</dependency>

不要使用 kafka-clients 作为 0.8.x 的 artifactId。

消费者接收<String,String>键值对消息并将它们打印到stdout的最小示例如下所示：

import java.util.HashMap;
import java.util.Map;
import java.util.Properties;

import kafka.consumer.Consumer;
import kafka.consumer.ConsumerConfig;
import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;

public class ConsumerExample {

    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("zookeeper.connect", "localhost:2181");
        props.put("group.id", "myGroup");

        final String topic = "test";

        ConsumerConnector consumer = Consumer.createJavaConsumerConnector(new ConsumerConfig(props));

        Map<String, Integer> topicCountMap = new HashMap<String, Integer>();
        topicCountMap.put(topic, new Integer(1)); // number of consumer threads

        KafkaStream<byte[], byte[]> stream = consumer.createMessageStreams(topicCountMap).get(topic).get(0);

        ConsumerIterator<byte[], byte[]> it = stream.iterator();

        // infinite loop
        while(it.hasNext()) {
            System.out.println(new String(it.next().message()));
        }

        // non-reachable code...
        consumer.shutdown();
    }
}

一个完整的例子——使用多个消费者线程，包括正确关闭——可以在这里找到：https://cwiki.apache.org/confluence/display/KAFKA/Consumer+Group+Example

要对此进行测试，请遵循 quickstart 指南并通过 Kafka 的 console-producer 发送消息。