kafka消费者api和分区分配和offset消费

kafka消费者

消费者的消费方式为主动从broker拉取消息，由于消费者的消费速度不同，由broker决定消息发送速度难以适应所有消费者的能力

拉取数据的问题在于，消费者可能会获得空数据

消费者组工作流程

Consumer Group（CG）：消费者组

由多个consumer组成。形成一个消费者组的条件，是所有消费者的groupid相同。
消费者组内每个消费者负责消费不同分区的数据，一个分区只能由一个组内消费者消费。
消费者组之间互不影响。所有的消费者都属于某个消费者组（即使只有一个消费者），即消费者组是逻辑上的一个订阅者
分区和消费者的分配取决于具体的分配策略
如果消费者组中的消费者数量超过分区数量，则会由部分消费者处于空闲状态，不会接受任何消息

在这里插入图片描述

初始化流程

在这里插入图片描述

每个broker上都有coordinator
选择coordinator作为消费者组的初始化和分区分配的协调者，使用消费者组的groupid的hashcode%50（即__consumer_offsets的分区数量）得到对应的broker id。该broker将作为整个消费者组的协调者。消费者组中的消费者向该分区提交offset
所有消费者向coordinator发送请求加入消费者组
coordinator随机选择一个consumer作为leader
将要消费的topic情况发送给leader消费者
leader制定消费方案，并将方案发送到coordinator
coordinator把消费方案分发给哥哥消费者
每个消费者都会和coordinator保持心跳（默认3s） - 一旦超时（session.timeout.ms=45s），该消费者会被移除，并触发再平衡；- 或者消费者处理消息的时间过长（max.poll.interval.ms 5分钟），也会触发再平衡
尽量避免消费者组的再平衡，非常消耗性能

消费流程

在这里插入图片描述

消费者创建ConsumerNetworkClient，用于和kafka集群进行通信
消费者开始初始化抓取数据的参数 - fetch.min.bytes，每批次最小抓取大小，默认1字节- fetch.max.wait.ms，超时时间即使数据批次未达到大小也会抓取，默认500ms- fetch.max.bytes，每批次最大抓取大小，默认50m
参数初始化完成后，开始调用send方法发送请求
通过onSuccess回调拉取数据，存放在消息队列中
消费者开始拉取数据（max.poll.records，一次拉取数据返回消息的最大值，默认500条）
将消息进行反序列化和拦截器（kafka本身并不处理数据）

消费者相关参数

bootstrap.servers 向 Kafka 集群建立初始连接用到的 host/port 列表。
key.deserializer 和 value.deserializer 指定接收消息的 key 和 value 的反序列化类型。一定要写全类名。
group.id 标记消费者所属的消费者组。
enable.auto.commit 默认值为 true，消费者会自动周期性地向服务器提交偏移量
auto.commit.interval.ms 如果设置了 enable.auto.commit 的值为 true，则该值定义了消费者偏移量向 Kafka 提交的频率，默认 5s
auto.offset.reset 当 Kafka 中没有初始偏移量或当前偏移量在服务器中不存在（如，数据被删除了），该如何处理？ - earliest：自动重置偏移量到最早的偏移量。- latest：默认，自动重置偏移量为最新的偏移量。- none：如果消费组原来的（previous）偏移量不存在，则向消费者抛异常。- anything：向消费者抛异常
offsets.topic.num.partitions ，即__consumer_offsets 的分区数，默认是 50 个分区。
heartbeat.interval.ms Kafka 消费者和 coordinator 之间的心跳时间，默认 3s。该条目的值必须小于 session.timeout.ms (45s)，也不应该高于 session.timeout.ms 的 1/3。
session.timeout.ms Kafka 消费者和 coordinator 之间连接超时时间，默认 45s。超过该值，该消费者被移除，消费者组执行再平衡

消费者API

创建topic

kafka-topics.sh --bootstrap-server 127.0.0.1:9092 --topic first --create --partitions 3 --replication-factor 3

独立消费者

