WebSocket爬虫实战：从握手到断线重连的完整指南

1. 为什么我最终选择了WebSocket这条“非主流”爬虫路线

做爬虫这行的朋友都有一个共识：HTTP抓取是基本功，但真正让人头疼的，往往是那些“不按套路出牌”的数据源。我最早接触WebSocket爬虫，是因为一个行情监控的需求——目标站点的数据不是通过传统的XHR轮询返回的，而是走了一条长连接，浏览器和服务器之间保持着一条双向通道，数据像流水一样持续推送过来。用常规的requests加BeautifulSoup那一套，抓到的永远是初始页面，真正的数据一条都拿不到。

这就是WebSocket爬虫要解决的核心问题：当目标数据通过全双工长连接主动推送、而非客户端请求-响应模式传输时，如何稳定地接入这条通道、解析推送帧、并把数据落库或落盘。它适合已经掌握Python基础、写过简单HTTP爬虫、但遇到动态推送类站点就卡壳的开发者。读完这篇，你应该能独立完成一个可运行、可断线重连、可解析二进制帧的WebSocket采集脚本。

先说清楚一件事：WebSocket爬虫和HTTP爬虫不是替代关系，而是互补。HTTP负责拿页面骨架、拿历史数据、拿配置接口；WebSocket负责拿实时流。两者配合，才是一个完整的采集方案。我见过不少新手一上来就想用WebSocket抓所有东西，结果连握手都过不去，这就是没搞清楚适用边界。

从协议层面理解，WebSocket的握手阶段其实是一次特殊的HTTP升级请求（Upgrade: websocket），握手成功后协议切换，之后的通信就不再是HTTP语义了。这个特性决定了：你可以用HTTP的经验去完成握手，但握手之后的一切都要用全新的思路处理。这也是为什么很多HTTP爬虫老手第一次写WebSocket会翻车——他们把长连接当成了“慢一点的HTTP请求”。

我个人的经验是，判断一个站点要不要上WebSocket，看三个信号：一是Network面板里有没有状态码101的请求；二是WS标签页里有没有持续不断的新帧；三是页面数据在你没有任何操作的情况下自动更新。三个信号中两个以上命中，基本就可以确定要上WebSocket方案了。

2. 动手之前必须搞懂的协议细节与工具选型

2.1 握手阶段到底发生了什么

WebSocket的握手本质上是一次HTTP GET请求，但带了几个关键头：Upgrade: websocket、Connection: Upgrade、Sec-WebSocket-Key（一个Base64编码的16字节随机值）、Sec-WebSocket-Version: 13。服务器如果同意升级，会返回101状态码，并在响应头里带上Sec-WebSocket-Accept——这个值是把客户端的Key拼上一个固定的GUID字符串后做SHA-1再Base64得到的。

为什么要讲这个？因为很多站点的握手请求里带了鉴权参数，比如token、cookie、签名、时间戳。你用现成的库直接连，库会帮你生成Key和Accept校验，但你得自己把鉴权参数塞进握手头里。我踩过的坑是：某站点的token放在URL的query里而不是header里，我一开始只往header里塞，握手一直失败，排查了半天才发现问题。

import websocket ws_url = "wss://example.com/stream?token=YOUR_TOKEN&ts=1700000000" ws = websocket.create_connection( ws_url, header={ "Origin": "https://example.com", "User-Agent": "Mozilla/5.0 ..." }, cookie="sessionid=xxxxx" )

上面这段是最朴素的握手方式。注意Origin头，有些站点会校验它，缺失或不对会直接拒绝升级。cookie参数在websocket-client里是单独传的，不是塞进header字典，这点和requests的习惯不一样，容易搞混。

2.2 库的选型：websocket-client、websockets、aiohttp怎么选

这是新手最容易纠结的地方。我把三个主流库的适用场景列个表，你对着自己的需求挑：

我的建议很直接：如果你只是要抓一两个连接的数据，用websocket-client，别折腾异步。异步的收益在多连接场景才体现得出来，单连接用异步纯属给自己找麻烦。我早期为了“显得专业”硬上asyncio，结果一个断线重连的逻辑写了三百行，后来换回同步版本，五十行搞定，稳定性还更好。

