写在前面先解决前面留下的问题如果生产者已经发送了大量消息但在最后提交之前突然宕机事务协调器会如何处理这个未完成的事务呢答案是自动终止事务协调器如果在一段时间内没有收到生产者的任何消息或者提交事务的请求会利用__transaction_state中记录的信息向所有涉及到的分区发送 Abort 命令写入的消息也会被标记为废弃。这个时长是由transaction.timeout.ms参数控制的默认是 1 分钟值为 60000。什么是消费者组回答完遗留的问题我们进入今天的正题第一个问题是什么是消费者组Consumer Group这是 Kafka 中比较有亮点的设计。简单来说Consumer Group 就是 Kafka 提供的可扩展且具有容错性的消费机制。消费者组内部包含多个消费者实例它们共享一个 Group ID。组内的所有消费者实例共同订阅一个或多个 Topic 的所有分区。每个分区只能由同一个消费者组的一个消费者实例来消费。核心特性消费者组有以下特性负载均衡Kafka 会把订阅的分区均衡的分配给组内的所有消费者这样就可以实现并行处理了。理想情况下消费者数量应该等于消费组订阅主题的分区总数。点对点模式还记得之前我们讲过 Kafka 既支持点对点模型又支持发布/订阅模型吗实际就是通过 Consumer Group 来支持的。如果所有消费者都在同一个组内每条消息都只会由其中一个消费者处理类似于传统的消息队列。这就是点对点模型的实现。发布/订阅模型如果多个不同的消费者组订阅同一个 Topic那么每个组都会收到一份完整的消息互相之间不会干扰。这种就是 发布/订阅模型。故障转移如果消费组内某个实例宕机了Kafka 会自动检测到并触发重平衡。将其负责的分区分配给正常的实例以此保证数据不丢失。消费者组重平衡流程触发条件聊完消费者组的定义和特性之后我们再来看一下它最核心也是最让人头疼的机制——重平衡。重平衡本质上是一种自动调度的机制它确保 Topic 的所有分区都有对应的消费者来消费并且分配的相对均衡。触发重平衡的条件有三个组员数量发生变动有新消费者加入组或者某个实例崩溃。订阅主题数变动消费者组可以使用正则表达式订阅主题当创建了新的满足条件的主题时就会触发。订阅主题的分区数发生变化订阅的主题扩展分区。消费者组的状态Kafka 为消费者组定义了 9 种状态它们的含义如下状态含义UNKNOWN未知状态通常是客户端与 Broker 的版本不兼容或者发生了无法识别的异常。PREPARING_REBALANCE重平衡准备中协调器收到了加入申请或者心跳超时准备重新分配。COMPLETING_REBALANCE重平衡同步中所有成员都已经加入组Leader 正在计算协调方案。STABLE稳定状态重平衡完成。DEAD消费者组注销元信息在协调者端已被移除。EMPTY组内没有任何成员通常是刚创建或者所有的消费者都正常关闭元信息依然保留在协调者端。ASSIGNING分配中这是 2.4 版本新引入的Leader 计算好了一部分新的分配计划正准备下发。RECONCILING协调中/一致性同步中也是 2.4 版本新引入的成员正在释放不再属于自己的分区准备接手新的分区。NOT_READY未就绪通常是组刚刚启动或者协调者端在进行迁移。状态流转流程如下图接着我们来看一下最经典的重平衡过程。正常情况下消费者组的状态是 STABLE也可能是 Empty协调者与消费者组中的组员之间维护有心跳消息。当有一个新的成员要加入时会给协调者发送一个 JoinGroup 请求。协调者收到请求后会通过心跳消息通知其他消费者。所有消费者此时会停止消费重新向协调者发送 JoinGroup 请求消费者组进入 PREPARING_REBALANCE 状态。当所有成员都到齐之后协调者会从中选出一个作为 Leader然后把所有的消费者信息通过 JoinGroup 的响应发送给 Leader。此时消费者组为 COMPLETING_REBALANCE 状态。Leader 计算好分配方案后会通过 SyncGroup 请求将其发送给协调者。其他的成员也会发送 SyncGroup 请求这是为了方便协调者将分配方案包装进 SyncGroup 的响应中返回给所有的消费者。消费者拿到新的任务之后就开始继续工作了。消费者组的状态恢复成 STABLE。这就是一次完整的重平衡流程这里有一个问题是在这整个过程中消费者都是不处理消息的也就是我们常说 Stop-The-World 问题。如果你有几百个 Consumer 实例那么一次 Rebalance 可能需要几个小时这简直令人崩溃。基于这种问题Kafka 在 2.4 版本推出了增量协作重平衡机制。这种机制下重平衡的过程不再要求所有的消费者都放下手中的工作而是只处理那些需要变动的分区这样就极大的提升了稳定性。只需要在大于 2.4 版本的客户端中将partition.assignment.strategy设置为CooperativeStickyAssignor即可。新版本重平衡流程如下协调者检测到变动时开启第一轮协商此时状态由 STABLE 变为 PREPARING_REBALANCE。所有成员到齐后协调者把消费者的信息发送给 Leader消费者组状态变为 COMPLETING_REBALANCE。Leader 算出哪些分区需要释放并通知给相关消费者此时消费者组状态变为 RECONCILING。释放完成后消费者组状态回到 STABLE此时存在一些游离分区需要认领。协调器接着开启新一轮的协商通过相同的报道步骤状态从 STABLE 变为 PREPARING_REBALANCE 再变为 COMPLETING_REBALANCE。这一阶段主要目的是为游离分区进行重新分配此时状态变成 ASSIGNING。当所有游离分区都有消费者认领后再次回到稳定的 STABLE 状态。Kafka 通过多次小的调整来避免整个集群长时间停止工作以此来减少重平衡对于整体集群的影响。这一进化是不是有点像 JVM 的 GC 从传统垃圾回收器进化到 G1 和 ZGC。如何避免重平衡虽然新版本的重平衡机制有了很大的进步但还是会对系统性能造成一定的影响。那如何才能避免重平衡呢首先完全消除重平衡是不可能的。我们要做的就是消除掉非预期的重平衡。什么是非预期的呢你可以理解为是由于配置不当或者系统抖动引起的重平衡。我们分别从参数调优、代码健壮性和架构设计三个层面来看一下如何调整。参数调优首先是参数调优非预期重平衡触发最常见的两个原因一个是心跳超时另一个是逻辑处理超时。为了避免因为网络抖动导致误判心跳超时我们可以适当调大session.timeout.ms这个参数决定了 Consumer 存活性的时间间隔除了这个参数还需要调整heartbeat.interval.ms这个是用来控制发送心跳消息的频率的。发送的越频繁协调者越能更快响应 Consumer 掉线并开启重平衡但随之而来的问题是消耗的资源也越多。通常可以把它设置为session.timeout.ms的三分之一。逻辑处理超时的参数主要是max.poll.interval.ms它用来控制两次 poll 之间的间隔如果你的业务逻辑复杂需要处理时间比较长那么就需要调大这个参数。例如你在业务代码中访问了第三方存储整个过程需要 5 分钟那么这个参数可以设置为 6 分钟。除了调大max.poll.interval.ms之外我们也可以调整max.poll.records它是用来控制每次 poll 的消息条数通过减少消息条数能够缩短 poll 一次的逻辑处理时间。代码健壮性介绍完了参数调优之后我们再来看一下代码层面有哪些需要调整或者注意的地方。首先是最基本的 try-catch 防护我们应该确保所有的异常都在消费逻辑中处理掉一旦消费者因为没捕获异常而崩溃那么必然会触发重平衡。其次就是优雅关闭 Consumer在停止时手动调用consumer.close()这样会给协调者发送 LeaveGroup 请求协调者收到请求后可以立即开启重平衡缩短“空窗期”。架构设计在架构层面除了使用我们前面提到的增量协作重平衡协议之外还可以设置group.instance.id这是为每个消费者实例设置一个固定的 ID这样在实例重启时只要在session.timeout.ms时间内回来协调者都会认出它不会触发重平衡。总结