即使只有单独的消费者，也必须配置消费者组id
kafka命令行启动消费者，如果不填写消费者组id，则会被自动填充随机的消费者组id

订阅主题进行消费

importjava.time.Duration;importjava.util.ArrayList;importjava.util.Properties;importorg.apache.kafka.clients.consumer.ConsumerConfig;importorg.apache.kafka.clients.consumer.ConsumerRecord;importorg.apache.kafka.clients.consumer.ConsumerRecords;importorg.apache.kafka.clients.consumer.KafkaConsumer;importorg.apache.kafka.clients.producer.ProducerConfig;importorg.apache.kafka.common.serialization.StringDeserializer;publicclassSimpleConsumer{publicstaticvoidmain(String[] args){// configureProperties properties =newProperties();// connect
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"localhost:9092");// key，value反序列化
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,StringDeserializer.class.getName());
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,StringDeserializer.class.getName());// create consumer
        properties.put(ConsumerConfig.GROUP_ID_CONFIG,"test");KafkaConsumer<String,String> kafkaConsumer =newKafkaConsumer<String,String>(properties);ArrayList<String> topics =newArrayList<String>();
        topics.add("first");
        kafkaConsumer.subscribe(topics);while(true){ConsumerRecords<String,String> consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(1));for(ConsumerRecord<String,String>ConsumerRecord: consumerRecords){System.out.println(ConsumerRecord);}}}}
output:ConsumerRecord(topic = test, partition =0, leaderEpoch =2,offset =3,CreateTime=1629169606820, serialized key size =-1,serialized value size =8, headers =RecordHeaders(headers =[],isReadOnly =false), key =null, value = hello1)ConsumerRecord(topic = test, partition =1, leaderEpoch =3,offset =2,CreateTime=1629169609524, serialized key size =-1,serialized value size =6, headers =RecordHeaders(headers =[],isReadOnly =false), key =null, value = hello2)

订阅分区进行消费

...ArrayList<TopicPartition> topics =newArrayList<TopicPartition>();
topics.add(newTopicPartition("test",0));// 指定消费分区0的数据
kafkaConsumer.assign(topics);

消费者组消费数据

只需要启动多个消费者即可，消费者按照消费者组的id自动归属于同一个消费者组

properties.put(ConsumerConfig.GROUP_ID_CONFIG,"test");KafkaConsumer<String,String> kafkaConsumer =newKafkaConsumer<String,String>(properties);ArrayList<String> topics =newArrayList<String>();
topics.add("first");
kafkaConsumer.subscribe(topics);

分区的分配和再平衡

分区的分配设计到同一个topic中的partition由那个consumer来消费的问题

Kafka有四种主流的分区分配策略（所谓的分区分配策略就是消费方案）：

Range
RoundRobin
Sticky
CooperativeSticky

可以通过配置参数

partition.assignment.strategy

，修改分区的分配策略。默认策略是

Range

CooperativeSticky

。Kafka可以同时使用多个分区分配策略

range策略

Range 是对每个 topic 而言的。

对同一个 topic 里面的分区按照序号进行排序，假如现在有 7 个分区，3 个消费者，排序后的分区将会是0,1,2,3,4,5,6
对消费者按照字母顺序进行排序。消费者排序完之后将会是C0,C1,C2。
通过 partitions数/consumer数来决定每个消费者应该消费几个分区。如果除不尽，那么前面几个消费者将会多消费 1 个分区。

注意：如果只是针对 1 个 topic 而言，C0消费者多消费1 个分区影响不是很大。但是如果有 N 个 topic，那么针对每个 topic，消费者 C0都将多消费 1 个分区，topic越多，C0消费的分区会比其他消费者明显多消费 N 个分区。 容易产生数据倾斜

在这里插入图片描述

注意：

修改主题的分区数，只能增加不能减少
如果在消费过程中某个consumer挂掉，当超出45s后，则该consumer消费的所有分区都会整体分配给某一个其他消费者
消费者被移出消费者组，消费策略按照存活的消费者重新分配分区
Kafka 默认的分区分配策略是 Range + CooperativeSticky