当然，如果你要同时监听几十个交易对、几十个频道，那异步是必须的。这时候websockets库的async for message in ws写法会比回调清爽很多。选型的核心判断标准是连接数量和是否需要与其他异步任务协同，而不是哪个库更“高级”。

2.3 数据帧的类型：文本帧、二进制帧、控制帧

WebSocket传输的数据分三类：文本帧（opcode 0x1）、二进制帧（opcode 0x2）、控制帧（ping 0x9、pong 0xA、close 0x8）。文本帧就是UTF-8字符串，直接json.loads就行；二进制帧通常是protobuf、MessagePack或者自定义的紧凑格式，需要按协议解；控制帧是协议层自动处理的，但ping/pong的响应策略会影响连接存活。

我遇到过一个坑：某站点的数据是二进制帧，但前几个字节是自定义头，后面才是protobuf体。我一开始直接拿整个payload去解protobuf，一直报解析错误。后来抓了原始字节dump出来看，才发现前面有4字节的长度字段和2字节的类型字段。遇到二进制帧，第一件事永远是dump原始字节，用十六进制看清楚结构，别急着上解析库。

# 二进制帧处理的基本套路 import struct def parse_binary_frame(data: bytes): # 假设前4字节是长度，接着2字节是消息类型 length = struct.unpack(">I", data[:4])[0] msg_type = struct.unpack(">H", data[4:6])[0] body = data[6:6+length] return msg_type, body

这段代码里的>表示大端序，网络传输默认大端。字节序搞反是二进制解析最常见的错误，症状是数字大得离谱或者小得离谱，遇到这种情况先检查字节序。

3. 一个完整可跑的WebSocket采集脚本是怎么搭起来的

3.1 整体架构：连接层、解析层、存储层分离

我不喜欢把所有逻辑塞在一个文件里，哪怕是个小脚本。WebSocket采集的稳定性问题特别多，分层之后排查起来快得多。我的习惯是分三层：连接层只管连、断、重连；解析层只管把原始帧变成结构化字典；存储层只管写文件或写库。三层之间用队列或者回调传递数据。

这样分的好处是：连接断了不影响解析逻辑的测试，解析逻辑改了不用动连接代码，存储从CSV换成数据库只改一层。我见过太多“一锅炖”的脚本，改一个字段名要全文搜索替换，维护成本极高。

import json import time import websocket from collections import deque class WSCollector: def __init__(self, url, headers=None, cookie=None): self.url = url self.headers = headers or {} self.cookie = cookie self.ws = None self.buffer = deque(maxlen=10000) self.running = False def connect(self): self.ws = websocket.create_connection( self.url, header=self.headers, cookie=self.cookie, timeout=10 ) self.running = True def reconnect(self, max_retry=5): for i in range(max_retry): try: self.connect() return True except Exception as e: wait = min(2 ** i, 30) print(f"重连失败 {i+1}/{max_retry}，{wait}s后重试: {e}") time.sleep(wait) return False

这段是连接层的骨架。注意maxlen=10000的deque，这是给缓冲区设上限，防止解析速度跟不上推送速度导致内存爆掉。这个细节很多人忽略，跑几个小时内存就上G了。

3.2 心跳与断线重连：让脚本能跑一整夜

WebSocket连接不是永久的。服务器会主动断开空闲连接，网络会抖动，代理会超时。一个不能自动重连的采集脚本，实际价值约等于零。重连逻辑要处理三件事：检测断开、退避重试、恢复后补数据。

检测断开最可靠的方式是捕获异常加心跳超时。websocket-client的recv()在连接断开时会抛异常，但如果是静默断开（TCP半开），recv()会一直阻塞。所以必须设timeout，超时后主动发ping，连续几次pong没回来就判定断开。

def heartbeat(self, interval=30): last_pong = time.time() while self.running: try: self.ws.settimeout(interval) msg = self.ws.recv() if msg: self.buffer.append(msg) last_pong = time.time() except websocket.WebSocketTimeoutException: # 超时，主动ping try: self.ws.ping() except Exception: self.running = False except Exception as e: print(f"连接异常: {e}") self.running = False

