手写消息队列这个系列写到第六篇终于轮到消费者心跳检测heartbeat这个机制。前面几篇我们解决了网络通信、消息存储、生产端写入和消费端拉取消息流已经能跑通但分布式环境下消费端进程随时可能崩溃或者卡死Broker必须有一种手段感知消费者是否还活着。本文就完整记录一下我在这套手写MQ里实现心跳检测的过程从设计取舍到落地代码再到踩过的坑一并整理出来。心跳检测是很多中间件的基础设施市面上主流的消息队列产品应用层都有类似的会话保活机制。别看它功能听起来简单真要从零实现牵扯到超时策略、线程模型、会话状态、位点提交和重平衡细节相当多。这篇文章适合两类人看一类是正在学习消息队列原理、想自己动手撸一个MQ的同学另一类是工作中要维护自研中间件或者研究开源MQ源码的工程师。我会把实现思路、代码骨架、参数选择原因、常见坑全部讲清楚尽量给你可以直接照抄的方案。1. 为什么消费者心跳检测是刚需1.1 消费者失效是分布式系统的常态单机进程里一个服务挂了通常有守护进程可以拉起但跨进程、跨节点之后Broker是没办法直接拿到消费者进程存活状态的。消费者可能因为内存溢出被系统杀掉也可能因为物理机宕机而失联甚至可能只是代码里某个死循环把CPU占满消费线程阻塞完全没法继续干活。我们假设一个场景消费组里有3个消费者其中一个突然停机。Broker如果不做任何检测它持有的那部分队列就再也无人消费消息越堆越多积压告警最终爆发。更麻烦的是另一种情况消费者进程没死TCP连接也还开着但消费线程阻塞住了它既不拉消息也不提交位点等于整个消费能力消失。传输层连接检查根本发现不了这类故障只有应用层的心跳才管得住。心跳检测正是用来解决“连接通着但没在干活”和“连接也断了”这两类问题的统一手段。它让Broker以较低的成本建立一条会话存活的证据链只要消费者还在周期上报我就认为它可以继续持有队列一旦链条断裂我就回收它的资源。这也是为什么几乎所有消息队列和分布式协调组件都离不开心跳。1.2 心跳检测要解决什么设计之前先把目标拆清楚。心跳至少要解决三个问题第一存活检测。Broker通过接收心跳判断消费者是否还活着超过一定时间没有上报就判定失效。这个“一定时间”不是无限大必须有界否则失效消费者会一直占着坑。第二会话保活。消费者通过周期性上报向Broker续约自己的会话维持消费资格。这就像租房续约租客每隔一段时间交一次房租房东看到续约就知道租客还在到期不续约房东就可以把房子收回来分给下一个租客。心跳就是那份续约协议会话就是租赁合同。第三触发重平衡。一旦消费者被判定失效消费组必须做出响应把失效消费者名下的队列重新分配出去。这个过程在消息队列领域叫重平衡它直接决定了系统的自愈速度。心跳检测做得越快、越准重平衡就能越快执行消息积压时间就越短。1.3 心跳在MQ整体架构中的位置在前面几篇的架构里我们已经有了Broker和消费者注册机制。客户端启动时向Broker注册Broker维护一张消费者注册表注册表里记录消费者ID、消费组、订阅的队列当时还预留了一个字段lastHeartbeat就等着这次心跳功能来填。后续所有依赖消费者在线的能力比如位点提交、队列分配都基于这张注册表是否还包含该消费者。所以心跳不是孤立功能它是Broker会话管理的基础设施。我们把心跳挂在注册表上而不是单独另起一套存储是因为没必要维护两份相互对照的数据会话状态和消费分配表天然应该放在同一个模型里。2. 心跳机制设计先把模型想清楚2.1 心跳间隔与超时时间怎么定我一开始把心跳间隔 heartbeatIntervalMs 默认设成3秒超时时间 sessionTimeoutMs 设成10秒。这个组合意味着一个消费者如果连续三个心跳周期没有上报基本可以认定失效。为什么不是1秒太频繁会让Broker的压力和网络开销上升而且心跳本身不携带大量数据在高并发场景下每多一秒都是成本。为什么不是30秒失效发现太慢消息会持续积压用户体验很差。行业里常见的经验是把超时时间设为心跳间隔的3倍左右这样给网络抖动留一两个周期的缓冲不会因为一次丢包就误杀消费者。实际落地时还需要考虑消费者端的GC停顿和瞬时负载。我最终没有用固定倍数而是做成可配置项并给出三套推荐参数按场景选心跳间隔超时时间适用场景1秒3秒对实时性要求高消费者负载低3秒10秒通用默认兼顾故障发现速度和稳定性5秒30秒消费者负载高GC停顿频繁需要更高容忍度这个表格不是一个死标准但它能帮新手快速理解调参方向你想更快发现故障就把两个值都往下压你想降低误杀率和网络开销就往上抬。生产环境建议先压测再根据实际的心跳送达率调整。2.2 会话与注册表的数据模型心跳不是一个独立存储它需要和已有注册表配合。我设计的会话模型大致如下public class ConsumerSession { private final String sessionId; // 每次注册生成UUID区分同一个consumerId的不同年代 private final String groupId; private final String consumerId; private final String channelId; // 与Broker之间的网络连接标识 private final AtomicLong lastHeartbeatMs new AtomicLong(System.currentTimeMillis()); private final long registeredAtMs; private volatile boolean expired; }这里 lastHeartbeatMs 用 AtomicLong 而不是普通 long 是吃过亏以后换的。因为心跳接收线程、扫描线程、客户端恢复线程都会读这个值普通字段在多线程下可能出现可见性问题即使加了volatile复合操作也不是原子的多个线程同时刷新旧值会覆盖新值最后看到的时间戳比实际时间早。AtomicLong 配合 set 或者 accumulateAndGet 可以彻底避开这个坑。服务端用一个 ConcurrentHashMapString, ConsumerSession 保存所有会话key就是「groupId # consumerId」。为什么不直接用HashMap因为心跳接收线程、扫描线程、拉取请求线程都会并发读写这张表HashMap在并发赋值时有概率出现不可预期的结果。ConcurrentHashMap 的 putIfAbsent、compute 这些原子方法省了我们很多加锁的心。2.3 为什么不用TCP KeepAlive很多新手会问TCP本身不是有KeepAlive吗做MQ心跳是不是多此一举这里要澄清一个关键点系统默认的TCP KeepAlive周期非常长通常以小时为单位而且它只能发现连接层级的问题比如物理链路断开、对端主机关机发现不了应用层卡死、消费线程阻塞这种状态。我们这里的消息队列心跳是应用层主动上报的语义心跳既能确认连接可用又能确认消费者业务逻辑还在运行还能携带元数据比如当前看到的组版本号。TCP KeepAlive是操作系统维护的协议栈能力对应用层不可见被负载高或者内核延迟影响时应用根本无从定位。手写MQ既然要可控必须自己做应用层心跳。3. 客户端实现心跳发送器3.1 心跳命令协议网络层沿用系列里已经实现的命令模型我给协议增加了一个新的操作码 HEARTBEAT。请求体和响应体设计如下HeartbeatRequest { sessionId, opCodeHEARTBEAT, groupId, consumerId, groupVersion, timestamp } HeartbeatResponse { opCodeHEARTBEAT_RESP, status, groupVersion, errorMessage }第一个版本没带 sessionId后来发现一个消费者可能在同一进程里属于多个消费组组和组之间必须隔离。更重要的场景是消费者网络抖动恢复后重新注册它应该是一个全新的会话旧会话如果还能发心跳就会和新会话打架。sessionId 就是为了区分“同一台机器同一个消费组里的不同代际会话”Broker端根据 sessionId 判断心跳来自旧会话就不再刷新新会话的时间。编解码器复用之前实现的那套二进制协议头部固定长度后面跟变长字段。这次只是新增了操作码和几个字段没有改动既有结构所以对旧版本命令完全兼容。3.2 后台发送线程的编写客户端消费者启动流程里注册成功后立刻启动一个后台心跳线程。我一开始想用 ScheduledExecutorService.scheduleAtFixedRate它的好处是任务按固定周期调度但坏处是如果某次发送耗时过长任务结束时会立刻补跑打出一小撮突刺流量而且内部单线程被阻塞后后续任务全部积压。最后我选择用 while 循环加 Thread.sleep 的自定义线程。发送过程中如果出现网络异常不立刻放弃而是走重试逻辑重试仍然失败就标记连接断开交由上层统一重建连接。核心代码如下class HeartbeatSender implements Runnable { private final NettyChannel channel; private final String groupId; private final String consumerId; private final String sessionId; private final long intervalMs; private final AtomicBoolean stopped new AtomicBoolean(false); Override public void run() { while (!stopped.get()) { long start System.currentTimeMillis(); try { channel.writeAndFlush(buildHeartbeatRequest()).sync(); } catch (Exception e) { log.warn(heartbeat send error, consumer{}, consumerId, e); notifyConnectionBroken(); } long cost System.currentTimeMillis() - start; long sleep Math.max(50, intervalMs - cost); try { Thread.sleep(sleep); } catch (InterruptedException e) { Thread.currentThread().interrupt(); break; } } } }Thread.sleep 为什么加了50毫秒的下限如果某次发送因为网络抖动花了3秒按 intervalMs - cost 计算出来是个负数外层循环就会变成没有间隔的空转心跳发送TPS瞬间被拉高这是雪上加霜。给它一个下限至少保证每轮之间有喘息时间也让调度更平滑。3.3 客户端优雅停止与心跳撤离消费者进程执行 close 时除了停止消费线程、提交最后一次位点还要主动向Broker发送一个注销指令让Broker不用等待心跳超时就能快速把队列分配给其他消费者。我这里的关闭顺序非常讲究绝对不能把顺序搞反。第一步先停止拉取消息第二步同步提交已经处理完毕的位点第三步再发送注销指令第四步停止心跳线程。如果反过来先发注销再提交位点Broker可能已经把这个消费者踢掉并触发重平衡位点提交请求到达时会被拒绝最后恢复出来的消费起点就偏了。心跳线程也要设置 stopped 并 join。很多人会忽略这一步结果消费者业务已经关闭后台心跳线程还在往Broker发数据Broker收到后认为会话仍然存活队列迟迟不释放其他消费者干着急。4. 服务端实现接收与过期扫描4.1 服务端心跳处理流程Broker收到心跳命令后处理逻辑并不复杂校验sessionId是否合法更新ConsumerSession的lastHeartbeatMs返回心跳响应。但麻烦的是更新时机和线程模型。这里有一个关键设计不要把所有请求都塞进同一个业务线程池尤其是心跳这种高频短耗时命令如果和长耗时的拉取请求排队心跳响应一旦延迟客户端就会误判Broker故障。我把心跳单独走一条快速通道在Netty里通过自定义Handler并指定独立EventExecutorGroup处理让心跳的稳定性和业务吞吐互不干扰。Handler核心逻辑如下class HeartbeatHandler extends SimpleChannelInboundHandlerHeartbeatRequest { Override protected void channelRead0(ChannelHandlerContext ctx, HeartbeatRequest req) { ConsumerSession session sessionManager.get(req.getGroupId(), req.getConsumerId()); if (session null || !session.getSessionId().equals(req.getSessionId())) { ctx.writeAndFlush(new HeartbeatResponse(FAILED, req.getGroupVersion(), SESSION_NOT_FOUND)); return; } session.refreshHeartbeat(); int newVersion rebalanceManager.getGroupVersion(req.getGroupId()); ctx.writeAndFlush(new HeartbeatResponse(SUCCESS, newVersion, null)); } }你可能想问心跳请求到底要不要回包理论上可以不回但回包有两个好处一是客户端能立刻感知自己是否还处于有效会话中如果收到 SESSION_NOT_FOUND就会主动走重新注册流程二是响应携带组版本号客户端发现版本和本地不一致时就能及时重新拉取队列分配结果。这样我们顺手把重平衡感知能力做进心跳里了。4.2 过期扫描器与时间窗口服务端需要一个扫描器定期把长时间没有心跳的会话标记为过期。我用一个后台线程每隔 sessionTimeoutMs 的三分之一扫描一次。为什么是这个频率如果扫描太频繁CPU浪费明显如果太稀疏发现失效的时间不稳定。取超时时间的三分之一可以保证最坏情况下的发现延迟可控同时扫描开销很小。扫描实现要非常小心不能在遍历集合时直接删除或者修改应该先收集过期列表再逐个处理。我采用的思路如下class ExpireScanner implements Runnable { Override public void run() { long now System.currentTimeMillis(); ListConsumerSession expired new ArrayList(); for (ConsumerSession s : sessionManager.sessions()) { if (now - s.getLastHeartbeatMs() sessionTimeoutMs) { expired.add(s); } } for (ConsumerSession s : expired) { if (s.compareAndSetExpired(false, true)) { consumerRebalance(s.getGroupId(), s.getConsumerId()); channelManager.close(s.getChannelId()); log.info(consumer session expired, group{}, consumer{}, session{}, s.getGroupId(), s.getConsumerId(), s.getSessionId()); } } } }compareAndSetExpired 这里我用CAS语义保证同一个会话只被清理一次。如果扫描线程多次同时执行或者和心跳恢复线程竞争没有这个CAS同一个消费者可能被触发多次重平衡造成资源浪费甚至重平衡风暴。4.3 失效消费者清理动作清理一个失效消费者不只是删除注册表记录它是一串动作的组合。第一个动作是把该消费者名下的队列重新标记为可分配让它们回到待分配池第二个动作是通知同消费组的其他消费者触发重平衡第三个动作是更新持久化位点表中的状态决定后续从哪个位置继续消费第四个动作是主动关闭这个消费者的网络连接防止它继续发送命令污染状态。这里有一个重要取舍清理时到底要不要等待消费者最后提交的位点如果消费者是彻底宕机位点可能还停留在它的内存里Broker侧只有最近一次提交的位点。所以Broker必须单独维护一份持久化的消费位点表心跳只负责判断存活位点交给消费提交请求处理两件事解耦。这是消息队列里非常核心的分层思想不要试图在一次心跳请求里把位点也带上来那样会让心跳协议变得臃肿也失去独立性。5. 心跳与位移提交、组重平衡的联动5.1 心跳周期内的Offset提交消费者要保证故障恢复后不丢消息必须把已经处理完的消息位点提交到Broker。位点提交的时机有两种同步提交和异步提交。心跳和位点提交不能混在一起位点是数据层面的东西可能高频变化而且每次提交可能涉及多个分区心跳是会话层面的东西周期固定。二者分开处理更清晰。不过二者有明确的时间关系一个消费者只有在心跳有效、会话存活的前提下它提交的位点才被接受。如果会话已经过期位点提交请求应该被拒绝或者忽略因为它的队列分配关系已经转移再让它提交位点会污染新的消费进度。我在服务端处理位点提交时第一步就是检查会话是否在线会话不在直接返回错误。这个检查看起来多余但能挡住很多异常场景。5.2 基于心跳的消费组重平衡这个系列实现重平衡的逻辑比较轻量Broker端维护一个每消费组可见的重平衡版本号 groupVersion每当过期消费者被清理或者新消费者加入version就自增。消费者拉取消息时发现版本变了就重新从Broker拉取队列分配结果。心跳在这里扮演的角色很有意思客户端的心跳请求里携带它当前看到的groupVersion服务端发现心跳带过来的版本和服务端不一致就把最新版本塞到心跳响应里返回。客户端收到响应后发现版本不一致立刻打断当前拉取并重新加入消费组。这就是我前面说心跳响应要携带版本号的意义所在。把这一块做通以后消费组才真正从静态配置变成动态系统。消费者可以随时加入、退出、崩溃、恢复队列分配始终收敛到一个均衡状态不会出现一台消费者消瘦到饿死、另一台消费者累死的不均衡局面。5.3 隔离假死消费者与消息重复这里要特别强调一个容易被忽略的坑心跳过期只代表Broker认为消费者不可用但消费者可能并没有真正死亡只是网络抖动或者GC停顿导致短时间内没上报心跳。此时如果Broker立刻把它踢掉并让其他消费者接管队列那么它恢复之后很可能继续拉取消息最后出现两个消费者同时消费同一批消息的局面消息重复率大幅上升。业界通常会用两层防线。第一层是容忍窗口让超时判定变得宽一些例如超时时间为心跳间隔的3到4倍。第二层是会话隔离我在注册时会生成sessionId让同一个consumerId在新旧会话之间彻底隔离。旧会话即使恢复并发起请求也会因为sessionId不匹配而被拒绝。消费者恢复后必须重新注册并加入消费组不再使用旧会话。如果你的业务对消息重复非常敏感还必须在应用消费侧做幂等去重。中间件能做的就是尽量降低重复率但不可能完全消除这个边界要让业务开发了解。6. 踩坑记录和排查清单6.1 误杀问题GC停顿与心跳抖动第一次联调测试时我模拟消费者JVM老年代GC停顿结果心跳线程也被冻住了。恢复之后Broker的扫描器发现会话超时把它标记过期并触发了重平衡其他消费者纷纷抢队列。停顿结束后原消费者也缓过来了向Broker发请求系统日志里瞬间刷出大量“SESSION_NOT_FOUND”的错误。这个坑的解法不是把会话超时无脑调大而是要基于消费者负载情况留余量。如果消费者所在的进程里有大量的CPU密集计算老年代常有长时间GC心跳间隔就不能设得太小超时时间更要留够。我最后把默认超时设到心跳的3倍再结合负载监控调整。另外一个容易被忽略的点是线程优先级。消费者内部线程很多时心跳线程的优先级不能设得太低否则高负载下它完全抢不到CPU系统看起来就像心跳线程被废弃了。我用守护线程并设置了带有consumerId的线程名方便出问题时用线程转储快速定位。6.2 并发更新导致的时间戳回退实现过程中出现过一次让人挠头的问题消费者端有两个线程并发消费不同的队列每个队列处理完都会调用同一个会话的 refresh 方法。我一开始是这样写的if (now session.getLastHeartbeatMs()) { session.setLastHeartbeatMs(now); }逻辑看起来没问题但读和写分离两个线程同时通过判断后后写入的旧值可能把新值覆盖导致时间戳回退。表面现象是心跳明明一直在发但Broker扫描时却认为超时了。修复方式很简单把字段从 volatile long 改成 AtomicLong并在更新时使用 accumlateAndGet 取较大值。并发场景下的时间戳更新必须用带原子性的方式不能靠“先读再写”这种组合操作。6.3 网络分区下心跳与长连接的表现线上最害怕的一种情况是网络分区Broker一直收不到心跳但TCP连接没有断开。扫描器发现超时后会关闭连接、清理会话并触发重平衡。等网络恢复后被踢的消费者重新发起连接再次注册此时它可能已经丢失了一部分未提交位点的消息造成从旧位点重复消费。对这个问题的策略是在重平衡之前先根据最后提交位点尽量分配同时尽量通知原消费者“你已被接管”。如果原消费者还能短暂通信就让它补交一次位点如果完全失联只能等它恢复后重新注册从Broker持久化的位点继续。没有完美的银弹能做到的就是尽早检测、尽快恢复把重复窗口压到最小。6.4 心跳扫描效率问题如果你的消费组数量很大成千上万个消费者都在发心跳扫描线程每次遍历所有会话肯定有性能问题。我做了一个优化按groupId分组维护一个嵌套的索引结构外层groupId到ConsumerSession集合的映射内层是每个消费者会话。扫描时只处理活跃组已经空掉的组直接跳过。另一个细节是不要把扫描和心跳刷新放在同一个锁内。心跳刷新是高频写操作扫描是低频率读操作两者共用一把锁会让心跳路径被拖慢。我最终把过期检查设计成无锁读取加CAS标记扫描线程只是在等到期时做一次失效标记刷新路径完全不加锁这样吞吐量才上得去。7. 最后说点心得心跳检测看起来只是定时发个包、定期清一下超时消费者但把它和重平衡、位点提交、会话管理串起来以后设计复杂度会上升很多。我最大的体会是中间件里的“小功能”都是有生命周期的从注册、续约到失效每一步都要处理并发、异常和恢复缺了哪一环线上都会出现诡异的问题。如果你也在从零写MQ建议不要最后才来补心跳最好在设计消费者注册表的时候就预留 lastHeartbeatMs 和 groupVersion 字段。另外日志里一定要打印清楚sessionId、groupId、超时时长这些上下文否则问题出现时只能靠猜排查成本会非常高。希望这份实现笔记能帮你少走几个弯路。