RoundRobin策略

针对所有topic而言

把所有的 partition 和所有的 consumer 都列出来，然后按照 hashcode 进行排序
按照轮询算法将partition分配给消费者

注意

需要修改分区分配策略properties.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, "org.apache.kafka.clients.consumer.RoundRobinAssignor");
如果在消费过程中某个consumer挂掉，超过45s后，该消费者的分区会重新按照轮询的方式在其他消费者中分配
消费者被移出消费者组，消费策略按照存活的消费者重新分配分区

Sticky策略

在执行一次新的分配之前，考虑上一次分配的结果，尽量少的调整分配的变动，可以节省大量的开销

粘性分区是 Kafka 从 0.11.x 版本开始引入这种分配策略，首先会尽量均衡的放置分区到消费者上面，在出现同一消费者组内消费者出现问题的时候，会尽量保持原有分配的分区不变化

如果有0,1,2,3,4,5,6分区和C0,C1,C2消费者，则最终分配比例仍旧是223，但是每个消费者分配中的partition是随机的

注意

需要修改分区分配策略ArrayList<String> startegys =newArrayList<>();startegys.add("org.apache.kafka.clients.consumer.StickyAssignor");properties.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG,startegys);
如果在消费过程中某个consumer挂掉，超过45s后，该消费者的分区会按照粘性规则，尽可能均衡分配给其他的消费者
消费者被移出消费者组，消费策略按照存活的消费者重新分配分区

offset位移

offset维护的位置在不同版本的kafka中存在区别

0.9版本之前存储在zk中，如果client和zk之前存在大量网络通信，则会导致性能我呢提
0.9版本后存储在kafka集群中的_consumer_offsets主题中

在内部主题中采用kv的方式存储offset

key的值为，group.id+topic+ 分区号
value的值为，当前offset
每隔一段时间，kafka对主题中的数据进行compact

默认内部主题不可消费

修改config/comsumer.properties文件中的参数exclude.internal.topics=false, 默认是 true，表示不能消费系统主题
查看消费者消费主题kafka-console-consumer.sh --topic __consumer_offsets --bootstrap-server 127.0.0.1:9092 --consumer.config config/consumer.properties --formatter"kafka.coordinator.group.GroupMetadataManager\$OffsetsMessageFormatter" --from-beginning

自动提交offset

kafka提供了自动提交offset的功能，使用户专注于自身的业务逻辑

enable.auto.commit：是否开启自动提交offset功能，默认是true
auto.commit.interval.ms：自动提交offset的时间间隔，默认是5s

在这里插入图片描述

在java消费者中添加消费者参数

// 是否自动提交 offset，实际上不用设置此参数，默认为true
properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,true);// 提交 offset 的时间周期 1000ms，默认 5s
properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG,1000);

手动提交offset

手动提交offset的方法有两种：分别是

commitSync

（同步提交）和

commitAsync

（异步提交）

相同点：都会将本次提交的一批数据最高的偏移量提交

不同点：同步提交阻塞当前线程，一直到提交成功，并且会自动失败重试（由不可控因素导致，也会出现提交失败）；而异步提交则没有失败重试机制，故有可能提交失败

commitSync（同步提交）：必须等待offset提交完毕，再去消费下一批数据
commitAsync（异步提交）：发送完提交offset请求后，就开始消费下一批数据

在这里插入图片描述

在java消费者中添加消费者参数

同步提交存在重试机制，因此更加可靠，但是由于阻塞提交效率较低（吞吐量低）

// 是否自动提交 offset
properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,false);//4. 设置消费主题 形参是列表
consumer.subscribe(Arrays.asList("first"));//5. 消费数据while(true){// 读取消息ConsumerRecords<String,String> consumerRecords = consumer.poll(Duration.ofSeconds(1));// 输出消息for(ConsumerRecord<String,String> consumerRecord :
consumerRecords){System.out.println(consumerRecord.value());}// 同步提交 offset
     consumer.commitSync();// 异步提交 offset// consumer.commitAsync();}

