手写消息队列这个系列写到第六篇，终于轮到消费者心跳检测（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 可以彻底避开这个坑。
服务端用一个 ConcurrentHashMap<String, 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, opCode=HEARTBEAT, groupId, consumerId, groupVersion, timestamp } HeartbeatResponse { opCode=HEARTBEAT_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 SimpleChannelInboundHandler<HeartbeatRequest> { @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(); List<ConsumerSession> 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、超时时长这些上下文，否则问题出现时只能靠猜，排查成本会非常高。希望这份实现笔记能帮你少走几个弯路。
