Kafka消费者原理与实践本篇介绍Kafka消费者的工作原理与核心机制包括消费者组原理、分区分配与再平衡策略、offset管理、漏消费与重复消费问题及解决方案以及消费者吞吐量调优。参考课程尚硅谷大数据技术之KafkaV3.0.01. 消费方式Kafka消费者从Broker拉取数据有两种可能的模式push推模式由Broker主动向消费者推送数据。缺点是Broker决定发送速率很难适应所有消费者的消费能力消费速度慢的消费者容易被压垮。pull拉模式消费者主动从Broker拉取数据。Kafka采用这种方式消费者可以根据自身处理能力控制拉取速率灵活适配不同场景。pull模式的不足如果Kafka中没有数据消费者会陷入循环空轮询一直返回空数据。Kafka通过fetch.max.wait.ms参数解决这个问题消费者在没有拉取到足够数据时会等待一段时间再返回避免无意义的空轮询。2. 消费者工作流程2.1 总体工作流程消费者从Leader副本拉取数据每个消费者的offset由消费者自己维护提交到Kafka内置的系统主题__consumer_offsets中保存Kafka 0.9版本之前保存在Zookeeper中。核心规则每个分区的数据只能由消费者组中的一个消费者消费但一个消费者可以消费多个分区的数据。2.2 消费者组原理Consumer GroupCG由多个Consumer组成同一个组内所有消费者的group.id相同。消费者组内每个消费者负责消费不同分区的数据一个分区只能由组内一个消费者消费消费者组之间互不影响同一份数据可以被多个消费者组独立消费如果消费者数量超过分区数多余的消费者会闲置不会接收任何消息coordinator辅助实现消费者组初始化和分区分配的组件每个Broker上都有一个coordinator。消费者组对应哪个coordinator由以下公式决定coordinator节点 groupid的hashcode值 % __consumer_offsets的分区数默认50例如groupid的hashcode为1则1 % 50 1__consumer_offsets主题的1号分区在哪个Broker上该Broker上的coordinator就负责这个消费者组。2.3 消费者组初始化流程每个Consumer向coordinator发送JoinGroup请求coordinator从中选出一个Consumer作为Leadercoordinator把要消费的Topic情况发送给Leader消费者Leader制定消费方案哪个消费者消费哪个分区Leader把消费方案发回给coordinatorcoordinator将消费方案下发给各个Consumer每个Consumer与coordinator保持心跳默认3s超时默认45s则该Consumer被移除触发再平衡消费者处理消息时间过长默认5分钟也会触发再平衡2.4 消费者详细消费流程消费者内部通过ConsumerNetworkClient发送消费请求拉取到的数据放入completedFetches队列再通过反序列化、拦截器处理后交给业务逻辑处理。关键参数控制拉取行为Fetch.min.bytes每批次最小拉取数据量默认1字节fetch.max.wait.ms未达到最小拉取量时的最大等待时间默认500msFetch.max.bytes每批次最大拉取数据量默认50MBMax.poll.records一次poll返回的最大消息条数默认500条3. 消费者重要参数参数名默认值说明bootstrap.servers—连接Kafka集群的Broker地址列表group.id—消费者组ID必须配置enable.auto.committrue是否自动提交offsetauto.commit.interval.ms5s自动提交offset的时间间隔auto.offset.resetlatest无初始offset或offset不存在时的处理策略earliest从最早开始消费latest从最新开始消费none抛出异常offsets.topic.num.partitions50__consumer_offsets的分区数heartbeat.interval.ms3s消费者与coordinator的心跳间隔需小于session.timeout.ms的1/3session.timeout.ms45s消费者与coordinator连接超时时间超过则移除该消费者并触发再平衡max.poll.interval.ms5分钟消费者处理消息的最大时长超过则移除并触发再平衡fetch.min.bytes1字节每批次最小拉取数据量fetch.max.wait.ms500ms未达到最小拉取量时的最大等待时间fetch.max.bytes50MB每批次最大拉取数据量max.poll.records500条一次poll返回的最大消息条数4. 消费者API4.1 订阅主题PropertiespropertiesnewProperties();properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,hadoop102:9092);// key和value反序列化必须填全类名properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,StringDeserializer.class.getName());properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,StringDeserializer.class.getName());// 配置消费者组ID必须配置properties.put(ConsumerConfig.GROUP_ID_CONFIG,test);KafkaConsumerString,StringconsumernewKafkaConsumer(properties);// 订阅主题可以同时订阅多个consumer.subscribe(Arrays.asList(first));while(true){// 每隔1s拉取一批数据ConsumerRecordsString,Stringrecordsconsumer.poll(Duration.ofSeconds(1));for(ConsumerRecordString,Stringrecord:records){System.out.println(record);}}4.2 订阅分区如果只想消费某个Topic的特定分区使用assign()代替subscribe()// 只消费first主题的0号分区ListTopicPartitionpartitionsnewArrayList();partitions.add(newTopicPartition(first,0));consumer.assign(partitions);4.3 消费者组案例将多个消费者的group.id设置为相同值同时启动即可组成消费者组。组内各消费者会自动分配分区每个分区只由其中一个消费者消费。如果消费者数量超过分区数多余的消费者会闲置。5. 分区分配及再平衡消费者组内有多个Consumer一个Topic有多个Partition到底由哪个Consumer消费哪个Partition由分区分配策略决定。Kafka默认策略为Range CooperativeSticky可通过partition.assignment.strategy参数修改支持同时使用多种策略。5.1 Range策略Range是针对每个Topic单独计算的。分配规则将同一Topic的分区按序号排序消费者按字母顺序排序然后用分区数 / 消费者数决定每个消费者消费几个分区除不尽时前面的消费者多消费1个。例如7个分区3个消费者7/32余1则Consumer0消费3个0、1、2Consumer1消费2个3、4Consumer2消费2个5、6。缺点Topic越多Consumer0多消费的分区就越多容易造成数据倾斜。再平衡某个Consumer下线后消费者组需要等待session.timeout.ms默认45s确认其退出才会把任务重新按Range方式分配给其他Consumer。5.2 RoundRobin策略RoundRobin针对集群中所有Topic整体轮询分配。将所有Partition和所有Consumer都列出来按hashcode排序后通过轮询算法依次分配。相比RangeRoundRobin分配更均匀不容易产生数据倾斜。properties.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG,org.apache.kafka.clients.consumer.RoundRobinAssignor);5.3 Sticky策略粘性分区在执行新一轮分配之前尽量保留上一次的分配结果只做最小幅度的调整。首先尽量均衡地分配分区当某个Consumer下线触发再平衡时只把该Consumer的分区随机分给其他Consumer其他Consumer原有的分区分配不变。相比Range和RoundRobin每次再平衡都全部重新分配Sticky策略的再平衡开销更小。ListStringstrategiesnewArrayList();strategies.add(org.apache.kafka.clients.consumer.StickyAssignor);properties.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG,strategies);5.4 CooperativeSticky策略与Sticky策略类似区别在于支持协作式再平衡再平衡过程中不需要停止所有消费者只需要调整涉及变动的分区对消费过程的影响更小。Kafka默认策略Range CooperativeSticky中的CooperativeSticky即指此策略。6. offset位移6.1 offset默认维护位置Kafka 0.9版本之前offset保存在Zookeeper中。0.9版本之后offset默认保存在Kafka内置的系统主题__consumer_offsets中采用key-value方式存储key为group.id topic 分区号value为当前offset值。Kafka内部会定期对该主题进行compact保留每个key的最新数据。6.2 自动提交offset// 开启自动提交默认已开启properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,true);// 自动提交时间间隔默认5sproperties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG,1000);自动提交基于时间触发开发者无需手动管理但提交时机不可控可能导致重复消费或漏消费。6.3 手动提交offset手动提交分为同步和异步两种方式两者都会提交本次poll到的数据中最高的offset同步提交commitSync阻塞当前线程直到提交成功有失败重试机制可靠性高但吞吐量低。properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,false);while(true){ConsumerRecordsString,Stringrecordsconsumer.poll(Duration.ofSeconds(1));for(ConsumerRecordString,Stringrecord:records){System.out.println(record.value());}// 同步提交offsetconsumer.commitSync();}异步提交commitAsync发送提交请求后立即返回不等待结果没有失败重试机制吞吐量高但可能提交失败。while(true){ConsumerRecordsString,Stringrecordsconsumer.poll(Duration.ofSeconds(1));for(ConsumerRecordString,Stringrecord:records){System.out.println(record.value());}// 异步提交offsetconsumer.commitAsync();}6.4 指定offset消费auto.offset.reset参数控制当Kafka中无初始offset或当前offset不存在时的处理方式earliest从最早的offset开始消费等同于命令行的--from-beginninglatest默认从最新的offset开始消费none找不到消费者组的历史offset则抛出异常如果需要从任意指定的offset开始消费// 先订阅主题等待分区分配完成consumer.subscribe(Arrays.asList(first));SetTopicPartitionassignmentnewHashSet();while(assignment.size()0){consumer.poll(Duration.ofSeconds(1));assignmentconsumer.assignment();}// 为每个分区指定从offset1700开始消费for(TopicPartitiontp:assignment){consumer.seek(tp,1700);}注意每次执行完需要修改消费者组名否则已提交的offset会覆盖seek的效果。6.5 指定时间消费生产环境中如果某段时间的数据出现异常需要回溯到指定时间点重新消费// 获取分区分配信息SetTopicPartitionassignmentnewHashSet();while(assignment.size()0){consumer.poll(Duration.ofSeconds(1));assignmentconsumer.assignment();}// 构造每个分区对应的时间戳1天前MapTopicPartition,LongtimestampToSearchnewHashMap();for(TopicPartitiontp:assignment){timestampToSearch.put(tp,System.currentTimeMillis()-1*24*3600*1000);}// 获取对应时间戳的offsetMapTopicPartition,OffsetAndTimestampoffsetsconsumer.offsetsForTimes(timestampToSearch);// 为每个分区设置对应的offsetfor(TopicPartitiontp:assignment){OffsetAndTimestampoffsetAndTimestampoffsets.get(tp);if(offsetAndTimestamp!null){consumer.seek(tp,offsetAndTimestamp.offset());}}7. 漏消费与重复消费7.1 重复消费产生原因自动提交offset时Consumer每隔5s提交一次。如果提交后2s内Consumer挂掉重启则从上一次提交的offset继续消费导致这2s内已经消费过的数据被重复消费。7.2 漏消费产生原因手动提交offset时先提交offset再处理数据。如果offset提交后、数据还在内存中未落盘时Consumer挂掉这部分数据就永久丢失造成漏消费。7.3 消费者事务要做到既不漏消费也不重复消费精确一次语义需要将Kafka的offset提交和下游数据处理做原子绑定。具体做法是把Kafka的offset保存到支持事务的外部存储如MySQL消费数据和提交offset在同一个事务中完成要么都成功要么都回滚。这要求下游系统必须支持事务。8. 数据积压消费者如何提高吞吐量当消费速度跟不上生产速度数据开始积压时有两种处理方向方向一消费能力不足增加Topic的分区数同时增加消费者组的消费者数量消费者数 分区数两者缺一不可。单纯增加消费者但不增加分区多余的消费者会闲置不起作用。方向二下游处理能力不足提高每批次拉取的数据量让消费者一次拉取更多数据处理// 每批次最大拉取数据量默认50MBproperties.put(ConsumerConfig.FETCH_MAX_BYTES_CONFIG,52428800);// 一次poll返回的最大消息条数默认500条可适当调大properties.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG,500);小结Kafka消费者的设计核心是灵活性和可靠性的平衡消费者组机制实现了横向扩展分区分配策略在均衡性和再平衡开销之间取舍offset管理是数据可靠性的关键自动提交简单但不可控手动提交精确但需要自己处理异常精确一次消费需要结合外部事务系统单靠Kafka本身无法完全保证理解了这几个核心权衡点消费者的各种参数和机制就都有了落脚点。