4、WebSocket客户端基础：使用websockets库建立长连接，心跳机制（Ping/Pong），自动重连策略设计

好，咱们今天聊聊WebSocket客户端。说实话，在Tick数据采集系统里，这一步是真正的「命门」。你想想看，行情数据是实时推送的，如果连接断了，数据就断了，那整个系统就废了。所以，怎么建立一个靠谱的长连接，怎么保证它不掉线，掉线了怎么快速恢复——这些就是本章要解决的核心问题。

4.1 为什么选websockets库？

Python里做WebSocket的库不少，比如websocket-client、aiohttp自带的WebSocket支持。但我个人习惯用websockets这个库。为什么？

  • 它是纯异步的，基于asyncio，性能非常好
  • API设计简洁，上手快
  • 对Ping/Pong的支持很原生，不需要自己造轮子

安装也很简单：

pip install websockets

嗯，这里要注意：如果你用的是Python 3.7以下版本，可能需要额外装async-timeout。不过现在应该没人用那么老的版本了吧？

4.2 建立长连接——基础版

先来个最简单的例子。咱们连一个模拟的行情服务器：

import asyncio
import websockets

async def connect():
    uri = "ws://localhost:8765"
    async with websockets.connect(uri) as websocket:
        print("连接成功！")
        # 接收一条消息
        msg = await websocket.recv()
        print(f"收到: {msg}")

asyncio.run(connect())

这段代码看着简单，但有个大问题——它只接收一条消息就退出了。在实际的Tick数据采集场景里，我们需要持续接收数据。所以一般会写成循环：

async def connect():
    uri = "ws://localhost:8765"
    async with websockets.connect(uri) as websocket:
        print("连接成功！")
        async for message in websocket:
            # 处理Tick数据
            process_tick(message)

我在项目中遇到过一个问题：如果服务器突然断开，这个async for循环会抛出websockets.exceptions.ConnectionClosed异常。如果不处理，程序就挂了。所以，咱们得加异常处理。

4.3 心跳机制——Ping/Pong

为什么要心跳？说白了就是「确认对方还活着」。TCP本身有Keep-Alive机制，但时间间隔太长（默认2小时），对于Tick数据这种高频场景来说，太慢了。

websockets库内置了Ping/Pong支持。你可以在连接时设置：

async def connect_with_heartbeat():
    uri = "ws://localhost:8765"
    async with websockets.connect(
        uri,
        ping_interval=20,   # 每20秒发一次Ping
        ping_timeout=10     # 10秒内没收到Pong就认为断开
    ) as websocket:
        print("连接成功，心跳已启动")
        async for message in websocket:
            process_tick(message)

这里有个小细节：ping_interval和ping_timeout的单位是秒。我个人建议：

场景 ping_interval ping_timeout
高频行情（毫秒级） 10-15秒 5-8秒
普通行情（秒级） 20-30秒 10-15秒
低频数据（分钟级） 30-60秒 15-20秒

我曾经把ping_interval设成5秒，结果服务器以为我在攻击它，直接把我IP封了……所以，别太频繁。

4.4 自动重连策略设计

这才是本章的重头戏。自动重连，说白了就是「断了就重连，直到连上为止」。但怎么连？连不上怎么办？一直重连会不会把服务器搞崩？

我设计过一套重连策略，核心思路是：指数退避 + 最大重试次数。

指数退避：每次重连的等待时间翻倍，比如第一次等1秒，第二次等2秒，第三次等4秒……直到达到上限。

来看代码：

import asyncio
import websockets
import logging

logger = logging.getLogger(__name__)

class TickWebSocketClient:
    def __init__(self, uri, max_retries=10, base_delay=1, max_delay=60):
        self.uri = uri
        self.max_retries = max_retries
        self.base_delay = base_delay
        self.max_delay = max_delay
        self.retry_count = 0

    async def connect(self):
        while self.retry_count < self.max_retries:
            try:
                async with websockets.connect(
                    self.uri,
                    ping_interval=20,
                    ping_timeout=10
                ) as websocket:
                    logger.info(f"连接成功，重试次数: {self.retry_count}")
                    self.retry_count = 0  # 重置重试计数
                    async for message in websocket:
                        self.process_tick(message)

            except (websockets.exceptions.ConnectionClosed,
                    websockets.exceptions.WebSocketException) as e:
                self.retry_count += 1
                delay = min(
                    self.base_delay * (2 ** (self.retry_count - 1)),
                    self.max_delay
                )
                logger.warning(
                    f"连接断开，{delay}秒后重试 "
                    f"(第{self.retry_count}次)"
                )
                await asyncio.sleep(delay)

        logger.error("达到最大重试次数，放弃连接")

    def process_tick(self, message):
        # 处理Tick数据的逻辑
        pass
小技巧：重连成功后，最好把retry_count重置为0。这样下次断开时，又会从1秒开始等待，而不是继续累加。

4.5 核心逻辑流程图

下面这张图，把整个WebSocket客户端的核心逻辑串起来了。从建立连接到心跳维持，再到断开重连，一目了然。

WebSocket客户端核心逻辑流程图 开始连接 建立WebSocket连接 启动心跳(Ping/Pong) 接收Tick数据 连接是否断开？ 否 是 指数退避重连 正常流程 异常/重连流程 心跳维持

4.6 避坑指南

我曾经踩过的坑：
  • 重连太频繁：有一次我把重连间隔设成0.5秒，结果服务器直接把我IP拉黑了。后来改成指数退避，再也没出过问题。
  • 忘记重置重试计数：如果不重置，重连成功后下次断开会从很大的延迟开始，导致恢复时间变长。
  • 忽略Ping超时：有些服务器不会主动发Pong，如果ping_timeout设得太短，会导致频繁误判断开。
我的个人建议：
  • 生产环境中，最好把重连日志单独输出到一个文件，方便排查问题
  • 可以加一个「健康检查」接口，手动触发重连，方便运维
  • 如果连接频繁断开，先检查网络，别急着改代码

4.7 完整示例

最后，给一个可以直接运行的完整示例。这个版本加了日志、异常处理、优雅退出：

import asyncio
import websockets
import logging
import signal

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("TickClient")

class TickClient:
    def __init__(self, uri):
        self.uri = uri
        self.running = True

    async def run(self):
        while self.running:
            try:
                async with websockets.connect(
                    self.uri,
                    ping_interval=20,
                    ping_timeout=10
                ) as ws:
                    logger.info("连接成功")
                    async for msg in ws:
                        logger.info(f"收到Tick: {msg[:50]}...")
            except Exception as e:
                logger.error(f"连接异常: {e}")
                if self.running:
                    await asyncio.sleep(5)

    def stop(self):
        self.running = False

async def main():
    client = TickClient("ws://localhost:8765")
    loop = asyncio.get_running_loop()
    loop.add_signal_handler(signal.SIGINT, client.stop)
    loop.add_signal_handler(signal.SIGTERM, client.stop)
    await client.run()

if __name__ == "__main__":
    asyncio.run(main())

嗯，到这里，WebSocket客户端的基础就讲完了。记住三个核心点：长连接、心跳保活、智能重连。把这三点做好了，你的Tick数据采集系统就稳了一大半。

公众号:蓝海资料掘金营，微信deep3321

← 上一章 📖 返回目录 下一章 →