退避重试的策略我一般用指数退避，但加个上限。第一次等1秒，第二次2秒，第三次4秒，最多等到30秒。为什么要退避？因为如果是服务器端限流导致的断开，你立刻重连只会被再次踢掉，越踢越频繁，最后可能被封IP。退避是给服务器一个“冷静期”。

恢复后补数据这块要看业务。如果是行情类数据，断了就断了，补历史意义不大，直接接着采就行；如果是订单类数据，断线期间的可能需要调HTTP接口补。这个决策取决于数据的时间敏感性和可补性，没有通用答案。

3.3 解析层：从原始帧到结构化数据

解析层的核心是把不同格式的帧统一成字典。文本帧直接json.loads，二进制帧按协议解，解完统一加个时间戳字段。我强烈建议在解析层加一个“原始帧落盘”的开关，调试阶段把原始帧原样写文件，出问题时可以离线复现解析逻辑，不用重新连服务器。

def parse_text_frame(raw: str): try: obj = json.loads(raw) except json.JSONDecodeError: return None return { "ts": time.time(), "type": obj.get("type", "unknown"), "payload": obj.get("data", obj) }

解析层最容易出的问题是字段缺失和类型不一致。同一个频道，有时候推{"price": "100.5"}字符串，有时候推{"price": 100.5}数字。你的解析代码必须对这两种情况都兼容，否则跑着跑着就抛异常。我的做法是统一做类型转换，数字字段一律float()，转换失败就记日志跳过，不让单条脏数据搞崩整个脚本。

3.4 存储层：什么时候写文件，什么时候写库

小规模采集，直接追加写JSONL文件最省事，一行一条，出问题好查。数据量上来了，或者需要按条件查询，就上SQLite或时序数据库。我的经验是：日增数据量在百万条以下，SQLite完全够用，别急着上重型数据库。很多人一上来就搭一套复杂的存储，结果采集脚本本身还没跑稳，本末倒置。

import json def save_jsonl(record, path="data.jsonl"): with open(path, "a", encoding="utf-8") as f: f.write(json.dumps(record, ensure_ascii=False) + "\n")

写文件要注意的是并发写。如果你用了多线程或者异步，多个协程同时写一个文件会串行。解决办法是单开一个写线程，其他线程往队列里丢，写线程从队列取。这个模式叫“单写者模式”，是IO密集场景的经典解法。

4. 实战中那些文档不会告诉你的坑

4.1 握手成功但收不到数据：先查订阅消息

很多WebSocket服务在握手成功后不会自动推数据，需要客户端先发一条订阅消息，告诉服务器“我要哪些频道的数据”。这个订阅消息的格式各站不同，有的是JSON，有的是自定义字符串。如果你连上了但一直没数据，八成是忘了发订阅。

排查方法很简单：在浏览器里打开目标页面，看WS标签页里客户端发出的第一帧是什么，照抄格式就行。注意有些订阅消息里带了签名或时间戳，需要动态生成，不能硬编码。

4.2 被限流或封禁：频率控制与身份伪装

WebSocket虽然是一条长连接，但服务器依然能识别你的行为模式。短时间内频繁重连、订阅大量频道、发送异常消息，都可能触发风控。我的经验是：重连间隔不要低于5秒，订阅频道分批发送、每批之间加延迟，消息发送频率控制在服务器能接受的范围内。

身份伪装方面，Origin、User-Agent、Cookie要和真实浏览器一致。有些站点还会校验Sec-WebSocket-Protocol子协议字段，这个也要照抄。我遇到过一个站点，子协议字段不对直接返回403，但错误信息很模糊，排查了很久。

4.3 内存泄漏：缓冲区不设上限的代价

前面提过deque的maxlen，这里再强调一次。WebSocket是持续推送的，如果你的解析速度慢于推送速度，缓冲区会无限增长。跑一晚上内存爆掉，脚本被系统杀掉，第二天发现数据缺了一大段。设maxlen是兜底，更好的做法是监控缓冲区长度，超过阈值就告警或降级处理。

4.4 时间戳与时区：数据对不上的元凶

