最近在做实时行情相关的项目，刚开始觉得很简单：连上、收消息、处理数据就完事了。结果线上跑几天后发现各种问题：
- 网络抖动导致连接断开
- 服务端主动踢连接
- 长时间没消息被中间设备断开
- 重连太频繁把自己服务器打挂
后来重新整理了一套比较稳的方案，核心有 3 个：
- 心跳
- 重连
- 退避机制
1.心跳不能省
WebSocket 建立连接不一定就一直在线。运营商网络、NAT 设备、负载均衡器都可能清理长时间无流量的连接。所以一定要做心跳。 如果服务端支持 ping/pong ，直接使用协议层心跳即可。 如果不支持，可以定时发送业务心跳消息。 例如：
async def heartbeat(ws):
while True:
try:
await ws.ping()
await asyncio.sleep(30)
except Exception:
break
30 秒左右发一次通常就够用了。
2.重连是标配
WebSocket 断开是常态，不是异常。所以不要把代码写成：
await websocket.recv()
然后指望它永远不掉线。 我觉得应该是：
while True:
try:
await connect()
except Exception as e:
print(e)
await asyncio.sleep(5)
连接断开以后自动重新建立连接。这样即使服务端重启或者网络闪断，也能自动恢复。
3.别直接疯狂重连
一个常见错误可能是⬇️
断线：
while True:
try:
connect()
except:
pass
结果：
- CPU 飙高
- 日志刷爆
- 服务端压力暴增
这时候就需要退避机制，也就是每次失败后逐渐增加等待时间。例如：
delay = 1
while True:
try:
await connect()
delay = 1
except Exception:
await asyncio.sleep(delay)
delay = min(delay \* 2, 60)
等待时间变成 1s 、2s 、4s 、8s 、16s 、32s 、60s...
连接恢复后再重置。这样会稳定很多。
4.WebSocket 行情订阅
连接建立后订阅需要的品种即可，以 BTC 为例：
import asyncio
import json
import websockets
WS\_URL = "wss://quote.alltick.io/quote-b-ws-api"
API\_TOKEN = "YOUR\_API\_TOKEN"
async def subscribe():
async with websockets.connect(WS\_URL) as ws:
sub\_msg = {
"cmd\_id": 22002,
"seq\_id": 1,
"trace": "test",
"data": {
"symbol\_list": \[
{
"code": "BTCUSDT"
}
\]
},
"token": API\_TOKEN
}
await ws.send(json.dumps(sub\_msg))
while True:
msg = await ws.recv()
print(msg)
asyncio.run(subscribe())
如果是生产环境，我一般会把：
- 连接管理
- 心跳任务
- 重连逻辑
- 退避机制
全部封装到一个 ConnectionManager 里面。业务层只负责消费消息，不关心连接状态。这样后面维护会轻松很多。
Select Language