【Kafka】消费者Consumer详解

目录

消息消费

消息消费流程

参考：https://mp.weixin.qq.com/s/v6jUK8TIPi1Debfd40GU3w

消费者消费也主要分为两个阶段：

1. 信息注册阶段，即整个消费者组向集群注册消费信息等
2. 信息消费阶段，开始信息消息，确保消息可靠性等

信息注册

信息注册流程如下：

注册流程如下：

1. 消费者组内所有消费者都会向集群寻找自己的Coordinator（组协调器，消费者后续很多操作都需要发送请求到这个节点）
2. 找到Coordinator后，所有的Consumer都会向Coordinator发起join group加入消费者组的请求
3. Coordinator会选择一个最早发起请求的Consumer作为leader Consumer，其他的Consumer作为follower

消息消费

消费流程图如下：

消费流程如下：

1. leader消费者根据要消费的Topic及分区情况制定一个消费方案
2. leader告知给Coordinator，Coordinator再将消费方案告知给各个follower
3. 消费者到指定分区拉取消息
4. 消息经过反序列化转为正常的消息对象
5. 经过拦截器
6. 消费者提交位移到kafka

消费方式

我们知道消息队列一般有两种实现方式：Push(推模式) 和 Pull(拉模式)。

Kafka Consumer 采用的是主动拉取 Broker 数据进行消费的即 Pull 模式。

为什么不采用Push模式？

如果是选择 Push 模式最大缺点就是 Broker 不清楚 Consumer 的消费速度，且推送速率是 Broker 进行控制的， 这样很容易造成消息堆积，如果 Consumer 中执行的任务操作是比较耗时的，那么 Consumer 就会处理的很慢， 严重情况可能会导致系统 Crash。

为什么采用Pull模式？

如果选择 Pull 模式，这时 Consumer 可以根据自己的情况和状态来拉取数据, 也可以进行延迟处理。但是 Pull 模式也有不足，Kafka 又是如何解决这一问题？**如果 Kafka Broker 没有消息，这时每次 Consumer 拉取的都是空数据, 可能会一直循环返回空数据。 针对这个问题，Consumer 在每次调用 Poll() 消费数据的时候，顺带一个 timeout 参数，当返回空数据的时候，会在 Long Polling 中进行阻塞，等待 timeout 再去消费，直到数据到达。

反序列化

Kafka的broker中所有的消息都是字节数组，消费者获取到消息之后，需要先对消息进行反序列化处理，然后才能交给用户程序消费处理。

消费者的反序列化器包括key的和value的反序列化器：

key.deserializer

、

value.deserializer

。

反序列化器需要实现接口

org.apache.kafka.common.serialization.Deserializer<T>

，如下;

packageorg.apache.kafka.common.serialization;importjava.io.Closeable;importjava.util.Map;importorg.apache.kafka.common.header.Headers;publicinterfaceDeserializer<T>extendsCloseable{defaultvoidconfigure(Map<String,?> configs,boolean isKey){}Tdeserialize(String var1,byte[] var2);defaultTdeserialize(String topic,Headers headers,byte[] data){returnthis.deserialize(topic, data);}defaultvoidclose(){}}

同时kafka也提供了一些默认的反序列化器：

除了上述提供的，还可以自定义序列化器，只要实现Deserializer接口即可。

importcom.lagou.kafka.demo.entity.User;importorg.apache.kafka.common.serialization.Deserializer;importjava.nio.ByteBuffer;importjava.util.Map;publicclassUserDeserializerimplementsDeserializer<User>{@Overridepublicvoidconfigure(Map<String,?> configs,boolean isKey){}@OverridepublicUserdeserialize(String topic,byte[] data){ByteBuffer allocate =ByteBuffer.allocate(data.length);
        allocate.put(data);
        allocate.flip();int userId = allocate.getInt();int length = allocate.getInt();System.out.println(length);String username =newString(data,8, length);returnnewUser(userId, username);}@Overridepublicvoidclose(){}}

消费时配置写上我们自定义的反序列化器：

Map<String,Object> configs =newHashMap<>();...
configs.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,UserDeserializer.class);KafkaConsumer<String,User> consumer =newKafkaConsumer<String,User>(configs);
consumer.subscribe(Collections.singleton("tp_user_01"));...

拦截器

消费者在拉取了分区消息之后，经过反序列化器对key和value进行反序列化处理后，消费端设置了拦截器，则需要经过拦截器的处理之后，才能返回给消费者应用程序进行处理。

拦截器特点：

