3、Python WebSocket客户端实战：使用websockets库建立连接、发送订阅消息、接收行情数据

好，咱们直接进入正题。前面两章我们把WebSocket协议和行情数据的格式讲清楚了，这一章就是动手干活的时刻。我个人习惯用Python的websockets库来做客户端，原因很简单：它原生支持asyncio，代码写起来干净利落，不像某些库回调套回调，调试起来头疼。

3.1 环境准备与库安装

先装库，这个没啥好说的：

pip install websockets

嗯，就这一行。如果你用的是Python 3.7以上版本，标准库里的asyncio已经够用了，不需要额外装。我在项目中遇到过有人装了websocket-client（注意少了个s），结果API完全不同，折腾了半天才发现装错了库。你想想看，这种低级错误多耽误事。

3.2 建立连接——别小看这一步

连接WebSocket服务器，说白了就是创建一个长连接。代码其实很简单：

import asyncio
import websockets

async def connect():
    uri = "wss://api.example.com/market"
    async with websockets.connect(uri) as websocket:
        print("连接成功")
        # 后续操作在这里写

这里有个坑，我曾经踩过：生产环境一定要加超时和重试机制。你想想看，行情服务器偶尔会挂，或者网络抖动一下，连接就断了。不加重试的话，你的策略可能就傻等在那里，错过一波行情。

带重试的写法是这样的：

async def connect_with_retry(uri, max_retries=5):
    for i in range(max_retries):
        try:
            async with websockets.connect(uri, ping_interval=20) as ws:
                print(f"连接成功，第{i+1}次尝试")
                return ws
        except Exception as e:
            print(f"连接失败: {e}，5秒后重试")
            await asyncio.sleep(5)
    raise Exception("连接失败，已达最大重试次数")

3.3 发送订阅消息——格式要对

连接建立之后，下一步就是告诉服务器：我要哪些数据。不同的交易所，订阅消息的格式不一样。以常见的JSON格式为例：

async def subscribe(websocket, symbols):
    sub_msg = {
        "op": "subscribe",
        "args": [f"trade.{s}" for s in symbols]
    }
    await websocket.send(json.dumps(sub_msg))
    print(f"已订阅: {symbols}")

这里要注意的是：订阅消息一定要等连接完全建立后再发。我见过有人刚connect完就send，结果消息丢了，数据一直收不到。正确的做法是等服务器返回一个"welcome"或者"connected"的确认消息后再发订阅。

3.4 接收行情数据——循环与解析

接收数据是个持续的过程，需要用循环来不断读取。这里有个关键点：不要阻塞事件循环。如果你在接收循环里做耗时操作，比如写数据库、计算指标，那后面的数据就会堆积，延迟越来越高。

async def receive_market_data(websocket):
    async for message in websocket:
        data = json.loads(message)
        # 这里只做轻量处理
        await process_data(data)

那耗时操作怎么办？我的做法是：用队列解耦。接收线程只管把数据丢进队列，另一个消费者线程慢慢处理。

import asyncio
from collections import deque

data_queue = asyncio.Queue()

async def receive(websocket):
    async for msg in websocket:
        await data_queue.put(json.loads(msg))

async def consumer():
    while True:
        data = await data_queue.get()
        # 在这里做耗时操作，比如存数据库
        await save_to_db(data)

你看，这样接收和消费就分开了，互不干扰。我在做高频行情接入时，这个模式帮了大忙。

3.5 完整示例代码

把上面几块拼起来，就是一个完整的客户端了：

import asyncio
import json
import websockets

class MarketDataClient:
    def __init__(self, uri, symbols):
        self.uri = uri
        self.symbols = symbols
        self.queue = asyncio.Queue()
    
    async def connect(self):
        async with websockets.connect(self.uri) as ws:
            await self._subscribe(ws)
            await asyncio.gather(
                self._receive(ws),
                self._consumer()
            )
    
    async def _subscribe(self, ws):
        msg = {"op": "subscribe", "args": [f"trade.{s}" for s in self.symbols]}
        await ws.send(json.dumps(msg))
    
    async def _receive(self, ws):
        async for msg in ws:
            await self.queue.put(json.loads(msg))
    
    async def _consumer(self):
        while True:
            data = await self.queue.get()
            # 处理数据，比如打印
            print(f"收到行情: {data['symbol']} 价格: {data['price']}")

# 运行
async def main():
    client = MarketDataClient("wss://api.example.com/market", ["BTCUSDT", "ETHUSDT"])
    await client.connect()

asyncio.run(main())

3.6 核心逻辑流程图

下面这张图把整个流程串起来了，你一看就明白：

3.7 常见问题与调试技巧

最后分享几个我在实战中遇到的坑：

问题	现象	解决方案
连接超时	程序卡在connect()不动	设置connect_timeout参数，比如5秒
数据断流	收不到行情，但连接没断开	加ping/pong心跳，超时自动重连
JSON解析错误	抛出json.decoder.JSONDecodeError	用try/except包裹，记录原始数据
内存泄漏	程序运行越久，内存占用越大	检查队列是否无限增长，设置最大长度

嗯，这一章的内容就这些。说白了，WebSocket客户端就是个"连接-订阅-接收"的循环，但细节决定成败。你按照上面的代码和注意事项去写，基本不会出大问题。

---

公众号:蓝海数据掘金营，微信deep3321