长连接断线重连是行情接入里最容易被低估的一段代码。多数团队会认真实现「断了就重连」，却很少处理重连之后的状态一致性——结果就是下游收到重复的 tick，指标被重复累加，告警被重复触发，而排查时又很难复现。
这篇文章拆解三个最常见的订阅状态陷阱，并给出一个可用的重连状态机。
为什么重连一定会产生重复
要先接受一个前提：重连产生重复是正常的，去重才是应用层的责任。
原因在于 WebSocket 的语义。断线发生时，服务端可能已经推送了若干帧但你没收到；重连后重新订阅，服务端会从当前时点继续推。这两段区间之间没有「接着上次继续」的机制，所以边界处必然存在重叠。
好消息是 TIQEX 的 tick 帧带有 at 字段（RFC3339 时间戳），这让去重有据可依——用 (symbol, at) 作为幂等键即可。
陷阱一：把「连接成功」当成「订阅成功」
这是最常见的错误。代码大致长这样：
async with websockets.connect(URL, additional_headers=HEADERS) as ws:
await ws.send(json.dumps({"action": "subscribe", "market": "crypto", "symbols": SYMBOLS}))
print("已订阅") # 错：这只是「已发送」，不是「已订阅」
async for raw in ws:
handle(raw)
问题在于：send 只保证消息进了本地发送缓冲区。订阅是否被服务端接受，要看后续的响应帧。TIQEX 会返回一条 type=subscribed 的确认帧，携带 market 与 symbols；如果订阅失败，则会收到 type=error 帧。
正确做法是等到确认帧再认为订阅生效。否则在订阅被拒（例如 MARKET_NOT_ENABLED 或 NO_SUBSCRIPTION）的情况下，代码会安静地卡在 async for 上，既不报错也不推数据，排查起来非常费时。
陷阱二：重连后不清理本地状态
重连后，很多实现会保留上一次连接累积的状态：去重缓存、聚合窗口、告警去抖计时器。这些状态在重连边界上已经失效了。
具体表现是：
- 去重缓存不清：如果缓存按「最近 N 条」设计，重连后新数据的键可能与缓存里的旧键冲突，导致新 tick 被误判为重复而丢弃——这比重复更糟，因为它是静默丢数据。
- 聚合窗口不清：重连前未完成的分钟线聚合会与重连后的数据拼接，产生错误的 OHLC。
- 去抖计时器不清：跨越重连的告警计时会异常触发或漏触发。
原则很简单：重连后清空所有「与连接生命周期绑定」的状态，只保留业务层面的持久状态。 去重缓存应该带时间窗（比如只保留最近 5 秒的键），这样重连后自然过期，不需要额外清理逻辑。
陷阱三：心跳与订阅混在同一个超时里
心跳超时（多久没收到任何帧就判定断线）和订阅超时（发出 subscribe 后多久没收到确认帧就判定失败）是两件事，混用会导致两种误判：
- 用较长的订阅超时做心跳：真实断线要等很久才被发现，期间数据静默丢失。
- 用较短的心跳超时做订阅等待：在冷门市场或低波动时段，可能长时间没有 tick，于是被误判为订阅失败并反复重连——而重连握手是计入额度的，这会白白消耗配额。
正确做法是分开两个计时器：订阅确认用较短的固定超时（比如 10 秒），心跳用较长的静默超时（比如 60 秒），并且优先用协议层的 ping/pong 而不是靠数据帧间隔判断存活。
一个可用的重连状态机
把上面的原则落到状态上，重连逻辑就清晰了：
| 状态 | 进入条件 | 退出动作 |
|---|---|---|
| CONNECTING | 发起连接 | 握手成功 → SUBSCRIBING |
| SUBSCRIBING | 已发送 subscribe | 收到 subscribed → LIVE；超时或 error → BACKOFF |
| LIVE | 订阅已确认 | 断开或心跳超时 → BACKOFF |
| BACKOFF | 等待重连 | 退避结束 → CONNECTING（并清空连接级状态） |
关键点是 BACKOFF 状态承担了「清理」职责：所有与连接绑定的状态都在这里重置，避免重连后带着过期状态运行。
Python 实现
import asyncio, json, os, time
from collections import deque
import websockets
URL = "wss://api.tiqex.io/api/v1/market/stream"
API_KEY = os.environ["TIQEX_API_KEY"]
SYMBOLS = ["BTCUSDT", "ETHUSDT", "AAPL"]
SUBSCRIBE_TIMEOUT = 10 # 等待 subscribed 确认帧
IDLE_TIMEOUT = 60 # 静默超时（心跳）
DEDUP_WINDOW_SEC = 5 # 去重窗口
seen = {} # (symbol, at) -> 首次见到的时间
def is_duplicate(tick: dict) -> bool:
"""用 (symbol, at) 作为幂等键，配合滑动时间窗。"""
key = (tick.get("symbol"), tick.get("at"))
now = time.time()
# 清理过期键，重连后无需额外清理逻辑
for k, ts in list(seen.items()):
if now - ts > DEDUP_WINDOW_SEC:
del seen[k]
if key in seen:
return True
seen[key] = now
return False
async def run_once():
async with websockets.connect(
URL, additional_headers={"X-API-Key": API_KEY}, open_timeout=10,
) as ws:
await ws.send(json.dumps({
"action": "subscribe", "market": "crypto", "symbols": SYMBOLS,
}))
# 阶段一：等订阅确认，而不是假设订阅成功
deadline = time.monotonic() + SUBSCRIBE_TIMEOUT
while True:
raw = await asyncio.wait_for(ws.recv(), timeout=SUBSCRIBE_TIMEOUT)
frame = json.loads(raw)
data = frame.get("data") or {}
if not frame.get("succ"):
raise RuntimeError(f"订阅失败 code={data.get('code')} msg={frame.get('msg')}")
if data.get("type") == "subscribed":
print("订阅已确认:", data.get("symbols"))
break
if time.monotonic() > deadline:
raise TimeoutError("等待订阅确认帧超时")
# 阶段二：处理数据帧，带去重与静默超时
while True:
raw = await asyncio.wait_for(ws.recv(), timeout=IDLE_TIMEOUT)
frame = json.loads(raw)
data = frame.get("data") or {}
kind = data.get("type")
if kind == "tick":
tick = data["tick"]
if not is_duplicate(tick):
print(tick["symbol"], tick["last"], tick["at"])
elif kind == "error":
print("流内错误:", data.get("code"), frame.get("msg"))
async def main():
backoff = 1.0
while True:
try:
await run_once()
backoff = 1.0 # 正常结束则重置退避
except Exception as exc: # noqa: BLE001
print(f"连接中断: {exc}")
seen.clear() # 连接级状态在此清理
await asyncio.sleep(backoff)
backoff = min(backoff * 2, 60) # 指数退避，上限 60s
print(f"{backoff:.0f}s 后重连")
asyncio.run(main())
这段代码对应了前面三个陷阱的处理：等确认帧、去重带时间窗、订阅超时与静默超时分离。重连退避上限设为 60 秒，避免在服务端故障时用重连握手持续消耗额度。
如何验证没有重复
改完之后需要可观测的验证手段，而不是靠肉眼看日志：
- 统计唯一键比例：记录收到的总帧数与去重后的唯一
(symbol, at)数量，比值应接近 1；如果显著小于 1，说明去重生效但重复量大，需要检查重连频率。 - 统计重连次数：正常情况下一天应该只有个位数。如果频繁重连，问题通常在心跳设置或网络环境。
- 对账下游指标：如果告警系统出现「同一价格同一时间触发两次」，说明去重没覆盖到下游。
常见问题
为什么重连后要先收到 subscribed 才能处理数据？
因为 send 只表示消息已发出，不代表服务端接受了订阅。只有在收到 type=subscribed 确认帧后，才能确认 market 与 symbols 已生效。跳过这一步会让订阅失败变成静默故障。
去重应该用哪个字段做键？
用 (symbol, at)。tick 帧里的 at 是 RFC3339 时间戳，同一标的同一时刻的 tick 天然唯一，非常适合做幂等键。
重连会消耗额度吗？
会。建连握手计入每日额度与每分钟速率，握手之后的 tick 推送不计费。所以退避策略是必要的——不要用高频重连来「保活」。
静默超时设多少合适？
取决于标的的活跃度。冷门市场或非交易时段可能长时间没有 tick，静默超时设得太短会误判断线并反复重连。建议优先依赖协议层 ping/pong 判断连接存活，把静默超时设得宽松一些（例如 60 秒）。
订阅被拒时应该怎么处理？
按错误码区分：MARKET_NOT_ENABLED 或 NO_SUBSCRIPTION 属于权限问题，重连不会解决，应该立即告警并停止重试；INVALID_SYMBOL 或 UNKNOWN_SYMBOL 属于参数问题，应该修正符号后重新订阅。
下一步
把订阅确认、去重窗口与心跳超时这三件事分开处理，重复数据问题基本就消失了。订阅协议与错误码的完整定义见 WebSocket 流式接口文档；重连与心跳的更多实践可参考 断线重连指南 与 WebSocket 心跳；如果你还在纠结要不要上流式，可以对比 轮询与流式 的差异。