本文主要实际编程讲解kafka生产者相关内容，版本kafka_2.11-0.10.1.0。
kafka消费者相对生产者来说，要复杂一些，主要是涉及到消息的拉取，消息的处理，偏移量offset的提交
前面博文中均已详细介绍，不再赘述。
linux集群安装参考大数据平台搭建-kafka集群的搭建
window单机版安装参考kafka知识体系50-生产者编程实践
kafka 0.10.1.0版本中采用KafkaConsumer对象用来向kafka broker集群拉取消息。
编写代码前先引入相关依赖包：
消费者位置和偏移量
消费者的位置表示下一条消息的偏移量，其比消费者在分区中已经消费的消息的偏移量还要大1。
提交的偏移量表示消费完成后，安全记录保存的偏移量，也是消费进程失败或者重启后的恢复点。可以同步提交commitSync或者异步提交偏移量commitAsync，也可以让消费者通过线程定时提交偏移量。
消费组和订阅主题
kafka允许多个消费者订阅同一个主题并共享同一个消费组，来实现多个进程来消费和处理消息。其中这些进程可以可以在同一台机器上运行，也可以分布在多台机器上，为处理提供可扩展性和容错能力。
组中的每个消费者都可以通过其subscribe方法动态设置要订阅的主题列表 。Kafka将订阅主题中的每条消息传递给每个消费者组中的一个进程。这是通过平衡消费者组中所有成员之间的分区来实现的，这样每个分区就只分配给该组中的一个使用者。因此，如果存在具有四个分区的主题和具有两个进程的使用者组，则每个进程将使用两个分区。
当消费者失败或者新增消费者，以及新增分区都会发生消费组内均衡操作。同时当均衡操作发生时，可以通过ConsumerRebalanceListener去监听，以便完成必要的应用程序级逻辑，例如状态清理，手动偏移提交等。
消费者故障检测
消费者在订阅主题后，调用poll方法时将自动加入组，poll方法旨在确保消费者的活力。只要继续调用poll方法，消费者将留在组中并继续从分配的分区中接收消息。在底层，是消费者会定期向服务端发送心跳。当消费者崩溃或者无法再持续的时间内(session.timeout.ms)发送心跳，则消费者被视为死亡，其占用的分区也会被重新分配。
此外，消费者可能遇到活锁的情况，即仍在发送心跳，但没有消费消息。
为了防止消费者在这种情况下无限期地占用其分区，我们使用该max.poll.interval.ms设置提供了活跃度检测机制。基本上，如果您不至少与配置的最大间隔一样频繁地调用轮询，则客户端将主动离开该组，以便其他使用者可以接管其分区。发生这种情况时，您可能会看到偏移提交失败（如CommitFailedException异常）。这是一种安全机制，可确保只有组中的活动成员才能提交偏移量。
消费者提供两个配置设置来控制轮询循环的行为：
max.poll.interval.ms：通过增加预期轮询之间的间隔，您可以为消费者提供更多时间来处理从中返回的一批记录poll(long)。缺点是增加此值可能会延迟组重新平衡，因为消费者将仅在轮询调用内加入重新平衡。
max.poll.records：使用此设置可限制从单个调用返回到poll的总记录数。这可以更容易地预测每个轮询间隔内必须处理的最大值。通过调整此值，您可以减少轮询间隔，这将减少组重新平衡的影响。
编码实践单线程自动提交偏移量
package com.molyeo.kafka; import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.clients.consumer.ConsumerRecords; import org.apache.kafka.clients.consumer.KafkaConsumer; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.util.Arrays; import java.util.Properties; public class SingleKafkaConsumerDemo { static Logger logger= LoggerFactory.getLogger(SingleKafkaConsumerDemo.class.getName()); public static void main(String[] args){ Properties props = new Properties(); props.put("bootstrap.servers","LAPTOP-2CBRDCI0:9092"); props.put("group.id","mygroup"); 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 consumer=new KafkaConsumer(props); consumer.subscribe(Arrays.asList("TEST")); try{ while (true){ ConsumerRecords<String,String> records=consumer.poll(200); for(ConsumerRecord<String,String> record:records){ logger.info("topic={},offset = {}, key = {}, value = {}", record.topic(), record.offset(), record.key(), record.value()); } } }catch (Exception e){ e.printStackTrace(); }finally { consumer.close(); } } }