Apache Kafka：简单的生产者例子(三)

ConsumerRecord API

  ConsumerRecord API用于从Kafka集群接收记录。这个应用编程接口由一个主题名、分区号和一个偏移量组成，从这个主题名、分区号接收记录，偏移量指向Kafka分区中的记录。消费者记录类用于创建具有特定主题名称、分区计数和<键，值>对的消费者记录。

public ConsumerRecord(string topic,int partition, long offset,K key, V value)

  Topic:从卡夫卡群组收到的消费者记录的主题名称。

  Partition:主题的分区。

  Key:如果不存在关键字，将返回记录的关键字null。

  Value:记录内容。

ConsumerRecords API

  ConsumerRecords API充当ConsumerRecord的容器。该应用编程接口用于保存特定主题的每个分区的消费者记录列表。其构造函数定义如下。

public ConsumerRecords(java.util.Map<TopicPartition,java.util.List
<Consumer-Record>K,V>>> records)

  TopicPartition:返回特定主题的分区。

  Records:返回消费者记录列表。

  ConsumerRecords类定义了以下方法。

  public int count():所有主题的记录数。

  public Set partitions():该记录集中包含数据的分区集(如果没有返回数据，则该集为空)。

  public Iterator iterator():迭代器使您能够循环遍历集合，获取或重新移动元素。

  public List records():获取给定分区的记录列表。

配置选项

  bootstrap.servers：引导代理人列表。

  group.id：将单个消费者分配给组。

  enable.auto.commit：如果值为true，则启用偏移量自动提交，否则不提交。

  auto.commit.interval.ms：返回更新的消耗偏移量写入ZooKeeper的频率。

  session.timeout.ms：指示Kafka在放弃并继续消费消息之前，将等待ZooKeeper对请求(读或写)做出响应的毫秒数。

简单消费者应用程序

  生产者应用程序步骤在这里保持不变。首先，启动你的ZooKeeper和Kafka经纪人。然后用名为SimpleConsumer的java类创建一个简单消费者应用程序，并键入以下代码。

import java.util.Properties;
import java.util.Arrays;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.ConsumerRecord;

public class SimpleConsumer {
   public static void main(String[] args) throws Exception {
      if(args.length == 0){
         System.out.println("Enter topic name");
         return;
      }
      //Kafka consumer configuration settings
      String topicName = args[0].toString();
      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("session.timeout.ms", "30000");
      props.put("key.deserializer", 
         "org.apache.kafka.common.serializa-tion.StringDeserializer");
      props.put("value.deserializer", 
         "org.apache.kafka.common.serializa-tion.StringDeserializer");
      KafkaConsumer<String, String> consumer = new KafkaConsumer
         <String, String>(props);
      
      //Kafka Consumer subscribes list of topics here.
      consumer.subscribe(Arrays.asList(topicName))
      
      //print the topic name
      System.out.println("Subscribed to topic " + topicName);
      int i = 0;
      
      while (true) {
         ConsumerRecords<String, String> records = con-sumer.poll(100);
         for (ConsumerRecord<String, String> record : records)
         
         // print the offset,key and value for the consumer records.
         System.out.printf("offset = %d, key = %s, value = %s\n", 
            record.offset(), record.key(), record.value());
      }
   }
}

  可以使用以下命令编译应用程序。

javac -cp “/path/to/kafka/kafka_2.11-0.9.0.0/lib/*” *.java

  可以使用以下命令执行应用程序

java -cp “/path/to/kafka/kafka_2.11-0.9.0.0/lib/*”:. SimpleConsumer <topic-name>

  打开生产者命令行界面，并向主题发送一些消息。您可以将示例输入作为“Hello Consumer”。

输出

Subscribed to topic Hello-Kafka
offset = 3, key = null, value = Hello Consumer