手写最小 WebSocket 封装：百行左右理解心跳、重连与消息分发
上一篇《心跳、重连、回执、活跃状态：一个 WebSocket 封装到底替你扛了多少事》拆完那层两百多行的封装，一个很自然的念头是：“这些骨架我是不是也能写？”
单拎出心跳、重连、topic 分发，每个都是百字内讲完的知识点；但“能发送心跳”和“能检测半开连接”、“会重连”和“不会制造重连风暴”之间都隔着不少细节。这篇先用百行左右写出一个可运行骨架，再通过实验主动暴露它的缺口，看清最小版和生产级之间的距离。
🎯 本文目标
读完你应该能：独立写出一个带单例、topic 分发、心跳发送、重连和资源清理的最小 WebSocket 封装；用两个实验验证断线重连，并观察“只发心跳”为什么仍然无法可靠判定半开连接；拿着补课清单知道自己的版本离生产级还差哪几步。
动手之前：先确认你真的需要它
写封装是有成本的，动手前先过三道检查，任何一道过关就别写，用现成的就够：
只有服务端到客户端的单向推送——比如纯消息下发，SSE 往往更合适：浏览器 EventSource 内置断线重连，也更容易复用 HTTP 基础设施。服务端仍可能需要发送注释心跳，避免代理因长时间无数据而关闭连接，但客户端连接管理通常比 WebSocket 简单。SSE 和 WebSocket 怎么选，我在流式输出那篇里专门对比过。
消息频率低到分钟级——轮询并没有死透，分钟级的"准实时"用轮询，成本和复杂度都更低。别为了技术正确性上长连接。
只有一个页面、一种消息——通常没必要上升为全局框架，但仍可以做一层很薄的局部封装，集中处理关闭、异常和测试。封装的收益来自复用连接与统一策略，复杂度要跟消费方数量和可靠性要求匹配。
三道检查都不过关——多个模块要消费消息、需要双向通信、频率在秒级——好，那就值得动手了。判断标准一句话：当"连接"成为被多个模块争用的全局资源时，才值得为它做一层治理。
百行左右：把骨架写出来
完整代码如下，保留单例、topic 分发、心跳发送、重连和销毁五个骨架。它仍然不是生产级实现，但每个缺口都会在后文明确标出来：
// mini-ws.js —— 最小可用的 WebSocket 封装
function createSocketContext(WebSocketImpl = WebSocket) {
const ctx = {
socket: null,
ready: false,
callbacks: new Map(), // topic -> Set<回调>
heartbeatTimer: null,
reconnectTimer: null,
reconnectCount: 0,
maxReconnectCount: 5,
manuallyClosed: false,
url: '',
}
ctx.onOpen = () => {
ctx.ready = true
ctx.reconnectCount = 0
ctx.startHeartbeat()
}
ctx.onMessage = (evt) => {
let msg
try {
msg = JSON.parse(evt.data)
} catch {
return console.warn('收到无法解析的 WebSocket 消息')
}
if (!msg || typeof msg !== 'object' || typeof msg.topic !== 'string') {
return console.warn('收到格式不合法的 WebSocket 消息')
}
const systemTopics = new Set(['heartbeat', 'receive'])
if (msg.content?.id && !systemTopics.has(msg.topic)) {
ctx.send({
topic: 'receive',
content: { type: 'system', receiveMessageId: msg.content.id },
})
}
const listeners = ctx.callbacks.get(msg.topic)
if (listeners) listeners.forEach((callback) => callback(msg))
else if (!systemTopics.has(msg.topic)) console.warn(`消息事件[${msg.topic}]未绑定`)
}
ctx.onClose = () => {
ctx.ready = false
ctx.stopHeartbeat()
if (!ctx.manuallyClosed) ctx.reconnect()
}
ctx.startHeartbeat = () => {
ctx.stopHeartbeat()
ctx.heartbeatTimer = setInterval(() => {
ctx.send({ topic: 'heartbeat', content: { type: 'system' } })
}, 10000)
}
ctx.stopHeartbeat = () => clearInterval(ctx.heartbeatTimer)
ctx.reconnect = () => {
if (ctx.reconnectTimer || ctx.reconnectCount >= ctx.maxReconnectCount) return
const delay = Math.min(1000 * 2 ** ctx.reconnectCount, 30000)
ctx.reconnectCount++
ctx.reconnectTimer = setTimeout(() => {
ctx.reconnectTimer = null
ctx.connect()
}, delay + Math.random() * 500)
}
ctx.send = (msg) => {
if (ctx.socket?.readyState === WebSocketImpl.OPEN) {
ctx.socket.send(JSON.stringify(msg))
}
}
ctx.subscribe = (topic, callback) => {
if (!ctx.callbacks.has(topic)) ctx.callbacks.set(topic, new Set())
ctx.callbacks.get(topic).add(callback)
return () => ctx.callbacks.get(topic)?.delete(callback)
}
ctx.connect = (url) => {
ctx.url = url ?? ctx.url
ctx.manuallyClosed = false
if (!ctx.url || [WebSocketImpl.CONNECTING, WebSocketImpl.OPEN].includes(ctx.socket?.readyState)) return
ctx.socket = new WebSocketImpl(ctx.url)
ctx.socket.onopen = ctx.onOpen
ctx.socket.onmessage = ctx.onMessage
ctx.socket.onclose = ctx.onClose
}
ctx.destroy = () => {
ctx.manuallyClosed = true
ctx.stopHeartbeat()
clearTimeout(ctx.reconnectTimer)
ctx.reconnectTimer = null
ctx.callbacks.clear()
const socket = ctx.socket
ctx.socket = null
if (socket) {
socket.onopen = socket.onmessage = socket.onclose = null
socket.close(1000, 'client destroy')
}
ctx.ready = false
}
return ctx
}
const socket = createSocketContext() // 模块级单例
export default socket
// 用法：
// socket.connect('wss://example.com/ws?token=xxx')
// const unsubscribe = socket.subscribe('notification', (msg) => render(msg.content))
// 页面卸载时：unsubscribe()逐个说清几个“看起来平平无奇，其实是有意为之”的地方。
为什么是模块级单例。createSocketContext() 只在模块加载时调用一次，之后同一 JavaScript 运行时中、引用同一份模块实例的 import 会拿到同一个 socket。它不会自动跨浏览器标签页、iframe 或独立打包的微前端共享连接。构造函数支持注入 WebSocketImpl，则是为了让测试可以传入 fake WebSocket。
为什么 onClose 里先停心跳再重连。断线后心跳定时器如果不清，会在重连等待期继续空转；重连成功后 startHeartbeat 又会开启新定时器。先清理旧定时器，是避免重复心跳最便宜的保险。
为什么重连同时需要锁、退避和额度。reconnectTimer 保证同一时刻只有一个待执行任务；等待时间按 1s、2s、4s……增长，并加一点随机抖动；累计发起 5 次仍未连接成功就停止。连接一旦成功，onOpen 把计数归零。额度的本质是承认一个事实：客户端能覆盖短暂网络抖动，却不能靠无限握手修复服务端故障。
为什么回调使用 Map<string, Set<Function>>。同一个 topic 可能有多个消费者，Set 避免后订阅者覆盖先订阅者；subscribe 返回反订阅函数，让页面卸载时可以清理自己的回调，而不必销毁整条全局连接。
为什么回执不能无条件发送。示例只给带有效 content.id 的非系统消息回 ACK，避免 receive 消息再次触发 receive。ACK 只表示连接层已收到消息；是否处理成功，仍应由业务协议单独确认。
为什么消息必须走 { topic, content } 协议。解析之后按 topic 查表分发，分发逻辑收敛在连接层，业务只跟 topic 打交道。不过“新增 topic 零成本”只是对连接层而言，前后端协议、类型定义、权限和监控仍然需要同步演进。
跑通它：两个实验
配任何一个 ws 服务就能跑。最省事的是 Node 起 ws 库的受控 echo 服务。注意不能把 ACK 也原样回发，否则客户端和服务端会互相制造 receive 消息：
// echo-server.js —— node echo-server.js
import { WebSocketServer } from 'ws'
const wss = new WebSocketServer({ port: 8765 })
wss.on('connection', (ws) => {
ws.on('message', (data) => {
const msg = JSON.parse(data.toString())
console.log('收到:', msg)
if (msg.topic === 'receive') return // ACK 到此为止，不再 echo
ws.send(JSON.stringify(msg))
})
})跑通之后，别急着庆祝，做两个实验——比读十篇文章管用：
实验一：断线重连。连接建立后杀掉服务进程，观察 onClose 触发，随后客户端按 1s、2s、4s……退避重连；服务恢复后重新触发 onOpen，心跳重启，重连计数归零。再开十个浏览器标签页做一次，你会看到模块单例并不能跨标签页复用连接，也能直观看到 jitter 为什么有价值。
实验二：证明“只发心跳”仍然不够。修改服务端：收到 heartbeat 后直接 return，但保持 WebSocket 不关闭。客户端仍会每秒或每 10 秒发送心跳，ctx.ready 也一直是 true，不会触发重连。这个结果恰好证明：发送心跳只能提供探测信号；要检测半开连接，客户端还必须等待 heartbeat ACK/pong，并在超过 N 个周期没有响应时主动 close()。
⚠️ 不要用“拔网线后触发了 onClose”证明心跳有效
拔网线后是否快速触发 onClose，主要取决于浏览器、操作系统和网络栈。即使删除应用层心跳，它也可能照样触发；反过来，即使持续发送心跳，没有响应超时逻辑也可能很久都不判死。验证心跳机制应控制“服务端是否响应”，而不是只控制本地网络开关。
从最小骨架到生产级：补课清单
最小版能跑之后，离生产级仍隔着一张清单。这里也不把上一篇的项目封装当作“标准答案”：最小版有些地方比它多做了一步，两者也都有尚未完成的能力。
| # | 能力 | 最小版现状 | 缺了会怎样 | 生产级做法 |
|---|---|---|---|---|
| 1 | 重连控制 | 已用单个 timer、指数退避和简单 jitter | 多实例仍可能同时重连；网络状态切换也未参与调度 | 使用全抖动退避，结合 online/offline，明确不同 close code 是否可重试 |
| 2 | 半开判定 | 只发心跳，不检测回应 | 心跳变成单向的自言自语，连接死了也不知道 | 心跳带序号，超过 N 个周期没收到回执即判死、主动断开重连 |
| 3 | 回执 | 只发送连接层 ACK | ACK 丢失、服务端未重发时仍然会丢消息 | 服务端保存未确认消息并重发，客户端按 ID 去重；业务成功另设应答 |
| 4 | 订阅生命周期 | 已支持多回调和 unsubscribe | 忘记调用仍会泄漏；单个回调异常可能打断后续回调 | 框架生命周期自动解绑，逐个隔离回调异常并记录监控 |
| 5 | 页面活跃状态 | 无 | 后台仍按前台策略接收消息 | visibilitychange 同步 active/inactive，具体降级由服务端决定 |
| 6 | 跨上下文共享 | 每个标签页、iframe 各自连接 | 同一用户可能挂多条连接 | SharedWorker 统一持有连接，或用 BroadcastChannel/postMessage 建消息桥 |
| 7 | 销毁清理 | 已清 timer、回调并主动 close | 极端事件时序仍可能访问旧实例 | 解绑原生事件、验证 fake timer 与 close/reconnect 竞态 |
| 8 | 可测试性 | 已支持注入 WebSocket 构造器 | 尚未提供现成 fake 与时钟控制 | fake WebSocket + fake timer 覆盖 open/message/error/close 全时序 |
| 9 | 发送背压 | 未处理 bufferedAmount | 消费端变慢时内存持续堆积 | 设队列上限、丢弃或降采样策略，监控 bufferedAmount |
| 10 | 鉴权续期 | URL 由调用方传入 | token 过期后可能陷入无效重连 | 使用短时凭证，按关闭原因刷新凭证后再重连 |
这张表的信息量不在“怎么做”，而在能力不能靠命名获得：有 heartbeat() 不等于能判定半开，有 receive 不等于 at-least-once，有 destroy() 也不等于所有竞态都清干净。判断一个封装是否可靠，要沿着失败时序检查状态是否真的闭环。
“什么时候停止补课”没有固定序号。只要用于生产，半开判定、可控重连、完整销毁、消息校验和可测试性属于基线；可靠投递、跨上下文共享、背压与鉴权续期，则取决于业务是否真的需要。封装的度要与失败成本匹配，而不是与代码行数匹配。
结语：写最小版的意义
最后回答一个可能一直悬着的问题：既然网上和公司里都有现成封装，为什么还要自己写一遍这个最小骨架？
因为最小版是你理解任何封装的坐标系。写完它，再去读上一篇的真实包，就能看出哪些代码在管理资源、哪些代码在定义协议，也能识别“字段存在但能力没有闭环”的情况。选型时不再只看 API 名字，而会追问：重连任务能否并发？心跳有没有响应超时？销毁后还有没有定时器？知识可以搬运，判断力来自对失败路径的验证。
这也是这个站反复出现的方法：先手写最小实现，再看框架替你做了什么。RAG 是这么写的（最小检索链路），Agent 是这么写的（手写循环对账 LangGraph），WebSocket 也不会是最后一个。
🤔 动手时的几个追问（点开看）
为什么 ACK 放在业务分发之前，还要过滤系统 topic？ 连接层 ACK 表示“消息已收到并通过基础解析”，不依赖业务回调是否成功，所以放在分发之前。它只应发送给协议约定需要确认、且带有效 ID 的消息；给 receive 本身再回 ACK，会在 echo 或错误服务端实现下形成回执循环。
心跳间隔怎么定？ 没有脱离环境的安全数字。间隔应短于链路中最小的 idle timeout，同时结合服务端容量、期望故障发现时间、移动网络耗电和后台定时器节流确定。更关键的是定义“允许连续丢几个响应”，而不是只决定多久发一次。
怎么给这个封装写单测？ 示例已经让 createSocketContext 接收 WebSocket 构造器。测试时传一个实现 CONNECTING/OPEN、send/close 和事件触发方法的 fake 类，再配 fake timer，就能稳定模拟 open/message/close 的任意时序。重点覆盖重复 close、销毁后 close、重连等待中恢复和非法消息，而不只是“能否连接成功”。
为什么指数退避之后还要 jitter？ 如果 1000 个客户端同时断线，它们即使都按 1s、2s、4s 退避，仍会在相同时间发起握手。jitter 把每个客户端的等待时间打散，避免服务恢复瞬间再次被同步请求压垮。生产实现通常采用 full jitter，而不是只固定加 500ms。