推送数据里的时间戳可能是毫秒、微秒、秒，也可能是本地时间、UTC、带时区偏移。存储时统一转成UTC毫秒时间戳，展示时再转本地。我踩过的坑是：采集端存的是本地时间，分析端按UTC解读，结果所有数据偏移了8小时，排查了半天才发现是时区问题。

4.5 日志与监控：出问题时你能看到什么

采集脚本一定要打日志，而且要分级。连接建立、断开、重连打INFO；解析失败、字段缺失打WARNING；连续重连失败、缓冲区溢出打ERROR。日志里带上时间戳和连接标识，多连接场景下能区分是哪个连接出的问题。

监控方面，至少监控三个指标：连接存活状态、单位时间接收帧数、缓冲区长度。这三个指标异常，基本能覆盖大部分故障场景。我一般用一个简单的定时任务，每分钟把这三个指标打到日志里，出问题时看日志就能定位。

5. 从单连接到多连接：规模上去之后怎么改

单连接跑稳之后，需求往往会变成“同时监听多个频道”。这时候有两条路：多线程多连接，或者单连接多订阅。优先选单连接多订阅，因为连接数越少，被风控的概率越低，资源占用也越小。只有当单连接订阅数量有上限时，才考虑多连接。

多连接的管理我用一个连接池的思路：维护一个连接列表，每个连接有独立的状态和缓冲区，主线程定期检查各连接健康度，不健康的触发重连。连接之间用队列隔离，避免一个连接卡住影响其他连接。

class ConnectionPool: def __init__(self, urls): self.conns = [WSCollector(u) for u in urls] def start_all(self): for c in self.conns: c.connect() def health_check(self): for c in self.conns: if not c.running: c.reconnect()

规模再大，就要考虑分布式了。多台机器各采一部分，汇总到一个消息队列，再由消费端统一落库。但这是另一个话题了，单机场景下上面的方案足够用。

6. 几个我实际踩过的坑和对应的解法

第一个坑是SSL证书校验。有些站点的证书链不完整，websocket-client默认会校验失败。临时方案是sslopt={"cert_reqs": ssl.CERT_NONE}，但生产环境不建议关校验，正确做法是把缺失的中间证书补上或者更新本地证书库。

第二个坑是代理配置。如果你的网络环境需要走代理，websocket-client的代理配置和requests不一样，要用http_proxy_host、http_proxy_port参数，而且只支持HTTP代理。这个在文档里写得很隐蔽，我第一次配的时候找了半天。

第三个坑是消息分片。WebSocket协议允许一个大消息分成多个帧传输，websocket-client默认会自动合并，但有些库不会。如果你收到的消息总是“半截”，检查一下是不是分片没合并。websocket-client的recv()是合并后的，recv_data()是原始帧，别用错。

第四个坑是关闭握手。主动关闭连接时要发close帧并等对方回应，直接ws.close()有时候会抛异常。稳妥的做法是先发close，等一秒再强制关闭。这个细节在长时间运行的脚本里很重要，不然关闭时的异常会污染日志。

7. 关于合规与边界的一点个人看法

做采集这行，技术之外还有一条线要自己心里有数。我只采集公开的、不涉及个人隐私的数据，采集频率控制在不对目标服务造成压力的范围内，遇到明确的反爬声明就换方案或者放弃。这不是唱高调，而是长期做这行的基本素养——把对方服务搞挂了，你自己的数据源也就没了。

另外，采集到的数据怎么用、存多久、要不要脱敏，这些在动手之前就该想清楚。我见过太多人脚本写得飞起，数据存了一堆，最后不知道怎么处理，既占资源又有隐患。技术方案里应该包含数据生命周期的管理，而不是只管采不管收尾。

WebSocket爬虫这个方向，入门门槛比HTTP爬虫高一点，但一旦跑通，能拿到的数据质量和实时性是HTTP比不了的。我个人的体会是：先把单连接跑稳，把重连和解析这两个基本功练扎实，再考虑规模和并发。很多新手一上来就追求“高并发架构”，结果连一个连接都跑不过夜，这是本末倒置。把简单的事情做到极致稳定，比堆砌复杂的技术栈有价值得多。