Kafka消费者原理与实践

本篇介绍Kafka消费者的工作原理与核心机制，包括消费者组原理、分区分配与再平衡策略、offset管理、漏消费与重复消费问题及解决方案，以及消费者吞吐量调优。

参考课程：尚硅谷大数据技术之Kafka（V3.0.0）

1. 消费方式

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 Group（CG）：由多个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 消费者组初始化流程

1. 每个Consumer向coordinator发送JoinGroup请求
2. coordinator从中选出一个Consumer作为Leader
3. coordinator把要消费的Topic情况发送给Leader消费者
4. Leader制定消费方案（哪个消费者消费哪个分区）
5. Leader把消费方案发回给coordinator
6. coordinator将消费方案下发给各个Consumer
7. 每个Consumer与coordinator保持心跳（默认3s），超时（默认45s）则该Consumer被移除，触发再平衡；消费者处理消息时间过长（默认5分钟）也会触发再平衡
   

2.4 消费者详细消费流程

消费者内部通过ConsumerNetworkClient发送消费请求，拉取到的数据放入completedFetches队列，再通过反序列化、拦截器处理后交给业务逻辑处理。

关键参数控制拉取行为：

• Fetch.min.bytes：每批次最小拉取数据量，默认1字节
• fetch.max.wait.ms：未达到最小拉取量时的最大等待时间，默认500ms
• Fetch.max.bytes：每批次最大拉取数据量，默认50MB
• Max.poll.records：一次poll返回的最大消息条数，默认500条

3. 消费者重要参数

4. 消费者API

4.1 订阅主题

Propertiesproperties=newProperties();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");KafkaConsumer<String,String>consumer=newKafkaConsumer<>(properties);// 订阅主题，可以同时订阅多个consumer.subscribe(Arrays.asList("first"));while(true){// 每隔1s拉取一批数据ConsumerRecords<String,String>records=consumer.poll(Duration.ofSeconds(1));for(ConsumerRecord<String,String>record:records){System.out.println(record);}}

4.2 订阅分区

如果只想消费某个Topic的特定分区，使用assign()代替subscribe()：

// 只消费first主题的0号分区List<TopicPartition>partitions=newArrayList<>();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/3=2余1，则Consumer0消费3个（0、1、2），Consumer1消费2个（3、4），Consumer2消费2个（5、6）。

缺点：Topic越多，Consumer0多消费的分区就越多，容易造成数据倾斜。

再平衡：某个Consumer下线后，消费者组需要等待session.timeout.ms（默认45s）确认其退出，才会把任务重新按Range方式分配给其他Consumer。

5.2 RoundRobin策略

RoundRobin针对集群中所有Topic整体轮询分配。

将所有Partition和所有Consumer都列出来，按hashcode排序后，通过轮询算法依次分配。相比Range，RoundRobin分配更均匀，不容易产生数据倾斜。

properties.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG,"org.apache.kafka.clients.consumer.RoundRobinAssignor");

5.3 Sticky策略

粘性分区：在执行新一轮分配之前，尽量保留上一次的分配结果，只做最小幅度的调整。

首先尽量均衡地分配分区，当某个Consumer下线触发再平衡时，只把该Consumer的分区随机分给其他Consumer，其他Consumer原有的分区分配不变。相比Range和RoundRobin每次再平衡都全部重新分配，Sticky策略的再平衡开销更小。

List<String>strategies=newArrayList<>();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){ConsumerRecords<String,String>records=consumer.poll(Duration.ofSeconds(1));for(ConsumerRecord<String,String>record:records){System.out.println(record.value());}// 同步提交offsetconsumer.commitSync();}

异步提交（commitAsync）：发送提交请求后立即返回，不等待结果，没有失败重试机制，吞吐量高但可能提交失败。

while(true){ConsumerRecords<String,String>records=consumer.poll(Duration.ofSeconds(1));for(ConsumerRecord<String,String>record:records){System.out.println(record.value());}// 异步提交offsetconsumer.commitAsync();}

6.4 指定offset消费

auto.offset.reset参数控制当Kafka中无初始offset或当前offset不存在时的处理方式：

• earliest：从最早的offset开始消费，等同于命令行的--from-beginning
• latest（默认）：从最新的offset开始消费
• none：找不到消费者组的历史offset则抛出异常

如果需要从任意指定的offset开始消费：

// 先订阅主题，等待分区分配完成consumer.subscribe(Arrays.asList("first"));Set<TopicPartition>assignment=newHashSet<>();while(assignment.size()==0){consumer.poll(Duration.ofSeconds(1));assignment=consumer.assignment();}// 为每个分区指定从offset=1700开始消费for(TopicPartitiontp:assignment){consumer.seek(tp,1700);}

注意：每次执行完需要修改消费者组名，否则已提交的offset会覆盖seek的效果。

6.5 指定时间消费

生产环境中如果某段时间的数据出现异常，需要回溯到指定时间点重新消费：

// 获取分区分配信息Set<TopicPartition>assignment=newHashSet<>();while(assignment.size()==0){consumer.poll(Duration.ofSeconds(1));assignment=consumer.assignment();}// 构造每个分区对应的时间戳（1天前）Map<TopicPartition,Long>timestampToSearch=newHashMap<>();for(TopicPartitiontp:assignment){timestampToSearch.put(tp,System.currentTimeMillis()-1*24*3600*1000);}// 获取对应时间戳的offsetMap<TopicPartition,OffsetAndTimestamp>offsets=consumer.offsetsForTimes(timestampToSearch);// 为每个分区设置对应的offsetfor(TopicPartitiontp:assignment){OffsetAndTimestampoffsetAndTimestamp=offsets.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本身无法完全保证

理解了这几个核心权衡点，消费者的各种参数和机制就都有了落脚点。