指定offset消费

在命令行中创建消费者指定

--from-beginning

，表示从头开始消费。

当消费者组首次消费（没有初始偏移量时），根据以下参数进行消费行为

earliest，将偏移量重置为从头开始消费
latest（默认值），自动将偏移量重置为最新偏移量
none：如果未找到消费者组的先前偏移量，则向消费者抛出异常

如何在java消费者中指定offset进行消费

// 1 创建一个消费者KafkaConsumer<String,String> kafkaConsumer =newKafkaConsumer<>(properties);// 订阅主题ArrayList<String> topics =newArrayList<>();
topics.add("first");
kafkaConsumer.subscribe(topics);Set<TopicPartition> assignment=newHashSet<>();while(assignment.size()==0){
    kafkaConsumer.poll(Duration.ofSeconds(1));// 获取消费者分区分配信息（有了分区分配信息才能开始消费）
    assignment = kafkaConsumer.assignment();}// 遍历所有分区，并指定 offset 从 1700 的位置开始消费for(TopicPartition tp: assignment){
    kafkaConsumer.seek(tp,1700);}// 开始消费while(true){ConsumerRecords<String,String> consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(1));for(ConsumerRecord<String,String> consumerRecord : consumerRecords){System.out.println(consumerRecord);}}

指定时间开始消费

逻辑上可以通过指定某个时刻的offset来实现

Set<TopicPartition> assignment =newHashSet<>();while(assignment.size()==0){
    kafkaConsumer.poll(Duration.ofSeconds(1));// 获取消费者分区分配信息（有了分区分配信息才能开始消费）
    assignment = kafkaConsumer.assignment();}HashMap<TopicPartition,Long> timestampToSearch =newHashMap<>();// 封装集合存储，每个分区对应一天前的数据for(TopicPartition topicPartition : assignment){
    timestampToSearch.put(topicPartition,System.currentTimeMillis()-1*24*3600*1000);}// 获取从 1 天前开始消费的每个分区的 offsetMap<TopicPartition,OffsetAndTimestamp> offsets = kafkaConsumer.offsetsForTimes(timestampToSearch);// 遍历每个分区，对每个分区设置消费时间。for(TopicPartition tp : assignment){OffsetAndTimestamp offsetAndTimestamp = offsets.get(tp);// 根据时间指定开始消费的位置if(offsetAndTimestamp !=null){
        kafkaConsumer.seek(tp, offsetAndTimestamp.offset());}}

漏消费和重复消费问题

重复消费问题，当前一次自动提交offset后，消费者开始消费数据2s后挂掉。此时重启consumer会从上一次自动提交的offset开始消费，导致重复消费的问题

在这里插入图片描述

漏消费问题，在手动提交offset模式下，当提交offset后如果消费者数据还未落盘出现宕机，则这部分未落盘的数据由于offset已经更新无法再次消费

在这里插入图片描述

生产环境的消费者

消费者事务

控制consumer端精准消费同样需要事务支持（要Kafka消费端将消费过程和提交offset 过程做原子绑定）

此时需要将offset保存到支持事务的介质中

在这里插入图片描述

数据积压问题

消费者能力不足造成积压（考虑扩充消费者数量）

下游数据处理不及时导致数据积压，提升每批次拉取数据的量

kafka消费者api和分区分配和offset消费

kafka消费者

消费者组工作流程

初始化流程

消费流程

消费者相关参数

消费者API

分区的分配和再平衡

range策略

RoundRobin策略

Sticky策略

offset位移

自动提交offset

手动提交offset

指定offset消费

指定时间开始消费

漏消费和重复消费问题

生产环境的消费者

消费者事务

数据积压问题

发表评论

“kafka消费者api和分区分配和offset消费”的评论:

关于作者

overfit同步小助手

相关阅读

文章导航