1. 一个可插拔接口，允许拦截甚至更改消费者接收到的消息。首要的用例在于将第三方组件引入消费者应用程序，用于定制的监控、日志处理等。
2. 该接口的实现类通过configre方法获取消费者配置的属性，如果消费者配置中没有指定clientID，还可以获取KafkaConsumer生成的clientId。获取的这个配置是跟其他拦截器共享的，需要保证不会在各个拦截器之间产生冲突。
3. ConsumerInterceptor方法抛出的异常会被捕获、记录，但是不会向下传播。如果用户配置了错误的key或value类型参数，消费者不会抛出异常，而仅仅是记录下来。
4. ConsumerInterceptor回调发生在org.apache.kafka.clients.consumer.KafkaConsumer#poll(long)方法同一个线程。

消费端定义消息拦截器，需要实现

org.apache.kafka.clients.consumer.ConsumerInterceptor<K, V>

接口。

改接口源码如下：

packageorg.apache.kafka.clients.consumer;importjava.util.Map;importorg.apache.kafka.common.Configurable;importorg.apache.kafka.common.TopicPartition;publicinterfaceConsumerInterceptor<K,V>extendsConfigurable,AutoCloseable{ConsumerRecords<K,V>onConsume(ConsumerRecords<K,V> var1);voidonCommit(Map<TopicPartition,OffsetAndMetadata> var1);voidclose();}

案例：自己实现一个拦截器

importorg.apache.kafka.clients.consumer.ConsumerInterceptor;importorg.apache.kafka.clients.consumer.ConsumerRecords;importorg.apache.kafka.clients.consumer.OffsetAndMetadata;importorg.apache.kafka.common.TopicPartition;importjava.util.Map;publicclassOneInterceptorimplementsConsumerInterceptor<String,String>{@OverridepublicConsumerRecords<String,String>onConsume(ConsumerRecords<String,String> records){// poll方法返回结果之前最后要调用的方法System.out.println("One -- 开始");// 消息不做处理，直接返回return records;}@OverridepublicvoidonCommit(Map<TopicPartition,OffsetAndMetadata> offsets){// 消费者提交偏移量的时候，经过该方法 System.out.println("One -- 结束");}@Overridepublicvoidclose(){// 用于关闭该拦截器用到的资源，如打开的文件，连接的数据库等 }@Overridepublicvoidconfigure(Map<String,?> configs){// 用于获取消费者的设置参数
        configs.forEach((k, v)->{System.out.println(k +"\t"+ v);});}}

同理可配置多个拦截器TwoInterceptor、ThreeInterceptor。

消费者配置里添加自定义的拦截器：

Properties props =newProperties();...// 配置拦截器// One -> Two -> Three，接收消息和发送偏移量确认都是这个顺序
props.setProperty(ConsumerConfig.INTERCEPTOR_CLASSES_CONFIG,"com.lagou.kafka.demo.interceptor.OneInterceptor"+",com.lagou.kafka.demo.interceptor.TwoInterceptor"+",com.lagou.kafka.demo.interceptor.ThreeInterceptor");KafkaConsumer<String,String> consumer =newKafkaConsumer<String,String>(props);// 订阅主题 
consumer.subscribe(Collections.singleton("tp_demo_01"));

消费者原理剖析

消费者组和消费者

多个消费者组成消费者组，多个从同一个主题消费的消费者可以加入到一个消费组中，消费者消费消息都是以组为单位来订阅主题的。

消费者组是kafka提供的可扩展且具有容错性的消费者机制。

主要有如下特性：

1. 消费组有一个或多个消费者，消费者可以是一个进程，也可以是一个线程
2. 有group.id，是一个字符串，唯一标识一个消费组，消费组中的消费者共享group_id
3. 消费组订阅的主题每个分区只能分配给消费组一个消费者
4. 消费者在消费的过程中记录已消费的数据，即消费位移（offset）信息，每个消费组保存自己的位移信息

group_id一般设置为应用的逻辑名称。比如多个订单处理程序组成一个消费组，可以设置group_id为"order_process"。group_id通过消费者的配置指定：

group.id=xxxxx

。

消费组最重要的作用是均衡地给消费者分配分区，每个分区只由消费组中一个消费者消费。

一个拥有四个分区的主题，包含一个消费者的消费组。此时，消费组中的消费者消费主题中的所有分区。并且没有重复的可能：

如果在消费组中添加一个消费者2，则每个消费者分别从两个分区接收消息：

如果消费组有四个消费者，则每个消费者可以分配到一个分区：

如果向消费组中添加更多的消费者，超过主题分区数量，则有一部分消费者就会闲置，不会接收任何消息：

向消费组添加消费者是横向扩展消费能力的主要方式。

必要时，需要为主题创建大量分区，在负载增长时可以加入更多的消费者。但是不要让消费者的数量超过主题分区的数量。

