5.kafka API consumer

1.kafka consumer流程
1.1.在启动时或者协调节点故障转移时，消费者发送ConsumerMetadataRequest给bootstrap brokers列表中的任意一个brokers。在ConsumerMetadataResponse中，它接收消费者对应的消费组所属的协调节点的位置信息。

1.2.消费者连接协调节点，并发送HeartbeatRequest。如果返回的HeartbeatResponse中返回IllegalGeneration错误码，说明协调节点已经在初始化。消费者就会停止抓取数据，提交offsets，发送JoinGroupRequest给协调节点。在JoinGroupResponse，它接收消费者应该拥有的topic-partitions列表以及当前消费组的新的generation编号。这个时候消费组管理已经完成，消费者就可以开始抓取数据，并为它拥有的partitions提交offsets。

1.3.如果HeartbeatResponse没有错误返回，消费者会从它上次拥有的partitions列表继续抓取数据，这个过程是不会被中断的。

Coordinator协调节点的工作过程：
1.在稳定状态下，协调节点通过故障检测协议跟踪每个消费组中每个消费者的健康状况。

2.在选举和启动时，协调节点读取它管理的消费组列表，以及从ZK中读取每个消费组的成员信息。如果之前没有成员信息，它不会做任何动作。只有在同一个消费组的第一个消费者注册进来时，协调节点才开始工作(即开始加载消费组的消费者成员信息)。

3.当协调节点完全加载完它所负责的消费组列表的所有组成员之前，它会在以下几种请求的响应中返回CoordinatorStartupNotComplete错误码：HeartbeatRequest，OffsetCommitRequest，JoinGroupRequest。这样消费者就会过段时间重试(直到完全加载，没有错误码返回为止)。

4.在选举或启动时，协调节点会对消费组中的所有消费者进行故障检测。根据故障检测协议被协调节点标记为Dead的消费者会从消费组中移除，这个时候协调节点会为Dead的消费者所属的消费组触发一个平衡操作(消费者Dead之后，这个消费者拥有的partition需要平衡给其他消费者)。

5.当HeartbeatResponse返回IllegalGeneration错误码，就会触发平衡操作。一旦所有存活的消费者通过JoinGroupRequests重新注册到协调节点，协调节点会将最新的partition所有权信息在JoinGroupResponse的每个消费者之间通信(同步)，然后就完成了平衡操作。

6.协调节点会跟踪任何一个消费者已经注册的topics的topic-partition的变更。如果它检测到某个topic新增的partition，就会触发平衡操作。当创建一个新的topics也会触发平衡操作，因为消费者可以在topic被创建之前就注册它感兴趣的topics。

2.消费者组的使用场景

Kafka里的消费者组有两个使用的场景：
2.1“队列模式”：在同一组的消费者共同消费一个主题的所有消息，而且确保一条消息只被一个消费者处理。一个主题的所有的分区会和一个消费组的所有消费者做关联：一个分区只会与一个消费者关联，它的消息不会被其它的消费者接收。
最开始只有一个消费者时，所有的分区都分配给了它。当消息的规模增加时，我们就需要扩展消费者的数量，水平扩展处理能力，一直可以达到每个消费者只关联一个分区。大于分区数的消费者是会处在空闲状态，因为没有分配任何的分区。

2.2“发布/订阅模式”: 创建不同的消费者组意味一个主题的消息会发送给所有订阅它的消费者组，然后消费者组依照前面共同协作的场景进行分配。这往往是因为我们有不同的应用需求，比如一批交易数据，资金系统、ERP系统会消费它而风险监控也需要同时消费它。这就实现了数据的透明异步共用。

在两个场景中，消费者组有个重要的功能：rebalancing。当一个新的消费者加入一个组，如果还有有效的分区（消费者数<=主题分区数），会开始一个重新均衡分配的操作，会将一个已关联的分区（它的原消费者仍保有至少一个分区）重新分配给新加入的消费者。同样的，当一个消费者因为各种原因离开这个组，它的所有分区会被分配给剩下的消费者。

Subscribe（自动） assign（手动）
前面所说的自动分配是指在 KafkaConsumer API中的subscribe()方法。这个方法强制要求你为消费者设置一个消费者组，group.id参数不能为空。而你不需要处理分区的分配问题。而对应subscribe()方法。你可以采用手动的方式，指定消费者读取哪个主题分区，则：assign() 方法。当你需要精确地控制消息处理的负载，也能确定哪个分区有哪些消息时，这种手动的方式会很有用

3.自动提交方式api
[hadoop@h201 kafka_2.12-0.10.2.1]$ bin/kafka-topics.sh --create --zookeeper h201:2181,h202:2181,h203:2181 --replication-factor 2 --partitions 3 --topic topic11

[hadoop@h201 kkk]$ vi cc.java
import org.apache.kafka.clients.consumer.*;
import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.Arrays;
public class cc {
public static void main(String[] args) throws ExecutionException, InterruptedException {
Properties props = new Properties();
//设置kafka集群的地址
props.put("bootstrap.servers", "h201:9092,h202:9092,h203:9092");
//设置消费者组，组名字自定义，组名字相同的消费者在一个组
props.put("group.id", "g11");
//开启offset自动提交
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<String, String> consumer = new KafkaConsumer<>(props);
//消费者订阅主题，可以订阅多个主题
consumer.subscribe(Arrays.asList("topic11"));
//死循环不停的从broker中拿数据
while (true) {
ConsumerRecords<String, String> records = consumer.poll(100);
for (ConsumerRecord<String, String> record : records)
System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
}
}
}

[hadoop@h201 kkk]$ /usr/jdk1.8.0_144/bin/javac -classpath /home/hadoop/kafka_2.12-0.10.2.1/libs/kafka-clients-0.10.2.1.jar cc.java
[hadoop@h201 kkk]$ /usr/jdk1.8.0_144/bin/java cc

解释：
Poll方法用来获取消息 ，poll（拉取）
consumer.poll(100) ：100ms内拉取一次数据
Record ：为存储的消息，record.value 为消息的内容