1. 从“消费”这件事说起Consumer到底在解决什么问题很多人学KafkaProducer那关过得挺顺——发消息嘛指定个Topic把数据塞进去就完事了。到了Consumer就开始犯迷糊为什么我起了三个消费者只有一个在干活为什么重启之后消息又从头消费了一遍为什么明明消费了Lag却一直不降这些问题的根子都在于没搞清楚Consumer的设计定位。Consumer不是“被动接收消息的管道”它是一个主动拉取、自主管理位点、参与组内协调的复杂角色。你可以把它想象成一个快递站的取件员不是快递到了就自动推到你手上而是你拿着取件码主动去站点把包裹拉回来同时还要记录“我上次取到哪一件了”免得重复取或者漏取。Kafka的Consumer有几个核心特征理解了它们后面所有操作都不会迷路Pull模式而非Push模式Consumer自己决定什么时候去Broker拉数据拉多少。这跟很多消息队列的Push模式完全不同。好处是Consumer不会被压垮——我处理得慢就少拉点处理得快就多拉点主动权在我手里。位点Offset由Consumer自己维护Broker不关心你消费到哪了它只负责存消息。你消费到哪是你自己的事。这就解释了为什么“重启后重复消费”——因为你的位点没提交成功或者提交到了一个错误的位置。Consumer Group是逻辑上的“一个消费者”同一个组内的多个Consumer实例共同瓜分所有分区的消息。一个分区在同一时刻只能被组内一个Consumer消费。这就是为什么“起了三个消费者只有一个在干活”——如果你的Topic只有一个分区那确实只能有一个消费者分到活。我见过太多人把Consumer Group当成“负载均衡器”来理解结果在分区数不够的时候百思不得其解。记住一句话并行度的上限是分区数不是消费者数。2. 消费位点提交自动提交的“舒适区”与“雷区”位点提交是Consumer里最容易出问题的地方没有之一。很多线上事故——重复消费、消息丢失——追到根上都是位点提交的时机和方式没搞对。2.1 自动提交到底在什么时候提交先看一段最常见的配置enable.auto.committrue auto.commit.interval.ms5000这两行配置的意思是Consumer在后台每隔5秒自动提交一次当前已拉取消息的最大位点。注意关键词——已拉取不是已处理完。这就埋了一个雷。假设你的Consumer拉了一批消息位点100到200交给业务线程去处理。业务线程处理到第150条的时候卡住了花了10秒。而这10秒内后台的自动提交线程已经跑了两次把位点提交到了200。然后Consumer崩了。重启之后从200开始消费第151到200条消息——丢了。反过来如果自动提交间隔设得特别短比如100毫秒那又会出现另一个问题你刚拉下来还没处理完位点就提交了一旦处理失败消息同样丢失。而且频繁提交位点会给Broker带来额外压力。所以自动提交的“舒适区”是你对消息丢失有一定容忍度且处理逻辑足够快、足够稳定。比如日志采集、监控指标上报这类场景丢几条无所谓用自动提交省心省力。2.2 手动提交的两种姿势同步与异步一旦业务要求“至少一次”甚至“精确一次”就必须关掉自动提交自己控制时机enable.auto.commitfalse然后有两种提交方式同步提交consumer.commitSync()。调用后会阻塞直到Broker确认收到提交请求。优点是可靠提交失败会抛异常你能感知到缺点是吞吐量受影响每次提交都要等一个网络往返。异步提交consumer.commitAsync()。调用后立即返回不阻塞。可以传一个回调在提交成功或失败时做处理。优点是快缺点是失败时不会自动重试——因为如果异步提交失败后你重试可能提交的是一个更旧的位点反而造成重复消费。实际项目中我通常这样组合使用try { while (true) { 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(); } }这个模式的核心逻辑是日常用异步保吞吐退出用同步保可靠。因为退出时如果异步提交失败你没有机会再重试了所以必须用同步兜底。2.3 位点提交与分区再均衡的“致命交叉”比自动提交更隐蔽的坑是位点提交和分区再均衡Rebalance的交互。当有新Consumer加入组或者有Consumer离开组或者订阅的Topic分区数发生变化时就会触发Rebalance。Rebalance期间所有Consumer都会停止消费等协调者重新分配分区。问题来了如果一个Consumer在Rebalance之前拉了一批消息处理了一部分还没提交位点Rebalance发生了这个分区被分配给了另一个Consumer。新的Consumer从上次提交的位点开始消费——那些已经处理但没提交的消息会被重复消费。更严重的情况是Consumer A正在处理分区P的消息位点还没提交Rebalance把分区P分给了Consumer B。Consumer A处理完后提交位点但这个提交是针对分区P的——而分区P已经不属于A了。这个提交要么被拒绝要么造成位点混乱。解决办法是使用ConsumerRebalanceListener在分区被回收之前提交位点consumer.subscribe(Collections.singletonList(my-topic), new ConsumerRebalanceListener() { Override public void onPartitionsRevoked(CollectionTopicPartition partitions) { // 分区被回收前同步提交位点 consumer.commitSync(); } Override public void onPartitionsAssigned(CollectionTopicPartition partitions) { // 分区分配后可以从外部存储恢复位点 for (TopicPartition tp : partitions) { long offset loadOffsetFromDB(tp); if (offset 0) { consumer.seek(tp, offset); } } } });onPartitionsRevoked是在分区被拿走之前调用的这是你最后提交位点的机会。onPartitionsAssigned是在拿到新分区之后调用的你可以在这里从数据库或Redis里恢复位点实现更精确的控制。我踩过的一个坑在onPartitionsRevoked里用异步提交结果Rebalance太快提交还没完成分区就被收走了。后来改成同步提交虽然会稍微延长Rebalance时间但至少不会丢位点。3. Rebalance机制为什么你的消费者组总是“卡住”Rebalance是Kafka Consumer里最让人头疼的机制。它本身是为了实现弹性伸缩和高可用但触发条件多、过程复杂稍不注意就会导致消费停滞。3.1 三种触发条件与对应的排查方向Rebalance的触发条件有三类触发条件典型场景排查方向组成员数量变化消费者启动、崩溃、主动退出检查消费者实例是否频繁重启订阅Topic数量变化使用正则订阅新建了匹配的Topic检查订阅方式是否为正则订阅Topic分区数变化运维给Topic增加了分区检查是否有分区变更操作第一类是最常见的。消费者崩溃、网络抖动导致心跳超时、GC停顿过长导致心跳线程被挂起都会让协调者认为这个消费者“死了”从而触发Rebalance。这里有个关键参数session.timeout.ms默认45秒不同版本有差异。消费者需要在这个时间内发送心跳给协调者。如果超时协调者就认为消费者下线。还有一个参数max.poll.interval.ms默认5分钟。这是两次poll()调用的最大间隔。如果你的业务处理逻辑太慢两次poll之间超过了5分钟消费者会被认为“卡住了”主动离开组触发Rebalance。这两个参数的区别很重要session.timeout.ms管的是心跳线程max.poll.interval.ms管的是处理线程。心跳线程是后台独立运行的只要进程活着就会发心跳。但如果处理线程卡住poll间隔超时消费者会主动发起离开组的请求。实际排查经验如果日志里看到“Member ... sending LeaveGroup request due to consumer poll timeout”那就是max.poll.interval.ms超了需要优化处理逻辑或者调大这个参数。如果是“Disconnecting from node ... due to request timeout”那可能是网络问题。3.2 分区分配策略Range、RoundRobin与StickyRebalance的核心动作是“把分区重新分配给消费者”。分配策略决定了怎么分。Range策略默认按Topic维度把每个Topic的分区按范围分给消费者。假设Topic有7个分区组内有3个消费者那么消费者A分到分区0-2B分到3-5C分到6。问题在于如果组内订阅了多个Topic每个Topic都这样分会导致前面的消费者总是多分到分区负载不均。RoundRobin策略把所有订阅的Topic的所有分区放在一起轮询分配给消费者。这样负载更均衡但有一个前提——组内所有消费者订阅的Topic必须完全一致否则会出现分区分配错乱。Sticky策略推荐在Rebalance时尽量保留原有的分配方案只把需要变动的分区重新分配。比如消费者C崩溃了Sticky策略会把C原来负责的分区尽量均匀地分给A和B而不是把所有分区打散重来。这大大减少了Rebalance的开销。配置方式partition.assignment.strategyorg.apache.kafka.clients.consumer.StickyAssignorSticky策略还有一个好处它支持“ cooperative rebalance”协作再均衡。传统的Rebalance是“全部停止重新分配”而协作再均衡允许消费者在Rebalance期间继续消费那些不会被回收的分区只有需要变动的分区才暂停。这进一步减少了消费停滞的时间。3.3 如何避免“Rebalance风暴”Rebalance本身不可怕可怕的是频繁Rebalance。每次Rebalance都会导致消费暂停如果频繁发生消费Lag会飙升业务会受影响。避免Rebalance风暴的几个实操要点合理设置session.timeout.ms设太小网络稍微抖动就触发Rebalance设太大消费者真挂了要很久才能被发现。一般建议6-10秒根据网络质量调整。合理设置max.poll.interval.ms根据业务处理的最大耗时来设。如果一批消息最多处理2分钟那就设成3-5分钟留足余量。控制max.poll.records每次poll拉取的最大消息数。拉太多处理时间就长容易超时。一般设500-1000根据单条处理耗时调整。避免消费者频繁重启如果是容器化部署检查健康检查是否过于激进导致消费者被反复杀掉重启。使用静态成员Static MembershipKafka 2.3支持给消费者配置group.instance.id这样消费者重启时在session.timeout.ms内重新加入组不会触发Rebalance。group.instance.idconsumer-instance-1 session.timeout.ms30000静态成员特别适合Kubernetes环境下的消费者部署。Pod重启时只要在超时时间内回来协调者就认为还是原来的成员不触发Rebalance。4. 消费位移与Lag监控怎么知道消费者“健康不健康”Consumer跑起来容易跑得健康难。很多团队是等到业务方反馈“数据怎么没更新”才发现消费者挂了。其实通过Lag监控完全可以提前发现问题。4.1 Lag到底是什么怎么算Lag Log End Offset (LEO) - Current Offset。LEO是分区最新消息的位点Current Offset是消费者当前提交的位点。两者之差就是还没消费的消息数。Lag为0说明消费者追上了最新消息。Lag持续增长说明消费速度跟不上生产速度。Lag突然飙升可能是消费者挂了或者Rebalance了。获取Lag的方式有几种命令行工具kafka-consumer-groups.sh --bootstrap-server localhost:9092 \ --describe --group my-consumer-group输出里会显示每个分区的CURRENT-OFFSET、LOG-END-OFFSET和LAG。Java APIMapTopicPartition, OffsetAndMetadata committed consumer.committed(partitions); for (TopicPartition tp : partitions) { long leo consumer.endOffsets(Collections.singleton(tp)).get(tp); long current committed.get(tp).offset(); long lag leo - current; System.out.println(tp lag: lag); }JMX指标Kafka Consumer会暴露records-lag-max等JMX指标可以直接接入监控系统。4.2 用Lag做告警的几种策略Lag监控不是简单设个阈值就完事了。不同业务对Lag的容忍度完全不同。绝对阈值告警Lag超过某个值就告警。适合消息量稳定的场景。比如Lag超过10万就告警。增长率告警Lag的增长速度超过某个速率就告警。适合消息量波动大的场景。比如Lag在5分钟内从0涨到5万说明消费完全停滞了。消费时间告警根据Lag和消费速率估算“还需要多久才能追平”。如果估算时间超过业务容忍度就告警。我一般会组合使用绝对阈值做兜底增长率做灵敏告警。同时把Lag的历史曲线画出来方便排查是“缓慢增长”还是“突然飙升”。一个容易忽略的点Lag为0不一定代表健康。如果消费者挂了但位点已经提交到了LEOLag也是0。所以还要结合消费者的存活状态一起判断。4.3 消费停滞的常见原因排查清单当发现Lag持续增长时按这个顺序排查消费者进程是否存活检查进程、容器状态、日志有无异常退出。是否在频繁Rebalance查看消费者日志有无“Rebalance”相关记录检查session.timeout.ms和max.poll.interval.ms配置。业务处理是否变慢检查下游依赖数据库、外部接口的响应时间看是否有慢查询或超时。分区数是否足够如果消费者数已经等于分区数再加消费者也没用需要增加分区。是否有消息堆积导致GC频繁大量消息积压会导致内存压力大GC频繁进一步拖慢处理速度。5. 多线程消费提升吞吐的正确打开方式单线程Consumer的吞吐量是有上限的。当Lag持续增长加消费者又受限于分区数时就需要考虑多线程消费。5.1 为什么Kafka Consumer不是线程安全的Kafka Consumer设计上就是单线程的。它内部维护了网络连接、位点状态、分区分配等复杂状态多个线程同时操作会导致状态混乱。官方明确说了Consumer不是线程安全的多线程共享一个Consumer实例会抛ConcurrentModificationException。所以多线程消费的模式要么是“每个线程一个Consumer”要么是“一个Consumer拉取多个Worker处理”。5.2 模式一每线程一个Consumer这是最简单的多线程模式。每个线程创建自己的Consumer实例订阅相同的Topic但属于不同的Consumer Group或者同一个Group但线程数不超过分区数。int threadNum 4; for (int i 0; i threadNum; i) { new Thread(() - { KafkaConsumerString, String consumer new KafkaConsumer(props); consumer.subscribe(Collections.singletonList(my-topic)); while (true) { ConsumerRecordsString, String records consumer.poll(Duration.ofMillis(100)); for (ConsumerRecordString, String record : records) { process(record); } consumer.commitSync(); } }).start(); }这种模式的优点是简单每个Consumer独立管理位点互不干扰。缺点是Consumer实例多了网络连接和内存开销也大。而且如果线程数超过分区数多出来的线程分不到分区白白浪费。5.3 模式二一个Consumer拉取Worker池处理这种模式用一个Consumer负责拉取消息然后把消息提交给线程池处理。位点提交需要等Worker处理完再统一提交。KafkaConsumerString, String consumer new KafkaConsumer(props); consumer.subscribe(Collections.singletonList(my-topic)); ExecutorService executor Executors.newFixedThreadPool(8); while (true) { ConsumerRecordsString, String records consumer.poll(Duration.ofMillis(100)); MapTopicPartition, OffsetAndMetadata currentOffsets new HashMap(); for (ConsumerRecordString, String record : records) { executor.submit(() - { process(record); // 记录处理完成的位点 synchronized (currentOffsets) { currentOffsets.put( new TopicPartition(record.topic(), record.partition()), new OffsetAndMetadata(record.offset() 1) ); } }); } // 等待所有任务完成后再提交位点 // 实际实现需要更精细的同步控制 consumer.commitSync(currentOffsets); }这种模式的坑在于位点提交的时机。如果不等Worker处理完就提交会丢消息如果等所有Worker处理完再提交吞吐量又上不去。实际项目中通常需要引入一个“滑动窗口”机制跟踪每个分区的处理进度按分区粒度提交位点。我个人更倾向于模式一除非单线程处理确实扛不住。模式二的复杂度高位点管理容易出错而且一旦某个Worker卡住整个消费都会受影响。5.4 多线程消费的位点管理要点无论哪种模式多线程下的位点管理都要注意位点提交必须串行化多个线程同时提交位点会导致位点跳跃或回退。要么加锁要么只由一个线程负责提交。按分区粒度跟踪进度不同分区的消费进度不同不能用一个全局位点来代表所有分区。处理失败要有补偿机制Worker处理失败的消息不能简单跳过要记录到重试队列或死信队列。6. 消费端幂等重复消费不可怕重复处理才可怕Kafka只能保证“至少一次”消费语义在手动提交且处理成功后再提交的情况下。这意味着消息重复是常态不是异常。网络抖动、Rebalance、消费者重启都可能导致重复消费。所以消费端必须做幂等。幂等的核心思想是同一条消息处理多次结果和处理一次一样。6.1 用唯一ID做去重最常见的方案是给每条消息分配一个全局唯一的业务ID处理前先查这个ID是否已经处理过。// 伪代码 String msgId record.key(); // 假设key是唯一ID if (redis.setnx(consumed: msgId, 1) 1) { // 设置成功说明没处理过 process(record); redis.expire(consumed: msgId, 24 * 3600); // 设置过期时间 } else { // 已经处理过跳过 log.info(重复消息跳过: {}, msgId); }这个方案的关键是setnx的原子性以及过期时间的设置。过期时间太短可能还没过重复窗口就失效了太长Redis内存压力大。一般根据业务的重试窗口来定24小时是个比较安全的默认值。6.2 数据库唯一约束兜底如果业务操作是写数据库可以利用数据库的唯一索引来做幂等。比如订单创建用订单号做唯一索引重复插入会报唯一键冲突捕获这个异常即可。INSERT INTO orders (order_id, user_id, amount) VALUES (?, ?, ?) ON DUPLICATE KEY UPDATE update_time NOW();这种方式的好处是不依赖外部组件数据库自身就能保证。缺点是只适用于写数据库的场景而且如果业务逻辑复杂比如涉及多个表的写入单靠唯一索引不够。6.3 状态机幂等对于有状态流转的业务可以用状态机来做幂等。比如订单状态从“待支付”到“已支付”如果已经是“已支付”状态再次收到支付消息就忽略。// 伪代码 Order order orderDao.selectById(orderId); if (order.getStatus() OrderStatus.PAID) { log.info(订单已支付忽略重复消息); return; } order.setStatus(OrderStatus.PAID); orderDao.update(order);这种方式需要配合乐观锁或悲观锁防止并发状态变更。幂等方案的选择取决于业务场景。我的经验是能用数据库唯一约束就用数据库简单可靠需要跨服务去重就用Redis状态流转类业务用状态机最自然。7. 消费者配置调优那些默认值不一定适合你Kafka Consumer有一堆配置参数默认值是为通用场景设计的。实际项目中有几个参数几乎必须调整。7.1 fetch相关配置控制拉取行为fetch.min.bytes1 fetch.max.wait.ms500 max.partition.fetch.bytes1048576 max.poll.records500fetch.min.bytes一次拉取的最小字节数。默认1字节意味着只要有消息就拉。设大一点比如1MB可以减少网络往返但会增加延迟。适合吞吐优先的场景。fetch.max.wait.ms如果数据不够fetch.min.bytes最多等多久。默认500ms。和fetch.min.bytes配合使用。max.partition.fetch.bytes每个分区一次拉取的最大字节数。默认1MB。如果单条消息很大需要调大这个值否则会一直拉不到完整消息。max.poll.records一次poll返回的最大消息数。默认500。这个值直接影响处理时间进而影响max.poll.interval.ms。7.2 心跳与超时配置平衡灵敏度和稳定性session.timeout.ms10000 heartbeat.interval.ms3000 max.poll.interval.ms300000session.timeout.ms协调者等待消费者心跳的超时时间。默认45秒新版本或10秒老版本。建议设6-10秒根据网络质量调整。heartbeat.interval.ms消费者发送心跳的间隔。一般是session.timeout.ms的三分之一。设太小会增加网络开销设太大会导致心跳超时被误判。max.poll.interval.ms两次poll的最大间隔。默认5分钟。根据业务处理的最大耗时来设留2-3倍余量。7.3 不同场景的配置模板低延迟场景如实时监控fetch.min.bytes1 fetch.max.wait.ms100 max.poll.records100 session.timeout.ms6000 heartbeat.interval.ms2000高吞吐场景如日志采集fetch.min.bytes1048576 fetch.max.wait.ms1000 max.poll.records1000 max.partition.fetch.bytes5242880 session.timeout.ms30000 heartbeat.interval.ms10000大消息场景如文件传输max.partition.fetch.bytes10485760 fetch.max.bytes52428800 max.poll.records10调参这件事没有银弹。我的建议是先用默认值跑起来观察Lag和消费速率然后针对瓶颈调整。每次只调一个参数观察效果避免多个参数一起调导致无法定位问题。8. 从Consumer视角看消息丢失与重复的完整链路最后把消息丢失和重复的链路完整梳理一遍。理解了这条链路排查问题就有章可循。消息丢失的可能环节Producer端acks0或acks1时Broker还没落盘就确认Broker挂了消息丢失。Broker端副本数不够Leader挂了未同步的副本被选为新Leader消息丢失。Consumer端自动提交位点消息还没处理完位点就提交了处理失败后消息丢失。消息重复的可能环节Producer端重试导致同一条消息发送多次。Consumer端位点提交失败后重启从旧位点重新消费。Consumer端Rebalance时位点未及时提交新Consumer从旧位点消费。Consumer端手动提交时处理成功但提交失败下次重新消费。精确一次Exactly Once的实现思路Kafka 0.11支持事务和幂等Producer可以实现端到端的精确一次。核心思路是Producer开启幂等enable.idempotencetrue保证单分区内不重复。Producer开启事务transactional.id保证跨分区写入的原子性。Consumer设置isolation.levelread_committed只读已提交的事务消息。消费端将位点提交和业务处理放在同一个事务里比如数据库事务保证“处理成功”和“位点提交”要么都成功要么都失败。# Producer enable.idempotencetrue transactional.idmy-tx-id # Consumer isolation.levelread_committed enable.auto.commitfalse精确一次的实现成本很高对性能有影响。大多数业务场景“至少一次消费端幂等”已经足够。只有在金融、计费等对准确性要求极高的场景才值得上精确一次。我在实际项目中90%的场景都是用“手动提交幂等”的方案。简单、可靠、性能好。精确一次只在少数核心链路用过配置复杂排查问题也麻烦。Consumer这块的内容展开讲能讲一本书。上面这些是我在实际项目中踩过坑、填过坑之后觉得最值得分享的部分。每个参数、每个配置背后都有它的设计意图理解了意图才能根据业务场景做出正确的取舍。