网站建设维护面试英雄联盟网站模板
消费者总体工作流程
Consumer Group（CG）：消费者组，由多个consumer组成。形成一个消费者组的条件，是所有消费者的groupid相同。
• 消费者组内每个消费者负责消费不同分区的数据，一个分区只能由一个组内消费者消费。
• 消费者组之间互不影响。所有的消费者都属于某个消费者组，即消费者组是逻辑上的一个订阅者。
消费者组初始化流程
1、coordinator：辅助实现消费者组的初始化和分区的分配。 coordinator节点选择 = groupid的hashcode值 % 50（ __consumer_offsets的分区数量） 例如： groupid的hashcode值 = 1，1% 50 = 1，那么__consumer_offsets 主题的1号分区，在哪个broker上，就选择这个节点的coordinator 作为这个消费者组的老大。消费者组下的所有的消费者提交offset的时候就往这个分区去提交offset；
2、coordinator选出一个 consumer作为leader；
3、coordinator把要消费的topic情况发送给leader消费者；
4、leader会负责制定消费方案；
5、把消费方案发给coordinator；
6、Coordinator就把消费方 案下发给各个consumer；
7、每个消费者都会和coordinator保持心跳（默认3s），一旦超时 （session.timeout.ms=45s），该消费者会被移除，并触发再平衡； 或者消费者处理消息的时间过长（max.poll.interval.ms5分钟），也会触发再平衡
消费者组详细消费流程
左侧为Kafka集群，右侧为消费者组，消费者创建网络连接客户端，消费者组调用sendFetches，抓取数据，同时还会准备两个参数，Fetch.min.bytes：每批次最小抓取大小，默认1字节，fetch.max.wait.ms一批数据最小值未达到的超时时间，默认500ms，任一条件满足，都会拉取数据；Fetch.max.bytes每批次最 大抓取大小，默认50m
send->拉取数据将数据放进completedFetches队列，消费者一批次拉取默认500条进行处理：反序列化->拦截器->处理数据
package com.atguigu.kafka.consumer;import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;import java.time.Duration;
import java.util.ArrayList;
import java.util.Properties;public class CustomConsumer {public static void main(String[] args) {//配置Properties properties = new Properties();//链接properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop102:9092,hadoop103:9092");//反序列化properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,StringDeserializer.class.getName());//配置消费者组idproperties.put(ConsumerConfig.GROUP_ID_CONFIG,"test");//1.创建消费者KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(properties);//2。订阅主题ArrayList<String> topics = new ArrayList<>();topics.add("first");kafkaConsumer.subscribe(topics);//3.消费数据while(true){ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(1));//拉数据for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {System.out.println(consumerRecord);}}}
}