大数据学习笔记-------------------(17_2)

大数据学习笔记-------------------(17_2)

17.2 简单消费者实例

已经创建一个producer给Kafka集群发送信息。现在需要创建一个consumer来接收来自集群中的信息。KafkaConsumer API习惯用语接收来自Kafka集群的信息。KafkaConsumer类结构定义如下：

publicKafkaConsumer(java.util.Map configs)

config：返回消费者配置信息的map

KafkaConsumer类的重要方法列表如下：

17.2.1  consumerRecord API

ConsumerRecordAPI用于从Kafka集群接收记录。API由topic名称、分区号（从中接收记录）和指向Kafka分区中的记录的偏移量组成。ConsumerRecord类用于创建具有特定topic名称，分区计数、对的使用者记录。它有以下签名:

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

topic：从Kafka集群接收消费者记录的主题名称

partition：topic的分区数

key：记录的key，如果没有key存在将返回null

value：记录内容

ConsumerRecords API充当消费者记录的容器。API用于保存特定主题每个分区的ConsumerRecord列表。它的构造器定义如下：

public ConsumerRecords(java.util.Map>>records)

TopicPartition：返回特定topic分区map

Records：返回ConsumerRecord列表

ConsumerRecord类的定义方法列表：

17.2.3 配置设置

Consumer客户端API主要配置如下所示：

17.2.4 简单消费者应用程序

首先，启动ZooKeeper和Kafka代理。然后使用名为SimpleConsumer.java的Java类创建一个SimpleConsumer应用程序，并键入以下代码：

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; } String topicName = args[0].toString(); Properties props = new Properties(); //Kafka consumer configuration settings 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 consumer = new KafkaConsumer(props); consumer.subscribe(Arrays.asList(topicName)); //KafkaConsumer subscribes list of topics here. System.out.println("Subscribed to topic " + topicName); //print the topic name int i = 0; while (true) { Apache Kafka 33 ConsumerRecords records = con-sumer.poll(100); for (ConsumerRecord record : records) System.out.printf("offset = %d, key = %s, value = %s\n", record.offset(), record.key(), record.value()); // print the offset,key and value for the consumer records. } }}

编译：应用程序用下面的命令进行编译。

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

输入:打开生产者CLI发送一些信息到topic。可以输入相同如“Hello Consumer”

输出：

版权声明：本文内容由网络用户投稿，版权归原作者所有，本站不拥有其著作权，亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容，请联系我们jiasou666@gmail.com 处理，核实后本网站将在24小时内删除侵权内容。