除了通过增加消费者来横向扩展单个应用的消费能力之外，经常出现多个应用程序从同一个主题消费的情况：

此时，每个应用都可以获取到所有的消息。只要保证每个应用都有自己的消费组，就可以让它们获取到主题所有的消息。

横向扩展消费者和消费组不会对性能造成负面影响。

为每个需要获取一个或多个主题全部消息的应用创建一个消费组，然后向消费组添加消费者来横向扩展消费能力和应用的处理能力，每个消费者只处理一部分消息。

心跳机制

有时消费者宕机，退出消费组，或者于broker宕机，导致主题的某个分区宕机，这时为了维持原来的平衡状态，就会触发再平衡，后面还会详细讲解。

那么kafka如何知道消费者或broker宕机不可用呢？

这就需要心跳机制。

Kafka 的心跳是 Kafka Consumer 和 Broker 之间的健康检查，只有当 Broker Coordinator 正常时，Consumer 才会发送心跳。

broker 端参数

session.timeout.ms 

，sessionTimeoutMs 参数。

broker 处理心跳的逻辑在 GroupCoordinator 类中：如果心跳超期， broker coordinator 会把消费者从 group 中移除，并触发 rebalance。

consumer 端参数

session.timeout.ms 

，sessionTimeoutMs

max.poll.interval.ms 

，rebalanceTimeoutMs

如果客户端发现心跳超期，客户端会标记 coordinator 为不可用，并阻塞心跳线程；如果超过了poll 消息的间隔超过了 rebalanceTimeoutMs，则 consumer 告知 broker 主动离开消费组，也会触发在平衡。

消费者位移

概述

前面我们说过，Kafka中，消费者根据消息的位移（offset）顺序消费消息。

当消费者进行消费的时候，是要去指分区找消息的offset，从而找到消息进行消费。他与分区的最新消息的Offset是不一样的。他是存储在消费者组中，每个分区都有一个offset，用于标识，当前消费者组下一个要消费消息的offset。

消费者的位移由消费者管理，可以存储于zookeeper中，也可以存储于Kafka主题

__consumer_offsets

中。Kafka 旧版本（0.8版本之前）是重度依赖 Zookeeper 来实现各种各样的协调管理，当然旧版本的 Consumer Group 是把位移保存在 ZooKeeper 中，减少 Broker 端状态存储开销，鉴于 Zookeeper 的存储架构设计来说， 它不适合频繁写更新，而 Consumer Group 的位移提交又是高频写操作，这样会拖慢 ZooKeeper 集群的性能， 于是在新版 Kafka 中， 社区重新设计了 Consumer Group 的位移管理方式，采用了将位移保存在 Kafka 内部（这是因为 Kafka Topic 天然支持高频写且持久化），即__consumer_offsets。

它是由 Kafka 自动创建的，和普通的 Topic 相同，它的消息格式也是 Kafka 自己定义的，我们无法进行修改。

每个消费者组会为它消费的分区维护属于自己的位置信息，记录当前消费到该分区的哪个位置。在消费者组提交位移（后续讲解）时，会将位移数据作为普通的Kafka消息提交到

__consumer_offsets

主题中。这个主题的每个分区只会有一个生产者，因此可以保证消息的有序性和原子性。

这种设计使得Kafka能够支持高频的写操作，并且可以持久化保存位移信息。同时，通过将位移信息保存在Kafka内部主题中，Kafka可以提供更强大的一致性和可靠性保证，确保消息被正确地写入到Kafka中。

后面还会专门讲解__consumer_offsets的存储结构。

位移管理

Kafka提供了消费者API，让消费者可以管理自己的位移。

1. public void assign(Collection<TopicPartition> partitions)

给当前消费者手动分配一系列主题分区。

手动分配分区不支持增量分配，如果先前有分配分区，则该操作会覆盖之前的分配。如果给出的主题分区是空的，则等价于调用unsubscribe方法。

手动分配主题分区的方法不使用消费组管理功能。当消费组成员变了，或者集群或主题的元数据改变了，不会触发分区分配的再平衡。

手动分区分配assign(Collection)不能和自动分区分配subscribe(Collection,ConsumerRebalanceListener)一起使用。

如果启用了自动提交偏移量，则在新的分区分配替换旧的分区分配之前，会对旧的分区分配中的消费偏移量进行异步提交。

1. public Set<TopicPartition> assignment()

获取给当前消费者分配的分区集合。如果订阅是通过调用assign方法直接分配主题分区，则返回相同的集合。如果使用了主题订阅，该方法返回当前分配给该消费者的主题分区集合。如果分区订阅还没开始进行分区分配，或者正在重新分配分区，则会返回none。

