尧图网站建设 
简介这是一套面向WebSocket开发与测试人员的实用工具合集适合需要验证服务端与客户端通信有效性、排查连接异常或进行性能评估的开发者。资源围绕WebSocket协议的核心环节展开涵盖连接建立、握手协议、帧结构解析、文本与二进制数据传输、心跳保活及WSS安全通信等关键知识点可帮助读者快速搭建本地测试环境并观察真实数据交互过程。压缩包共19个文件约2.87MB包含js脚本、css与html页面、png和gif图示、exe可执行程序、dll动态库、rar子包及txt说明文档等兼顾在线测试页面与本地客户端、服务端工具便于按需选用。目前已有382人学习下载内容覆盖客户端测试工具、服务器端程序及使用说明读者可借助日志记录与错误报告定位帧解析异常、网络中断等问题并参考吞吐量与延迟测试思路评估服务承载能力适合实时聊天、在线协作等场景的调试与排错参考。1. WebSocket 测试工具为什么浏览器 DevTools 不够用调 WebSocket 的时候很多人第一反应是打开浏览器 F12看 Network 面板里的 WS 帧。这个做法在简单场景下没问题但一旦遇到心跳断连、二进制帧解析、批量压测、鉴权头定制这几类需求DevTools 就力不从心了。你没法主动构造一条带自定义 header 的握手请求没法按固定间隔自动发心跳没法模拟 500 个连接同时在线看服务端扛不扛得住更没法把收到的 protobuf 二进制帧直接反序列化成可读结构。这份 WebSocket 测试工具资源解决的就是上面这些「DevTools 干不了」的活。它面向的是需要主动调试、压测、验证 WebSocket 服务的后端和测试工程师——不是教你 WebSocket 协议本身而是给你一套能直接跑起来、能改参数、能看结果的工具。下面按「它是什么 → 怎么用 → 坑在哪 → 怎么进阶」的顺序拆开讲每一步都落到可复现的操作上。2. 工具能力拆解握手、帧收发、心跳、压测四件事2.1 握手阶段能改什么header、子协议、鉴权WebSocket 握手本质是一次 HTTP Upgrade 请求。测试工具的价值在于让你完全控制这次请求的头部。常见的做法是提供一个表单或配置文件让你填入 URL、自定义 header、Sec-WebSocket-Protocol 子协议列表。# 用 wscat 做一次带自定义 header 的握手工具包里通常也封装了等价命令 wscat -c wss://example.com/ws \ -H Authorization: Bearer eyJhbGciOi... \ -H X-Client-Version: 2.3.1 \ -s chat,superchat这里-H可以重复出现用来叠加多个 header-s指定子协议服务端会从中选一个回填到响应里。参数说明URL 必须带ws://或wss://前缀写成http://会直接握手失败。如果你用的是图形化工具对应位置一般在「Handshake」或「Headers」标签页逻辑一样。握手阶段最容易忽略的是 Origin 校验。很多服务端会检查Origin头是否在白名单里测试时如果没带上正确的 Origin会收到 403 而不是 101。工具里一般有单独的 Origin 输入框填成和服务端配置一致的值即可。2.2 帧收发文本、二进制、分片怎么测握手成功后进入帧收发阶段。WebSocket 的帧类型分文本opcode 0x1、二进制0x2、关闭0x8、Ping0x9、Pong0xA。测试工具需要能手动指定 opcode 发送而不是只发文本。# 用 websockets 库手动发一个二进制帧并读回响应 import asyncio import websockets async def send_binary(): async with websockets.connect(wss://example.com/ws) as ws: # 发送 4 字节二进制数据模拟 protobuf 消息头 await ws.send(b\x08\x01\x12\x03abc) resp await ws.recv() print(收到类型:, type(resp), 内容:, resp) asyncio.run(send_binary())逻辑说明websockets.connect返回的连接对象send传入bytes就发二进制帧传入str就发文本帧。参数上recv()返回的类型取决于服务端发的是什么可能是str也可能是bytes代码里必须做类型判断否则后续解析会翻车。分片帧是另一个测试点。协议允许一条消息拆成多个帧发送首帧 FIN0末帧 FIN1。大部分高级测试工具支持「手动分片」模式让你指定每片大小。常见做法是设成 1024 字节一片观察服务端能否正确重组。如果服务端没处理分片会在收到首帧后就报解析错误。2.3 心跳机制Ping/Pong 与业务层心跳的区别心跳是 WebSocket 测试里最容易出玄学问题的地方。协议层有 Ping/Pong 控制帧业务层又常常自己定义一条{type:ping}的文本消息。这两者不能混为一谈。// 浏览器端业务层心跳的典型实现 const ws new WebSocket(wss://example.com/ws); let heartbeatTimer null; ws.onopen () { heartbeatTimer setInterval(() { if (ws.readyState WebSocket.OPEN) { ws.send(JSON.stringify({ type: ping, ts: Date.now() })); } }, 30000); // 30 秒一次 }; ws.onmessage (e) { const msg JSON.parse(e.data); if (msg.type pong) { console.log(心跳正常RTT:, Date.now() - msg.ts, ms); } };参数说明间隔设 30 秒是常见值但必须小于服务端的空闲超时时间。如果服务端 60 秒没收到任何帧就断开你设 30 秒是安全的设 65 秒就会周期性断连。测试工具里一般能配置这个间隔并且能记录每次 Ping 到 Pong 的往返延迟用来判断链路质量。协议层 Ping 帧则由工具自动或手动触发。有些服务端只认协议层 Ping不认业务层 ping 文本测试时要两种都试。判断方法很简单发协议层 Ping看是否收到 Pong 控制帧发业务层 ping看是否收到对应文本。2.4 压测模式并发连接数与消息速率怎么设压测是这份工具区别于普通调试器的核心能力。你需要能设定并发连接数、每连接的消息发送速率、消息体大小然后观察服务端的 CPU、内存、连接数曲线。# 用工具包的压测脚本发起 200 并发连接每连接每秒发 10 条消息 python bench_ws.py \ --url wss://example.com/ws \ --connections 200 \ --rate 10 \ --size 512 \ --duration 60参数说明--connections是并发连接总数--rate是每连接每秒消息数--size是消息体字节数--duration是持续秒数。总消息速率 connections × rate 2000 条/秒。这个值要先估算服务端处理能力从小往大加别一上来就拉满否则服务端直接被打挂你连基线数据都拿不到。压测时重点看三个指标连接建立成功率、消息往返 P99 延迟、错误帧数量。成功率低于 99% 说明握手阶段有问题可能是文件描述符不够或鉴权限流P99 延迟突然抬升说明服务端处理队列积压错误帧数量增长说明有连接被异常关闭。3. 从零跑通一次完整测试环境、脚本、结果判读3.1 环境准备与依赖安装工具包通常提供 Python 和 Node.js 两套脚本。Python 侧依赖websockets和aiohttpNode.js 侧依赖ws。先确认版本避免因版本差异导致 API 不兼容。# Python 环境 python3 -m venv venv source venv/bin/activate pip install websockets12.0 aiohttp3.9.1 # Node.js 环境 npm init -y npm install ws8.16.0逻辑说明websockets12.0 的connect接口和 10.x 有差异老脚本直接跑可能报TypeError。参数上虚拟环境是为了隔离依赖避免污染系统 Python。Node.js 的ws8.x 默认不自动处理 Ping/Pong需要手动监听pong事件这点和 Python 库不同。3.2 编写第一个测试脚本连接、发消息、收响应import asyncio import websockets import json async def single_test(): uri wss://example.com/ws headers {Authorization: Bearer test-token} try: async with websockets.connect(uri, extra_headersheaders) as ws: # 发送一条业务消息 await ws.send(json.dumps({action: subscribe, channel: trades})) # 等待响应超时 5 秒 resp await asyncio.wait_for(ws.recv(), timeout5.0) print(响应:, resp) except asyncio.TimeoutError: print(超时服务端 5 秒内未响应) except websockets.exceptions.InvalidStatusCode as e: print(握手失败状态码:, e.status_code) asyncio.run(single_test())逻辑说明extra_headers用来传鉴权头注意不同版本参数名可能是additional_headers。asyncio.wait_for给recv加了 5 秒超时避免脚本卡死。参数上超时时间要大于服务端最慢响应时间否则会误判。捕获InvalidStatusCode能区分握手失败和业务超时排错时很有用。3.3 结果判读延迟、吞吐、错误率三个维度跑完脚本后结果不能只看「有没有报错」。要分三个维度看维度正常范围异常表现可能原因握手延迟 200ms 1sDNS 慢、TLS 握手重传消息 RTT 50ms波动大服务端 GC、网络抖动错误率0 0.1%连接被限流、帧格式错误判读时先看错误率有错误先解决错误错误率为零再看延迟分布P99 比平均值更有参考价值最后看吞吐是否达到预期。如果延迟正常但吞吐上不去瓶颈可能在客户端发送逻辑比如await send串行执行导致速率受限改成批量发送或并发发送能改善。4. 避坑与排查心跳断连、握手 403、二进制乱码4.1 现象连接 60 秒后必断日志显示 1006原因1006 是异常关闭码通常由服务端空闲超时触发。服务端在 60 秒内没收到任何帧就主动断开而客户端心跳间隔设成了 65 秒永远赶在超时之后。解决把心跳间隔改成小于服务端超时时间的一半比如服务端 60 秒超时心跳设 25 到 30 秒。同时确认心跳发的是协议层 Ping 还是业务层文本两者要和服务端约定一致。4.2 现象握手返回 403但用 curl 请求同 URL 正常原因WebSocket 握手会带Upgrade和Connection头部分网关或鉴权中间件对这两个头有额外校验。另外Origin头在浏览器外发起时可能为空服务端白名单校验不通过。解决在工具里显式设置Origin为服务端允许的值并确认Upgrade: websocket、Connection: Upgrade两个头没有被代理剥离。如果经过反向代理检查代理配置是否透传了这两个头。4.3 现象收到二进制帧后打印出来是乱码原因二进制帧内容是原始字节直接print会按终端编码解释自然乱码。这不是数据错误是显示方式问题。解决先判断帧类型二进制帧用bytes.hex()打印十六进制或按约定的 protobuf/MessagePack 反序列化。文本帧才用decode(utf-8)。测试工具里一般有「以十六进制显示」的开关打开后再看就正常了。4.4 现象压测到 200 并发时大量连接失败报 Too many open files原因操作系统对单进程文件描述符数量有限制默认 1024。每个 WebSocket 连接占一个 fd加上其他开销200 并发就可能触顶。解决临时调高限制ulimit -n 65535或在工具里把并发分批比如每批 50 个连接建连完成后保持再建下一批。长期方案是改系统配置但测试环境用ulimit就够了。4.5 现象消息发送成功但服务端说没收到原因send返回成功只代表数据写入了本地发送缓冲区不代表服务端已接收。如果连接在发送后立即关闭缓冲区数据可能还没发出去。解决发送后不要立刻关闭连接等收到服务端确认或至少等一个 RTT。压测脚本里常见做法是发送后await asyncio.sleep(0.1)再进入下一轮给缓冲区刷出时间。5. 进阶技巧用脚本批量验证心跳与断线重连5.1 自动化验证心跳稳定性手动测心跳只能看一时要验证长时间稳定性得靠脚本。思路是让脚本跑够长时间记录每次心跳的 RTT 和断连次数最后输出统计。import asyncio import websockets import time async def heartbeat_check(duration3600, interval30): uri wss://example.com/ws rtts [] disconnects 0 start time.time() while time.time() - start duration: try: async with websockets.connect(uri) as ws: while time.time() - start duration: t0 time.time() await ws.send({type:ping}) await asyncio.wait_for(ws.recv(), timeout10) rtts.append((time.time() - t0) * 1000) await asyncio.sleep(interval) except Exception as e: disconnects 1 print(断连:, e) await asyncio.sleep(2) # 重连前等待 print(f总心跳 {len(rtts)} 次断连 {disconnects} 次 f平均 RTT {sum(rtts)/len(rtts):.1f}ms f最大 RTT {max(rtts):.1f}ms) asyncio.run(heartbeat_check())逻辑说明外层while控制总时长内层while在单次连接内循环发心跳。断连后捕获异常、计数、等 2 秒重连。参数上duration设 3600 秒适合跑一小时稳定性interval要和 4.1 节的结论对齐。跑完后看断连次数如果一小时断超过 3 次说明心跳间隔或服务端配置还有问题。5.2 断线重连的退避策略重连不能死循环猛冲要用退避。常见做法是首次等 1 秒之后每次翻倍上限 30 秒。async def reconnect_with_backoff(uri, max_retries10): delay 1 for attempt in range(max_retries): try: ws await websockets.connect(uri) print(f第 {attempt1} 次重连成功) return ws except Exception as e: print(f重连失败{delay}s 后重试:, e) await asyncio.sleep(delay) delay min(delay * 2, 30) raise RuntimeError(重连次数耗尽)参数说明max_retries防止无限重试delay翻倍避免雪崩。上限 30 秒是经验值再长用户感知就差了。这段逻辑可以直接嵌到 5.1 的脚本里替换固定 2 秒等待。5.3 一个容易忽略的验证点关闭帧的 code 和 reason正常关闭应该收到 code 1000异常关闭是 1006 或其他。测试时主动发关闭帧并检查服务端返回的 code能发现不少服务端实现问题。async def check_close_code(): async with websockets.connect(wss://example.com/ws) as ws: await ws.close(code1000, reasonnormal closure) print(关闭码:, ws.close_code)如果close_code不是 1000说明服务端没按规范回关闭帧或者中间有代理改写了。这个检查点很多人不测但线上出问题时排查方向会因此少一个。我自己的习惯是每次改完服务端 WebSocket 逻辑先跑一遍 5.1 的心跳脚本跑 10 分钟再跑一遍 5.3 的关闭码检查两个都过了才认为这次改动是稳的。这个习惯帮我拦下过好几次「本地测着好好的、上线就断连」的问题。希望这套工具和上面的脚本能帮到你。本文还有配套的精品资源点击获取 WebSocket 测试工具实战：握手、心跳、压测与断线重连
简介这是一套面向WebSocket开发与测试人员的实用工具合集适合需要验证服务端与客户端通信有效性、排查连接异常或进行性能评估的开发者。资源围绕WebSocket协议的核心环节展开涵盖连接建立、握手协议、帧结构解析、文本与二进制数据传输、心跳保活及WSS安全通信等关键知识点可帮助读者快速搭建本地测试环境并观察真实数据交互过程。压缩包共19个文件约2.87MB包含js脚本、css与html页面、png和gif图示、exe可执行程序、dll动态库、rar子包及txt说明文档等兼顾在线测试页面与本地客户端、服务端工具便于按需选用。目前已有382人学习下载内容覆盖客户端测试工具、服务器端程序及使用说明读者可借助日志记录与错误报告定位帧解析异常、网络中断等问题并参考吞吐量与延迟测试思路评估服务承载能力适合实时聊天、在线协作等场景的调试与排错参考。1. WebSocket 测试工具为什么浏览器 DevTools 不够用调 WebSocket 的时候很多人第一反应是打开浏览器 F12看 Network 面板里的 WS 帧。这个做法在简单场景下没问题但一旦遇到心跳断连、二进制帧解析、批量压测、鉴权头定制这几类需求DevTools 就力不从心了。你没法主动构造一条带自定义 header 的握手请求没法按固定间隔自动发心跳没法模拟 500 个连接同时在线看服务端扛不扛得住更没法把收到的 protobuf 二进制帧直接反序列化成可读结构。这份 WebSocket 测试工具资源解决的就是上面这些「DevTools 干不了」的活。它面向的是需要主动调试、压测、验证 WebSocket 服务的后端和测试工程师——不是教你 WebSocket 协议本身而是给你一套能直接跑起来、能改参数、能看结果的工具。下面按「它是什么 → 怎么用 → 坑在哪 → 怎么进阶」的顺序拆开讲每一步都落到可复现的操作上。2. 工具能力拆解握手、帧收发、心跳、压测四件事2.1 握手阶段能改什么header、子协议、鉴权WebSocket 握手本质是一次 HTTP Upgrade 请求。测试工具的价值在于让你完全控制这次请求的头部。常见的做法是提供一个表单或配置文件让你填入 URL、自定义 header、Sec-WebSocket-Protocol 子协议列表。# 用 wscat 做一次带自定义 header 的握手工具包里通常也封装了等价命令 wscat -c wss://example.com/ws \ -H Authorization: Bearer eyJhbGciOi... \ -H X-Client-Version: 2.3.1 \ -s chat,superchat这里-H可以重复出现用来叠加多个 header-s指定子协议服务端会从中选一个回填到响应里。参数说明URL 必须带ws://或wss://前缀写成http://会直接握手失败。如果你用的是图形化工具对应位置一般在「Handshake」或「Headers」标签页逻辑一样。握手阶段最容易忽略的是 Origin 校验。很多服务端会检查Origin头是否在白名单里测试时如果没带上正确的 Origin会收到 403 而不是 101。工具里一般有单独的 Origin 输入框填成和服务端配置一致的值即可。2.2 帧收发文本、二进制、分片怎么测握手成功后进入帧收发阶段。WebSocket 的帧类型分文本opcode 0x1、二进制0x2、关闭0x8、Ping0x9、Pong0xA。测试工具需要能手动指定 opcode 发送而不是只发文本。# 用 websockets 库手动发一个二进制帧并读回响应 import asyncio import websockets async def send_binary(): async with websockets.connect(wss://example.com/ws) as ws: # 发送 4 字节二进制数据模拟 protobuf 消息头 await ws.send(b\x08\x01\x12\x03abc) resp await ws.recv() print(收到类型:, type(resp), 内容:, resp) asyncio.run(send_binary())逻辑说明websockets.connect返回的连接对象send传入bytes就发二进制帧传入str就发文本帧。参数上recv()返回的类型取决于服务端发的是什么可能是str也可能是bytes代码里必须做类型判断否则后续解析会翻车。分片帧是另一个测试点。协议允许一条消息拆成多个帧发送首帧 FIN0末帧 FIN1。大部分高级测试工具支持「手动分片」模式让你指定每片大小。常见做法是设成 1024 字节一片观察服务端能否正确重组。如果服务端没处理分片会在收到首帧后就报解析错误。2.3 心跳机制Ping/Pong 与业务层心跳的区别心跳是 WebSocket 测试里最容易出玄学问题的地方。协议层有 Ping/Pong 控制帧业务层又常常自己定义一条{type:ping}的文本消息。这两者不能混为一谈。// 浏览器端业务层心跳的典型实现 const ws new WebSocket(wss://example.com/ws); let heartbeatTimer null; ws.onopen () { heartbeatTimer setInterval(() { if (ws.readyState WebSocket.OPEN) { ws.send(JSON.stringify({ type: ping, ts: Date.now() })); } }, 30000); // 30 秒一次 }; ws.onmessage (e) { const msg JSON.parse(e.data); if (msg.type pong) { console.log(心跳正常RTT:, Date.now() - msg.ts, ms); } };参数说明间隔设 30 秒是常见值但必须小于服务端的空闲超时时间。如果服务端 60 秒没收到任何帧就断开你设 30 秒是安全的设 65 秒就会周期性断连。测试工具里一般能配置这个间隔并且能记录每次 Ping 到 Pong 的往返延迟用来判断链路质量。协议层 Ping 帧则由工具自动或手动触发。有些服务端只认协议层 Ping不认业务层 ping 文本测试时要两种都试。判断方法很简单发协议层 Ping看是否收到 Pong 控制帧发业务层 ping看是否收到对应文本。2.4 压测模式并发连接数与消息速率怎么设压测是这份工具区别于普通调试器的核心能力。你需要能设定并发连接数、每连接的消息发送速率、消息体大小然后观察服务端的 CPU、内存、连接数曲线。# 用工具包的压测脚本发起 200 并发连接每连接每秒发 10 条消息 python bench_ws.py \ --url wss://example.com/ws \ --connections 200 \ --rate 10 \ --size 512 \ --duration 60参数说明--connections是并发连接总数--rate是每连接每秒消息数--size是消息体字节数--duration是持续秒数。总消息速率 connections × rate 2000 条/秒。这个值要先估算服务端处理能力从小往大加别一上来就拉满否则服务端直接被打挂你连基线数据都拿不到。压测时重点看三个指标连接建立成功率、消息往返 P99 延迟、错误帧数量。成功率低于 99% 说明握手阶段有问题可能是文件描述符不够或鉴权限流P99 延迟突然抬升说明服务端处理队列积压错误帧数量增长说明有连接被异常关闭。3. 从零跑通一次完整测试环境、脚本、结果判读3.1 环境准备与依赖安装工具包通常提供 Python 和 Node.js 两套脚本。Python 侧依赖websockets和aiohttpNode.js 侧依赖ws。先确认版本避免因版本差异导致 API 不兼容。# Python 环境 python3 -m venv venv source venv/bin/activate pip install websockets12.0 aiohttp3.9.1 # Node.js 环境 npm init -y npm install ws8.16.0逻辑说明websockets12.0 的connect接口和 10.x 有差异老脚本直接跑可能报TypeError。参数上虚拟环境是为了隔离依赖避免污染系统 Python。Node.js 的ws8.x 默认不自动处理 Ping/Pong需要手动监听pong事件这点和 Python 库不同。3.2 编写第一个测试脚本连接、发消息、收响应import asyncio import websockets import json async def single_test(): uri wss://example.com/ws headers {Authorization: Bearer test-token} try: async with websockets.connect(uri, extra_headersheaders) as ws: # 发送一条业务消息 await ws.send(json.dumps({action: subscribe, channel: trades})) # 等待响应超时 5 秒 resp await asyncio.wait_for(ws.recv(), timeout5.0) print(响应:, resp) except asyncio.TimeoutError: print(超时服务端 5 秒内未响应) except websockets.exceptions.InvalidStatusCode as e: print(握手失败状态码:, e.status_code) asyncio.run(single_test())逻辑说明extra_headers用来传鉴权头注意不同版本参数名可能是additional_headers。asyncio.wait_for给recv加了 5 秒超时避免脚本卡死。参数上超时时间要大于服务端最慢响应时间否则会误判。捕获InvalidStatusCode能区分握手失败和业务超时排错时很有用。3.3 结果判读延迟、吞吐、错误率三个维度跑完脚本后结果不能只看「有没有报错」。要分三个维度看维度正常范围异常表现可能原因握手延迟 200ms 1sDNS 慢、TLS 握手重传消息 RTT 50ms波动大服务端 GC、网络抖动错误率0 0.1%连接被限流、帧格式错误判读时先看错误率有错误先解决错误错误率为零再看延迟分布P99 比平均值更有参考价值最后看吞吐是否达到预期。如果延迟正常但吞吐上不去瓶颈可能在客户端发送逻辑比如await send串行执行导致速率受限改成批量发送或并发发送能改善。4. 避坑与排查心跳断连、握手 403、二进制乱码4.1 现象连接 60 秒后必断日志显示 1006原因1006 是异常关闭码通常由服务端空闲超时触发。服务端在 60 秒内没收到任何帧就主动断开而客户端心跳间隔设成了 65 秒永远赶在超时之后。解决把心跳间隔改成小于服务端超时时间的一半比如服务端 60 秒超时心跳设 25 到 30 秒。同时确认心跳发的是协议层 Ping 还是业务层文本两者要和服务端约定一致。4.2 现象握手返回 403但用 curl 请求同 URL 正常原因WebSocket 握手会带Upgrade和Connection头部分网关或鉴权中间件对这两个头有额外校验。另外Origin头在浏览器外发起时可能为空服务端白名单校验不通过。解决在工具里显式设置Origin为服务端允许的值并确认Upgrade: websocket、Connection: Upgrade两个头没有被代理剥离。如果经过反向代理检查代理配置是否透传了这两个头。4.3 现象收到二进制帧后打印出来是乱码原因二进制帧内容是原始字节直接print会按终端编码解释自然乱码。这不是数据错误是显示方式问题。解决先判断帧类型二进制帧用bytes.hex()打印十六进制或按约定的 protobuf/MessagePack 反序列化。文本帧才用decode(utf-8)。测试工具里一般有「以十六进制显示」的开关打开后再看就正常了。4.4 现象压测到 200 并发时大量连接失败报 Too many open files原因操作系统对单进程文件描述符数量有限制默认 1024。每个 WebSocket 连接占一个 fd加上其他开销200 并发就可能触顶。解决临时调高限制ulimit -n 65535或在工具里把并发分批比如每批 50 个连接建连完成后保持再建下一批。长期方案是改系统配置但测试环境用ulimit就够了。4.5 现象消息发送成功但服务端说没收到原因send返回成功只代表数据写入了本地发送缓冲区不代表服务端已接收。如果连接在发送后立即关闭缓冲区数据可能还没发出去。解决发送后不要立刻关闭连接等收到服务端确认或至少等一个 RTT。压测脚本里常见做法是发送后await asyncio.sleep(0.1)再进入下一轮给缓冲区刷出时间。5. 进阶技巧用脚本批量验证心跳与断线重连5.1 自动化验证心跳稳定性手动测心跳只能看一时要验证长时间稳定性得靠脚本。思路是让脚本跑够长时间记录每次心跳的 RTT 和断连次数最后输出统计。import asyncio import websockets import time async def heartbeat_check(duration3600, interval30): uri wss://example.com/ws rtts [] disconnects 0 start time.time() while time.time() - start duration: try: async with websockets.connect(uri) as ws: while time.time() - start duration: t0 time.time() await ws.send({type:ping}) await asyncio.wait_for(ws.recv(), timeout10) rtts.append((time.time() - t0) * 1000) await asyncio.sleep(interval) except Exception as e: disconnects 1 print(断连:, e) await asyncio.sleep(2) # 重连前等待 print(f总心跳 {len(rtts)} 次断连 {disconnects} 次 f平均 RTT {sum(rtts)/len(rtts):.1f}ms f最大 RTT {max(rtts):.1f}ms) asyncio.run(heartbeat_check())逻辑说明外层while控制总时长内层while在单次连接内循环发心跳。断连后捕获异常、计数、等 2 秒重连。参数上duration设 3600 秒适合跑一小时稳定性interval要和 4.1 节的结论对齐。跑完后看断连次数如果一小时断超过 3 次说明心跳间隔或服务端配置还有问题。5.2 断线重连的退避策略重连不能死循环猛冲要用退避。常见做法是首次等 1 秒之后每次翻倍上限 30 秒。async def reconnect_with_backoff(uri, max_retries10): delay 1 for attempt in range(max_retries): try: ws await websockets.connect(uri) print(f第 {attempt1} 次重连成功) return ws except Exception as e: print(f重连失败{delay}s 后重试:, e) await asyncio.sleep(delay) delay min(delay * 2, 30) raise RuntimeError(重连次数耗尽)参数说明max_retries防止无限重试delay翻倍避免雪崩。上限 30 秒是经验值再长用户感知就差了。这段逻辑可以直接嵌到 5.1 的脚本里替换固定 2 秒等待。5.3 一个容易忽略的验证点关闭帧的 code 和 reason正常关闭应该收到 code 1000异常关闭是 1006 或其他。测试时主动发关闭帧并检查服务端返回的 code能发现不少服务端实现问题。async def check_close_code(): async with websockets.connect(wss://example.com/ws) as ws: await ws.close(code1000, reasonnormal closure) print(关闭码:, ws.close_code)如果close_code不是 1000说明服务端没按规范回关闭帧或者中间有代理改写了。这个检查点很多人不测但线上出问题时排查方向会因此少一个。我自己的习惯是每次改完服务端 WebSocket 逻辑先跑一遍 5.1 的心跳脚本跑 10 分钟再跑一遍 5.3 的关闭码检查两个都过了才认为这次改动是稳的。这个习惯帮我拦下过好几次「本地测着好好的、上线就断连」的问题。希望这套工具和上面的脚本能帮到你。本文还有配套的精品资源点击获取