Kafka - 新消费者
一、数据来源
数据使用上一个博文所配置的 Flume，将文本数据写入到 Kafka中。不过这次有所改变，数据的监控目录 有所改变，写入的Kafka的主题名也变更为A25。
flume写入数据.png-36kB
这里我们可以看到 Flume 对于新传上去的 A91 数据已经完成消费。
二、消费者代码
2.1 创建消费者
创建消费者所使用的属性和生产者使用的属性差距不是很大：
- bootstrap.servers：指定了 Kafka 集群的连接字符串。
- key.deserializer 和 value.deserializer 与生产者的 serializer 定义也很类似，不过它们不是使用指定的类把 Java 对象转成字节数组，而是使用指定的类把字节数组转成 Java 对象。
- group.id：非必需，指定了 KafkaConsumer 属于哪一个消费者群组。
创建消费者的代码如下：
Properties props = new Properties();
props.put("bootstrap.servers", "master:9092");
props.put("group.id", "TestConsumer");
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);
2.2 订阅主题
consumer.subscribe(Collections.singletonList("A25"));
因为我们使用Flume对数据进行 Sinks 消费的时候，指定的主题为A25，因此我们这里在对数据进行订阅的时候，也是A25。
同时，可以对订阅的主题传递正则表达式进行匹配，一次订阅多个主题。
2.3 轮询消费
try {
while(true) {
ConsumerRecords<String, String> records = consumer.poll(100);
logger.info("records length = {}", records.count());
for (ConsumerRecord record : records) {
logger.info("topic = {}, partition = {}, offset = {}, key = {}, value = {}\n",
record.topic(), record.partition(), record.offset(),
record.key(), record.value());
}
}
} finally {
consumer.close();
}
这是一个无限循环。消费者实际上是一个长期运行的应用程序，它通过持续轮询向 Kafka 请求数据。
消费者必须持续对 Kafka 进行轮询，否则会被认为已经死亡，它的分区会被移交给群组里的其他消费者。传给 poll() 方法的参数是一个超时时间，用于控制 poll() 方法的阻塞时间（在消费者的缓冲区里没有可用数据时会发生阻塞）。如果该参数被设为 0，poll() 会立即返回，否则它会在指定的毫秒数内一直等待 broker 返回数据。
poll() 方法返回一个记录列表。每条记录都包含了记录所属主题的信息、记录所在分区的信息、记录在分区里的偏移量，以及记录的键值对。我们一般会遍历这个列表，逐条处理这些记录。
在退出应用程序之前使用 close() 方法关闭消费者。网络连接和 socket 也会随之关闭，并立即触发一次再均衡，而不是等待群组协调器发现它不再发送心跳并认定它已死亡，因为那样需要更长的时间，导致整个群组在一段时间内无法读取消息。
运行结果如下：
<center>
kafka消费结果.png-30kB
</center>
三、其他配置
3.1 pom文件
<properties>
<java.version>1.8</java.version>
<kafka.version>1.1.0</kafka.version>
</properties>
<dependencies>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>${kafka.version}</version>
</dependency>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-api</artifactId>
<version>1.7.21</version>
</dependency>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-log4j12</artifactId>
<version>1.7.21</version>
</dependency>
</dependencies>
3.2 log4j.properties
log4j.rootLogger=INFO,console
log4j.additivity.org.apache=true
# 控制台(console)
log4j.appender.console=org.apache.log4j.ConsoleAppender
log4j.appender.console.Threshold=DEBUG
log4j.appender.console.ImmediateFlush=true
log4j.appender.console.Target=System.err
log4j.appender.console.layout=org.apache.log4j.PatternLayout
log4j.appender.console.layout.ConversionPattern=%d{yyyy-MM-dd HH:mm:ss} [%p] %c - %m%n