1. public Map<String, List<PartitionInfo>> listTopics()

获取对用户授权的所有主题分区元数据。该方法会对服务器发起远程调用。

1. public List<PartitionInfo> partitionsFor(String topic)

获取指定主题的分区元数据。如果当前消费者没有关于该主题的元数据，就会对服务器发起远程调用。

1. public Map<TopicPartition, Long> beginningOffsets(Collection<TopicPartition> partitions)

对于给定的主题分区，列出它们第一个消息的偏移量。注意，如果指定的分区不存在，该方法可能会永远阻塞。该方法不改变分区的当前消费者偏移量。

1. public void seekToEnd(Collection<TopicPartition> partitions)

将偏移量移动到每个给定分区的最后一个。该方法延迟执行，只有当调用过poll方法或position方法之后才可以使用。如果没有指定分区，则将当前消费者分配的所有分区的消费者偏移量移动到最后。如果设置了隔离级别为：isolation.level=read_committed，则会将分区的消费偏移量移动到最后一个稳定的偏移量，即下一个要消费的消息现在还是未提交状态的事务消息。

1. public void seek(TopicPartition partition, long offset)

将给定主题分区的消费偏移量移动到指定的偏移量，即当前消费者下一条要消费的消息偏移量。若该方法多次调用，则最后一次的覆盖前面的。如果在消费中间随意使用，可能会丢失数据。

1. public long position(TopicPartition partition)

检查指定主题分区的消费偏移量

1. public void seekToBeginning(Collection<TopicPartition> partitions)

将给定每个分区的消费者偏移量移动到它们的起始偏移量。该方法懒执行，只有当调用过poll方法或position方法之后才会执行。如果没有提供分区，则将所有分配给当前消费者的分区消费偏移量移动到起始偏移量。

位移提交

参考：https://mp.weixin.qq.com/s/v6jUK8TIPi1Debfd40GU3w

前面我们说过，消费者组需要向Kafka记录自己的位移数据，这个过程称为提交位移(Committing Offsets) 。

在消费者正常消费的时候，调用

poll(100)

方法拉取消息时，从消费者组本地中拿到当前需要消费的指定分区的offset，容然后去消费。在拉取完成后，它会记录每个消息的位移在本地。在处理完这些消息后，消费者可以选择手动提交位移，或者开启自动提交功能，把位移提交给kafka。

但是如果消费者出现异常，例如消费者宕机，他本地保存的offset就会丢失，下次重启后，再次去消费，就需要去kafka服务器中（也就是以前消费者提交过来的offset信息）按照一定策略找到offset。

消费者组需要为分配给它的每个分区提交各自的位移数据。 位移提交的由消费端负责的，Kafka只负责保管到

__consumer_offsets

主题中。__consumer_offsets 中的消息保存了每个消费组某一时刻提交的offset信息。

可以使用如下命令查询：

kafka-console-consumer.sh --topic __consumer_offsets --bootstrap-server node1:9092 --formatter "kafka.coordinator.group.GroupMetadataManager\$OffsetsMessageFormatter" -- consumer.config /opt/kafka_2.12-1.0.2/config/consumer.properties --from- beginning |head

结果类似如下：

上图中，标出来的，表示消费组为 test-consumer-group ，消费的主题为 __consumer_offsets ，消费的分区是4，偏移量为5。

__consumers_offsets 主题配置了compact策略，使得它总是能够保存最新的位移信息，既控制了该topic总体的日志容量，也能实现保存最新offset的目的。

位移提交分为自动提交和手动提交；也可以分为同步提交和异步提交。

组合起来可有三种提交方式：自动异步提交、手动同步提交、手动异步提交。

1. 自动异步提交

开启自动提交：

enable.auto.commit=true

配置自动提交间隔：Consumer端，

auto.commit.interval.ms

，默认 5s

enable.auto.commit

设置为true时，每隔

auto.commit.interval.ms

时间会自动提交已经已经拉取到的消息中最大的offset。

示例：

Map<String,Object> configs =newHashMap<>(); 
configs.put("bootstrap.servers","node1:9092"); 
configs.put("group.id","mygrp");// 设置偏移量自动提交。自动提交是默认值。这里做示例。
configs.put("enable.auto.commit","true");// 偏移量自动提交的时间间隔 
configs.put("auto.commit.interval.ms","3000"); 
configs.put("key.deserializer",StringDeserializer.class); 
configs.put("value.deserializer",StringDeserializer.class);KafkaConsumer<String,String> consumer =newKafkaConsumer<String,String>(configs); consumer.subscribe(Collections.singleton("tp_demo_01"));while(true){ConsumerRecords<String,String> records = consumer.poll(100);for(ConsumerRecord<String,String>record: records){System.out.println(record.topic()+"\t"+record.partition()+"\t"+record.offset()+"\t"+record.key()+"\t"+record.value());}}

