1. 从“消费”这件事说起Consumer到底在解决什么问题很多人第一次接触Kafka注意力几乎都被Producer吸引走了——毕竟“发消息”这件事听起来更主动、更有掌控感。但真正在生产环境里把Kafka用出问题的十有八九栽在Consumer这一侧。我见过太多团队Producer写得漂漂亮亮分区策略、幂等、压缩都调得明明白白结果上线没两天就出现消息重复消费、消费积压、消费者组莫名其妙rebalance最后排查半天发现是Consumer的配置和位移提交逻辑没搞对。所以这篇内容我想把Kafka Consumer这块从头到尾捋一遍。它是什么简单说Consumer就是从Kafka集群里把消息拉出来处理的那一端。它能做什么能按消费者组并行消费、能控制位移提交时机、能精确控制从哪开始读、能在多线程下保证分区内的顺序。它解决了什么问题解决了“海量消息如何被多个消费实例高效、有序、可恢复地处理”这个核心问题。适合谁看适合已经会写Producer、但对Consumer的组协调、位移管理、rebalance机制还停留在“能跑就行”阶段的开发者也适合那些正在被消费积压和重复消费折磨的运维同学。我个人的经验是Consumer这块的知识密度比Producer高不少因为它牵扯到分布式协调、状态管理、线程模型这几件麻烦事。下面我按自己的理解顺序把核心思路、关键细节、实操过程和踩坑记录都摊开讲。2. Consumer整体设计与核心思路拆解2.1 为什么Consumer是“拉”而不是“推”Kafka Consumer最根本的一个设计决策就是它采用拉取poll模型而不是由Broker主动推送。这个选择背后有很实在的理由。如果是Broker推消息给Consumer那Broker就得知道每个Consumer的处理能力。有的Consumer处理快有的处理慢Broker一旦按统一速率推慢的那端就会被压垮。而拉模型把节奏控制权交给了Consumer自己——你处理得过来就多拉点处理不过来就少拉点甚至暂停拉取。这就像自助餐厅你自己去取餐吃多少拿多少而不是服务员不停往你盘子里堆。另一个原因是批量。拉模型天然适合攒一批消息一起返回Consumer可以设置一次poll最多拿多少条、最多等多久、最少要凑够多少字节。这几个参数直接决定了吞吐和延迟的平衡后面会细讲。提示拉模型的代价是如果Broker没有新消息Consumer会空转轮询。Kafka用fetch.min.bytes和fetch.max.wait.ms来缓解这个问题让Broker在没有足够数据时先hold住请求等攒够或超时再返回。2.2 消费者组并行的基本单位单个Consumer的吞吐是有上限的所以Kafka引入了**消费者组Consumer Group**的概念。同一个组内的多个Consumer实例共同消费一个或多个Topic每个分区只会被组内的一个Consumer消费。这就实现了并行的水平扩展。这里有个关键约束必须记住组内Consumer数量超过分区数时多出来的Consumer会处于空闲状态。因为一个分区只能分配给一个Consumer这是保证分区内顺序消费的前提。我见过有人为了“提高消费速度”盲目加Consumer实例结果加到分区数的两倍一半实例干坐着还以为是配置问题。所以正确的扩容思路是先看Topic有多少分区Consumer实例数最多加到分区数。如果还不够那就得考虑增加分区数但增加分区会改变消息的路由这个要提前规划。2.3 位移Offset管理的本质Consumer消费到哪了靠的是位移来记录。每个分区都有一个当前消费位移表示下一条要读的消息的位置。这个位移存在哪老版本存在ZooKeeper里新版本存在Kafka内部的一个叫__consumer_offsets的Topic里。位移管理最核心的问题是什么时候提交位移。如果你先处理消息再提交位移处理到一半挂了重启后会从上次提交的位置重新读消息会重复消费。如果你先提交位移再处理消息提交完挂了这条消息就丢了。这就是经典的**至少一次at-least-once和至多一次at-most-once**的取舍。绝大多数业务场景选择至少一次也就是先处理再提交配合业务侧的幂等来去重。这个取舍没有标准答案取决于你的业务能不能容忍重复、能不能容忍丢失。2.4 Rebalance绕不开的“暂停键”当组内成员变化新Consumer加入、老Consumer离开、订阅的Topic分区数变化时Kafka会触发Rebalance重新分配分区归属。Rebalance期间整个组会短暂停止消费这就是很多人遇到的“消费突然卡顿几秒”的元凶。Rebalance的触发条件、分配策略、协调过程是Consumer这块最复杂也最容易出问题的部分。后面我会单独用一节来讲怎么减少不必要的Rebalance以及不同分配策略的差异。3. 核心细节解析与实操要点3.1 关键配置参数逐个拆解Consumer的配置项很多但真正影响行为和稳定性的就那么几个。我把它们分成三类拉取控制类、位移提交类、会话与心跳类。拉取控制类参数作用常见取值说明fetch.min.bytes一次poll最少返回多少字节1默认调大能提升吞吐但增加延迟fetch.max.wait.ms没凑够min.bytes时最多等多久500和上面配合使用max.partition.fetch.bytes每个分区一次最多返回多少字节1MB太大占内存太小吞吐低max.poll.records一次poll最多返回多少条500直接决定单次处理量max.poll.records这个参数我要特别强调。它决定了你一次poll拿到多少条消息也就决定了你两次poll之间的处理时间。如果这个值设得太大处理一批要很久就可能超过max.poll.interval.ms导致Consumer被踢出组触发Rebalance。我一般建议从500开始根据单条处理耗时反推。位移提交类enable.auto.commit是否自动提交默认true。生产环境我强烈建议设为false手动控制提交时机。auto.commit.interval.ms自动提交间隔默认5秒。只在自动提交开启时有效。auto.offset.reset没有初始位移或位移越界时怎么办。可选earliest、latest、none。这个参数坑过无数人后面单独讲。会话与心跳类session.timeout.ms会话超时时间默认10秒新版本45秒。超过这个时间没心跳Consumer被判定死亡。heartbeat.interval.ms心跳间隔一般是session timeout的三分之一。max.poll.interval.ms两次poll的最大间隔默认5分钟。超过这个时间没pollConsumer主动离开组。这里有个容易混淆的点session.timeout.ms管的是心跳线程max.poll.interval.ms管的是处理线程。心跳是后台线程自动发的只要进程活着就会发但如果你处理消息太慢超过了max.poll.interval即使心跳正常也会被踢出去。这两个超时机制是独立的排查问题时一定要分清。3.2 auto.offset.reset的正确理解这个参数的名字很有迷惑性很多人以为它是“自动重置位移”的意思其实它只在两种情况下生效一是Consumer组第一次消费某个分区找不到已提交的位移二是当前提交的位移在Broker上已经不存在了比如数据被清理了。它的三个取值earliest从最早的消息开始读。适合你想全量处理历史数据的场景。latest从最新的消息开始读。适合只关心新消息的场景也是默认值。none找不到位移就抛异常。适合你想明确知道位移丢失的场景。我踩过的坑是新上线一个Consumer组忘了设auto.offset.reset默认是latest结果历史数据一条没消费还以为是代码问题。所以新建Consumer组时一定要明确这个参数。注意这个参数只在“找不到位移”时生效一旦有了提交的位移它就再也不起作用了。想重新消费历史数据得用seek方法手动指定位移或者换一个新的group.id。3.3 位移提交的三种姿势手动提交位移有三种方式各有适用场景。同步提交consumer.commitSync()。会阻塞直到提交成功或抛异常。最安全但吞吐最低。适合对数据准确性要求极高的场景。异步提交consumer.commitAsync()。不阻塞提交结果通过回调通知。吞吐高但失败时不会自动重试可能丢位移。适合能容忍少量重复的场景。混合提交正常消费时用异步提交保证吞吐关闭Consumer前用同步提交保证最后一次位移落盘。这是我最推荐的模式兼顾性能和可靠性。try { while (running) { ConsumerRecordsString, String records consumer.poll(Duration.ofMillis(100)); for (ConsumerRecordString, String record : records) { process(record); } consumer.commitAsync(); } } catch (Exception e) { log.error(消费异常, e); } finally { try { consumer.commitSync(); } finally { consumer.close(); } }这段代码的关键在于循环内异步提交退出时同步提交兜底。异步提交失败的回调里可以记录日志但不建议在里面重试因为重试的位移可能已经过期了。3.4 分区分配策略的选择Kafka提供了几种分区分配策略通过partition.assignment.strategy配置。RangeAssignor按分区范围分配。比如两个Consumer订阅同一个TopicTopic有4个分区那Consumer1拿0、1Consumer2拿2、3。问题是如果多个Topic可能导致负载不均。RoundRobinAssignor轮询分配。把所有Topic的所有分区排序后轮流分配负载更均衡。但每次Rebalance时分配结果变化大。StickyAssignor粘性分配。尽量保持上次的分配结果只移动必要的分区。这能减少Rebalance后的状态重建成本是我现在最推荐的策略。CooperativeStickyAssignor协作粘性分配。配合增量Rebalance使用避免全局暂停。新版本Kafka的默认推荐。选择策略的核心考量是Rebalance时能不能少动分区、能不能避免全局暂停。Sticky系列在这两点上表现最好。4. 实操过程与核心环节实现4.1 一个完整的Consumer实现骨架下面这个骨架是我在多个项目里沉淀下来的包含了位移管理、优雅关闭、异常处理这几个关键环节。public class ReliableConsumer { private final KafkaConsumerString, String consumer; private volatile boolean running true; private final CountDownLatch shutdownLatch new CountDownLatch(1); public ReliableConsumer(String groupId, String topic) { Properties props new Properties(); props.put(bootstrap.servers, localhost:9092); props.put(group.id, groupId); props.put(key.deserializer, org.apache.kafka.common.serialization.StringDeserializer); props.put(value.deserializer, org.apache.kafka.common.serialization.StringDeserializer); props.put(enable.auto.commit, false); props.put(auto.offset.reset, earliest); props.put(max.poll.records, 500); props.put(max.poll.interval.ms, 300000); props.put(session.timeout.ms, 45000); props.put(partition.assignment.strategy, org.apache.kafka.clients.consumer.CooperativeStickyAssignor); this.consumer new KafkaConsumer(props); consumer.subscribe(Collections.singletonList(topic)); } public void run() { try { while (running) { ConsumerRecordsString, String records consumer.poll(Duration.ofMillis(200)); if (records.isEmpty()) continue; for (ConsumerRecordString, String record : records) { handle(record); } consumer.commitAsync((offsets, exception) - { if (exception ! null) { log.warn(异步提交失败, exception); } }); } } catch (WakeupException e) { if (running) throw e; } finally { try { consumer.commitSync(); } finally { consumer.close(); shutdownLatch.countDown(); } } } public void shutdown() throws InterruptedException { running false; consumer.wakeup(); shutdownLatch.await(); } private void handle(ConsumerRecordString, String record) { // 业务处理逻辑 } }这个骨架有几个设计点值得说明。第一用wakeup()来中断poll这是Kafka官方推荐的优雅关闭方式比直接close更安全。第二用CountDownLatch等待关闭完成避免主线程提前退出。第三异步提交加同步兜底兼顾吞吐和可靠性。4.2 多线程消费的两种模型单线程Consumer的吞吐有限多线程是必然选择。但KafkaConsumer本身不是线程安全的所以多线程消费有两种经典模型。模型一每个线程一个Consumer实例。每个线程独立创建Consumer、独立poll、独立提交位移。优点是简单、线程安全、分区内顺序天然保证。缺点是Consumer实例多内存和连接开销大Rebalance时每个实例都要参与。模型二一个Consumer拉取多个Worker处理。一个线程负责poll把消息分发给工作线程池处理处理完再统一提交位移。优点是Consumer实例少缺点是位移提交复杂需要等所有Worker处理完才能提交而且分区内顺序可能被打乱。我个人的选择是如果分区数不多比如几十个用模型一简单可靠。如果分区数上百Consumer实例开销太大才考虑模型二但模型二必须配合分区级别的顺序控制否则很容易出问题。4.3 位移提交的时机与幂等配合前面说了至少一次语义也就是先处理再提交。但这里有个细节提交的位移应该是“已处理完的消息的下一条”。Kafka的位移语义是“下一条要读的位置”所以如果你处理完了位移为100的消息应该提交101。在批量处理时如果一批消息处理到一半失败了你不能提交整批的位移否则失败的那条就丢了。正确做法是记录最后一条成功处理的位移只提交到那里。这也是为什么我建议在批量处理时逐条记录处理进度。配合幂等的话业务侧需要有一个唯一键比如消息里的订单号、事件ID。处理前先查这个键有没有处理过处理过就跳过。这样即使重复消费也不会产生副作用。提示幂等表最好和业务数据在同一个事务里保证“处理业务”和“记录幂等”原子性。如果做不到至少保证幂等记录的写入在处理之后、提交位移之前。4.4 消费积压的排查与处理消费积压是Consumer最常见的线上问题。排查思路我总结成三步。第一步确认积压量。用kafka-consumer-groups.sh脚本查看LAG列或者用JMX指标records-lag-max。LAG就是“最新消息位移 - 当前消费位移”代表还有多少条没消费。第二步定位瓶颈。积压无非三个原因消费速度慢、生产速度突然暴涨、Consumer挂了或卡住。先看Consumer进程是否存活再看单条处理耗时是否变长最后看生产速率是否异常。第三步对症处理。如果是消费慢先看是不是下游依赖数据库、外部接口变慢优化处理逻辑如果是生产暴涨临时加Consumer实例不超过分区数如果是Consumer卡住看是不是触发了Rebalance或者poll间隔超时。我遇到过一次积压排查半天发现是某个Consumer实例的max.poll.interval.ms设得太小处理一批消息偶尔超时导致频繁Rebalance整个组一直在重新分配分区根本没法正常消费。把参数调大后立刻恢复。5. 常见问题与排查技巧实录5.1 Rebalance频繁触发的排查清单Rebalance频繁是Consumer最烦人的问题之一。下面这张表是我整理的排查清单按可能性排序。现象可能原因排查方法解决方向消费频繁暂停几秒心跳超时看日志有无Rebalance记录调大session.timeout.ms处理一批后掉线poll间隔超时看max.poll.interval配置调大该值或减小max.poll.records新实例加入就Rebalance正常行为确认是否预期扩容用CooperativeSticky减少影响无规律RebalanceGC停顿过长看GC日志优化GC或调大超时一直Rebalance不停止处理逻辑死循环看线程栈修复死循环这里我要特别说GC这个问题。如果Consumer进程发生长时间Full GC心跳线程也会被暂停导致Broker认为Consumer死了触发Rebalance。这种情况调大session.timeout.ms能缓解但根本解法还是优化内存和GC。5.2 重复消费的根因分析重复消费的原因就那么几种但排查起来容易绕弯。原因一位移提交失败但消息已处理。异步提交失败没重试下次poll从旧位移开始消息重复。解法是异步提交加同步兜底或者业务侧幂等。原因二Rebalance导致位移回退。Rebalance后新分配的Consumer从上次提交的位移开始读如果上次提交的位移落后于实际处理进度就会重复。解法是Rebalance前主动提交位移用ConsumerRebalanceListener的onPartitionsRevoked回调。原因三手动seek导致。有人为了重放数据手动seek到旧位移这当然会重复。这是预期行为配合幂等即可。原因四auto.offset.reset设成earliest且位移丢失。位移过期被清理后从最早开始读大量重复。解法是监控位移过期或者用none及时报警。5.3 消费顺序性的保证与取舍Kafka只保证分区内有序不保证Topic级别有序。所以如果你的业务要求全局有序那只能用一个分区但这样就失去了并行能力。实际业务中大部分“有序”需求其实是“同一个业务实体的消息有序”。比如同一个订单的创建、支付、发货消息要按顺序处理。这种场景的解法是用订单号做keyKafka会按key哈希到同一个分区这样同一订单的消息天然有序。但要注意如果某个key的消息特别多会导致分区倾斜某个分区特别热。这时候可以考虑给key加后缀打散但打散后就失去顺序性了。这是一个需要权衡的点。5.4 优雅关闭的正确姿势Consumer关闭时如果处理不当会导致位移丢失或重复。正确的关闭流程是设置running标志为false调用consumer.wakeup()中断当前poll在poll的异常处理里捕获WakeupException在finally里同步提交位移调用consumer.close()等待关闭完成这套流程能保证正在处理的消息处理完、位移提交成功、Consumer正常离开组而不是被判定超时踢出。正常离开组会触发一次Rebalance但比超时踢出要快得多对组内其他成员影响也小。注意不要在onPartitionsRevoked回调里做耗时操作这个回调是同步的会阻塞Rebalance流程。提交位移可以但别做数据库操作。5.5 监控指标与告警设置Consumer的监控我重点关注这几个指标records-lag-max最大积压量超过阈值告警records-consumed-rate消费速率突然下降要关注fetch-latency-avg拉取延迟反映Broker响应速度commit-latency-avg提交延迟反映位移提交健康度rebalance-rate-per-hourRebalance频率频繁触发要排查这些指标通过JMX暴露可以用Prometheus的JMX Exporter采集配Grafana看板。我一般把积压告警设成“持续5分钟LAG大于1万”Rebalance告警设成“每小时超过3次”。6. 一些实战中的个人体会Consumer这块我踩的坑比Producer多得多因为它的状态更多、协调更复杂。有几个体会是文档里不会写的。第一新建Consumer组时一定要显式设置auto.offset.reset。默认的latest会让你的新组从最新消息开始读如果你期望的是全量处理那就等着数据缺失吧。我现在的习惯是任何新组上线前先确认这个参数。第二max.poll.records和max.poll.interval.ms要一起调。单条处理耗时乘以records数必须小于interval否则必然Rebalance。我一般留一倍余量比如单条处理50ms500条就是25秒interval设成60秒。第三Rebalance监听器里提交位移是保命操作。onPartitionsRevoked回调里同步提交一次位移能大幅减少Rebalance后的重复消费。这个操作成本很低但收益很高。第四多线程消费优先选“每线程一个Consumer”。虽然实例多但逻辑简单、不易出错。模型二那种共享Consumer的方案位移提交和顺序控制太容易出bug除非有明确的性能瓶颈否则不值得。第五消费积压时先别急着加实例。先看是不是下游变慢、是不是有Rebalance、是不是单条处理逻辑有问题。盲目加实例可能触发更多Rebalance反而更慢。最后分享一个排查小技巧如果怀疑Consumer卡住了用jstack看线程栈看poll线程是不是卡在某个处理逻辑上。如果是基本就是处理逻辑的问题跟Kafka本身无关。这个技巧帮我定位过好几次“假死”问题。