Kafka自动提交位移的原理是基于后台异步执行的。消费者提交消息偏移量时并不关心消息是否已经被处理了，导致消息的提交与处理进度可能有很大差距。很可能造成消息的丢失或重复处理。

重复消费：Consumer 每 5s 提交 offset，假设提交 offset 后的 3s 消费者宕机了或者发生再平衡（Rebalance）， 之后的所有 Consumer 从上一次提交的 offset 处继续消费，因此前 3s 的消息会被重复消费。

消息丢失：由于是异步提交，如果消费者消费速度慢，offset先提交了，消费者消费过程宕机了，那么offset已经被提交，但是消费者并没有消费成功，之后Consumer 从上一次提交的 offset 处继续消费，导致消息丢失。

自动提交原理

在poll执行完成后，会执行

maybeAutoCommitOffsetsAsync

方法，

代码如下：

publicbooleanpoll(Timer timer,boolean waitForJoinGroup){...// 检查是否自动提交位移maybeAutoCommitOffsetsAsync(timer.currentTimeMs());returntrue;}

maybeAutoCommitOffsetsAsync

代码如下：

publicvoidmaybeAutoCommitOffsetsAsync(long now){// 是否开启自动提交if(autoCommitEnabled){
        nextAutoCommitTimer.update(now);// 是否到达提交时间if(nextAutoCommitTimer.isExpired()){
            nextAutoCommitTimer.reset(autoCommitIntervalMs);// 进入提交下一步doAutoCommitOffsetsAsync();}}}

1. 手动同步提交

使用

KafkaConsumer#commitSync()

方法会提交

KafkaConsumer#poll()

返回的最新 offset。

同步提交，等待直到 offset 被成功提交才返回

示例：

// 这里的参数指的是轮询的时间间隔，也就是多长时间去拉一次数据ConsumerRecords<String,String> records = consumer.poll(Duration.ofMillis(3000));
records.forEach((ConsumerRecord<String,String>record)->{// 模拟消息的处理逻辑System.out.println("revice: key ==="+record.key()+" value ===="+record.value()+" topic ==="+record.topic());});try{//处理完当前批次的消息，在轮询更多的消息之前，调用commitSync方法提交当前批次最新的消息
      consumer.commitSync();}catch(CommitFailedException e){//todo 事务回滚
      e.printStackTrace();}

手动同步提交可以在任何时候提交offset，例如可以每消费一条进行一次提交。提交失败之后会抛出异常，可以在异常中做出补偿机制，例如事务回滚等操作。

但是因为手动同步提交是阻塞性质的，所以不建议太高的频率进行提交。

1. 手动异步提交

使用

KafkaConsumer#commitAsync()

方法。

示例：

while(true){// 这里的参数指的是轮询的时间间隔，也就是多长时间去拉一次数据ConsumerRecords<String,String> records = consumer.poll(Duration.ofMillis(3000));
    records.forEach((ConsumerRecord<String,String>record)->{System.out.println("revice: key ==="+record.key()+" value ===="+record.value()+" topic ==="+record.topic());});
      consumer.commitAsync();}

但是commitAsync出现问题不会自动重试，同步提交会一直进行自动重试。所以需要自己处理异常。

hile (true){// 这里的参数指的是轮询的时间间隔，也就是多长时间去拉一次数据ConsumerRecords<String,String> records = consumer.poll(Duration.ofMillis(3000));
    records.forEach((ConsumerRecord<String,String>record)->{System.out.println("revice: key ==="+record.key()+" value ===="+record.value()+" topic ==="+record.topic());});// 异步回调机制
      consumer.commitAsync(newOffsetCommitCallback(){@OverridepublicvoidonComplete(Map<TopicPartition,OffsetAndMetadata> offsets,Exception exception){if(exception!=null){System.out.println(String.format("提交失败:%s", offsets.toString()));}}});}

1. 异步加同步

可以使用同步+异步的形式保证数据能够准确提交。

while(true){ConsumerRecords records = consumer.poll(100);for(ConsumerRecordrecord: records){
        log.trace("Kafka消费信息ConsumerRecord={}",record.toString());}try{//先使用异步提交机制
        consumer.commitAsync();}catch(CommitFailedException e){// todo 补偿机制
        log.error("commitAsync failed", e)}finally{try{//再使用同步提交机制
        consumer.commitSync();}catch(CommitFailedException e){// todo 补偿机制
          log.error("commitAsync failed", e)}finally{
         consumer.close();}}}

再平衡

何时发生再平衡

重平衡其实就是一个协议，它规定了如何让消费者组下的所有消费者来分配topic中的每一个分区。比如一个topic有100个分区，一个消费者组内有20个消费者，在协调者的控制下让组内每一个消费者分配到5个分区，这个分配的过程就是重平衡。

重平衡的触发条件主要有三个：

1. 消费者组内成员发生变更，这个变更包括了增加和减少消费者，比如消费者宕机退出消费组。
2. 主题的分区数发生变更，kafka目前只支持增加分区，当增加的时候就会触发重平衡
3. 订阅的主题发生变化，当消费者组使用正则表达式订阅主题，而恰好又新建了对应的主题，就会触发重平衡

示例：

如下是正常消费的情况：

1. 消费者宕机，退出消费组，触发再平衡，重新给消费组中的消费者分配分区。如下:

1. 由于broker宕机，主题X的分区3宕机，此时分区3没有Leader副本，触发再平衡，消费者4没有对应的主题分区，则消费者4闲置。如下：

1. 主题增加分区，需要主题分区和消费组进行再均衡。如下：

1. 由于使用正则表达式订阅主题，当增加的主题匹配正则表达式的时候，也要进行再均衡。如下：

为什么说再平衡为人诟病呢？因为重平衡过程中，消费者无法从kafka消费消息，这对kafka的TPS影响极大，而如果kafka集内节点较多，比如数百个，那重平衡可能会耗时极多。数分钟到数小时都有可能，而这段时间kafka基本处于不可用状态。所以在实际环境中，应该尽量避免重平衡发生。

避免重平衡

要说完全避免重平衡，是不可能，因为你无法完全保证消费者不会故障。而消费者故障其实也是最常见的引发重平衡的地方，所以我们需要保证尽力避免消费者故障。

而其他几种触发重平衡的方式，增加分区，或是增加订阅的主题，抑或是增加消费者，更多的是主动控制。

如果消费者真正挂掉了，就没办法了，但实际中，会有一些情况，kafka错误地认为一个正常的消费者已经挂掉了，我们要的就是避免这样的情况出现。

首先要知道哪些情况会出现错误判断挂掉的情况。在分布式系统中，通常是通过心跳来维持分布式系统的，kafka也不例外。

在分布式系统中，由于网络问题你不清楚没接收到心跳，是因为对方真正挂了还是只是因为负载过重没来得及发生心跳或是网络堵塞。所以一般会约定一个时间，超时即判定对方挂了。而在kafka消费者场中，

session.timout.ms

参数就是规定这个超时时间是多少。

还有一个参数，

heartbeat.interval.ms

，这个参数控制发送心跳的频率，频率越高越不容易被误判，但也会消耗更多资源。

此外，还有最后一个参数，

max.poll.interval.ms

，消费者poll数据后，需要一些处理，再进行拉取。如果两次拉取时间间隔超过这个参数设置的值，那么消费者就会被踢出消费者组。也就是说，拉取，然后处理，这个处理的时间不能超过 max.poll.interval.ms 这个参数的值。这个参数的默认值是5分钟，而如果消费者接收到数据后会执行耗时的操作，则应该将其设置得大一些。

总结下三个参数：

• session.timout.ms控制心跳超时时间
• heartbeat.interval.ms控制心跳发送频率
• max.poll.interval.ms控制poll的间隔

这里给出一个相对较为合理的配置，如下：

• session.timout.ms：设置为6s
• heartbeat.interval.ms：设置2s
• max.poll.interval.ms：推荐为消费者处理消息最长耗时再加1分钟

再平衡原理

再均衡（Rebalance）本质上是一种协议，规定了一个消费组中所有消费者如何达成一致来分配订阅主题的每个分区。

比如某个消费组有20个消费组，订阅了一个具有100个分区的主题。正常情况下，Kafka平均会为每个消费者分配5个分区。这个分配的过程就叫再均衡。

再平衡时如何行组内分区分配？

三种分配策略：RangeAssignor和RoundRobinAssignor以及StickyAssignor。后面讲。

谁来执行再均衡和消费组管理？

Kafka提供了一个角色：Group Coordinator来执行对于消费组的管理。

Group Coordinator——每个消费组分配一个消费组协调器用于组管理和位移管理。当消费组的第一个消费者启动的时候，它会去和Kafka Broker确定谁是它们组的组协调器。之后该消费组内所有消费者和该组协调器协调通信。

如何确定coordinator？

1. 消费者向任意一个Broker节点发送包含其group.id的findCoordinator请求，以获取负责其组的Group Coordinator。
2. 收到请求的Broker节点会根据以下步骤计算该消费者组对应的Coordinator所属的Broker节点编号：首先计算groupId的hashCode，然后对总的分区数取模。比如group.id='stock’的hashCode是109770518，最终就会在（109770518 % 50 = 18）号分区保存消费位移。

Rebalance Generation

它表示Rebalance之后主题分区到消费组中消费者映射关系的一个版本，主要是用于保护消费组，隔离无效偏移量提交的。如上一个版本的消费者无法提交位移到新版本的消费组中，因为映射关系变了，你消费的或许已经不是原来的那个分区了。每次group进行Rebalance之后，Generation号都会加1，表示消费组和分区的映射关系到了一个新版本，如下图所示： Generation 1时group有3个成员，随后成员2退出组，消费组协调器触发Rebalance，消费组进入Generation 2，之后成员4加入，再次触发Rebalance，消费组进入Generation 3。

协议(protocol)

kafka提供了5个协议来处理与消费组协调相关的问题：

1. Heartbeat请求：consumer需要定期给组协调器发送心跳来表明自己还活着
2. LeaveGroup请求：主动告诉组协调器我要离开消费组
3. SyncGroup请求：消费组Leader把分配方案告诉组内所有成员
4. JoinGroup请求：成员请求加入组
5. DescribeGroup请求：显示组的所有信息，包括成员信息，协议名称，分配方案，订阅信息等。通常该请求是给管理员使用

组协调器在再均衡的时候主要用到了前面4种请求。

消费者如何向消费组协调器证明自己还活着？

通过定时向消费组协调器发送Heartbeat请求。如果超过了设定的超时时间，那么协调器认为该消费者已经挂了。一旦协调器认为某个消费者挂了，那么它就会开启新一轮再均衡，并且在当前其他消费者的心跳响应中添加“REBALANCE_IN_PROGRESS”，告诉其他消费者：重新分配分区。

再均衡过程

再均衡分为2步：Join和Sync

Join：加入组。所有成员都向消费组协调器发送JoinGroup请求，请求加入消费组。一旦所有成员都发送了JoinGroup请求，协调器从中选择一个消费者担任Leader的角色，并把组成员信息以及订阅信息发给Leader。

Sync：Leader开始分配消费方案，即哪个消费者负责消费哪些主题的哪些分区。一旦完成分配，Leader会将这个方案封装进SyncGroup请求中发给消费组协调器，非Leader也会发SyncGroup请求，只是内容为空。消费组协调器接收到分配方案之后会把方案塞进SyncGroup的response中发给各个消费者。

注意：在协调器收集到所有成员请求前，它会把已收到请求放入一个叫purgatory的地方。

然后是分发分配方案的过程，即SyncGroup请求：

注意：消费组的分区分配方案在客户端执行。Kafka交给客户端可以有更好的灵活性。Kafka默认提供三种分配策略：range和round-robin和sticky。可以通过消费者的参数：

partition.assignment.strategy

来实现自己分配策略。

消费组状态机

消费组组协调器根据状态机对消费组做不同的处理：

说明：

1. Dead：组内已经没有任何成员的最终状态，组的元数据也已经被组协调器移除了。这种状态响应各种请求都是一个response： UNKNOWN_MEMBER_ID
2. Empty：组内无成员，但是位移信息还没有过期。这种状态只能响应JoinGroup请求
3. PreparingRebalance：组准备开启新的rebalance，等待成员加入
4. AwaitingSync：正在等待leader consumer将分配方案传给各个成员
5. Stable：再均衡完成，可以开始消费。

参数配置

消费者常用参数：

1. bootstrap.servers：建立到Kafka集群的初始连接用到的host/port列表。客户端会使用这里指定的所有的host/port来建立初始连接。这个配置仅会影响发现集群所有节点的初始连接。形式：host1:port1,host2:port2…。这个配置中不需要包含集群中所有的节点信息。最好不要配置一个，以免配置的这个节点宕机的时候连不上。
2. group.id：用于定义当前消费者所属的消费组的唯一字符串。如果使用了消费组的功能 subscribe(topic) ，或使用了基于Kafka的偏移量管理机制，则应该配置group.id。
3. auto.commit.interval.ms ：如果设置了 enable.auto.commit 的值为true，则该值定义了消费者偏移量向Kafka提交的频率。
4. auto.offset.reset：如果Kafka中没有初始偏移量或当前偏移量在服务器中不存在（比如数据被删掉了）：earliest：自动重置偏移量到最早的偏移量。latest：自动重置偏移量到最后一个none：如果没有找到该消费组以前的偏移量没有找到，就抛异常。其他值：向消费者抛异常。
5. fetch.min.bytes：服务器对每个拉取消息的请求返回的数据量最小值。如果数据量达不到这个值，请求等待，以让更多的数据累积，达到这个值之后响应请求。默认设置是1个字节，表示只要有一个字节的数据，就立即响应请求，或者在没有数据的时候请求超时。将该值设置为大一点儿的数字，会让服务器等待稍微长一点儿的时间以累积数据。如此则可以提高服务器的吞吐量，代价是额外的延迟时间。
6. fetch.max.wait.ms：如果服务器端的数据量达不到 fetch.min.bytes 的话，服务器端不能立即响应请求。该时间用于配置服务器端阻塞请求的最大时长
7. fetch.max.bytes：服务器给单个拉取请求返回的最大数据量。消费者批量拉取消息，如果第一个非空消息批次的值比该值大，消息批也会返回，以让消费者可以接着进行。即该配置并不是绝对的最大值。broker可以接收的消息批最大值通过message.max.bytes (broker配置) 或 max.message.bytes (主题配置)来指定。需要注意的是，消费者一般会并发拉取请求。
8. enable.auto.commit ：如果设置为true，则消费者的偏移量会周期性地在后台提交。
9. connections.max.idle.ms ：在这个时间之后关闭空闲的连接。
10. isolation.level：控制如何读取事务消息。如果设置了 read_committed ，消费者的poll()方法只会返回已经提交的事务消息。如果设置了 read_uncommitted (默认值)，消费者的poll方法返回所有的消息，即使是已经取消的事务消息。非事务消息以上两种情况都返回。消息总是以偏移量的顺序返回。read_committed 只能返回到达LSO的消息。在LSO之后出现的消息只能等待相关的事务提交之后才能看到结果， read_committed 模式，如果有未提交的事务，消费者不能读取到直到HW的消息。read_committed 的seekToEnd方法返回LSO。
11. heartbeat.interval.ms：当使用消费组的时候，该条目指定消费者向消费者协调器发送心跳的时间间隔。心跳是为了确保消费者会话的活跃状态，同时在消费者加入或离开消费组的时候方便进行再平衡。该条目的值必须小于 session.timeout.ms ，也不应该高于session.timeout.ms 的1/3。可以将其调整得更小，以控制正常重新平衡的预期时间。
12. session.timeout.ms：当使用Kafka的消费组的时候，消费者周期性地向broker发送心跳数表明自己的存在。如果经过该超时时间还没有收到消费者的心跳，则broker将消费者从消费组移除，并启动再平衡。该值必须在broker配group.min.session.timeout.ms 和 group.max.session.timeout.ms 之间。
13. max.poll.records ：一次调用poll()方法返回的记录最大数量。
14. max.poll.interval.ms：使用消费组的时候调用poll()方法的时间间隔。该条目指定了消费者调用poll()方法的最大时间间隔。如果在此时间内消费者没有调用poll()方法，则broker认为消费者失败，触发再平衡，将分区分配给消费组中其他消费者。
15. max.partition.fetch.bytes：对每个分区，服务器返回的最大数量。消费者按批次拉取数据。如果非空分区的第一个记录大于这个值，批处理依然可以返回，以保证消费者可以进行下去。broker接收批的大小由 message.max.bytes （broker参数）或max.message.bytes （主题参数）指定。fetch.max.bytes 用于限制消费者单次请求的数据量。
16. send.buffer.bytes ：用于TCP发送数据时使用的缓冲大小（SO_SNDBUF），-1表示使用OS默认的缓冲区大小。
17. receive.buffer.bytes ：TCP连接接收数据的缓存（SO_RCVBUF）。-1表示使用操作系统的默认值。
18. retry.backoff.ms：在发生失败的时候如果需要重试，则该配置表示客户端等待多长时间再发起重试。该时间的存在避免了密集循环。
19. request.timeout.ms：客户端等待服务端响应的最大时间。如果该时间超时，则客户端要么重新发起请求，要么如果重试耗尽，请求失败。
20. reconnect.backoff.ms：重新连接主机的等待时间。避免了重连的密集循环。该等待时间应用于该客户端到broker的所有连接。
21. reconnect.backoff.max.ms：重新连接到反复连接失败的broker时要等待的最长时间（以毫秒为单位）。如果提供此选项，则对于每个连续的连接失败，每台主机的退避将成倍增加，直至达到此最大值。在计算退避增量之后，添加20％的随机抖动以避免连接风暴。
22. partition.assignment.strategy：当使用消费组的时候，分区分配策略的类名。
23. interceptor.classes：
24. 拦截器类的列表。默认没有拦截器拦截器是消费者的拦截器，该拦截器需要实现org.apache.kafka.clients.consumer.ConsumerInterceptor 接口。拦截器可用于对消费者接收到的消息进行拦截处理。