#020 Python 实时通信别再全靠轮询：websockets把双向连接写成了正常代码

一句话定位：websockets 是 Python 里一套很克制、也很靠谱的 WebSocket 实现。它不想把自己做成“大而全的网络框架”，而是专注把长连接、双向消息、保活、背压、广播、自动重连这些实时通信里的脏活累活处理好。

1. 为什么这篇值得写？

很多 Python 项目一开始都只有普通 HTTP：

• 调一个接口，拿一次结果；
• 提交一个任务，过几秒再查状态；
• 想看进度，就不断轮询；
• 想做“实时更新”，就继续加刷新频率。

前期当然能跑。

但只要你的系统开始出现下面这些需求，轮询就会越来越别扭：

• 任务进度要持续推送；
• 日志要实时刷到前端或终端；
• 服务端要主动通知客户端，而不是等客户端来问；
• 一个动作触发后，要立即同步到多个观察者；
• 连接需要长期保持，断了还得尽快发现。

这时候问题就不再是“HTTP 能不能凑合干”，而是：

你到底需不需要一条真正持续存在的双向通道。

websockets 的价值就在这里。

它把很多人脑子里“实时通信很复杂”的那一坨，尽量压缩成了几件直观的事：

• 服务端 serve(...)；
• 客户端 connect(...)；
• 收消息 await ws.recv()；
• 发消息 await ws.send(...)；
• 长连接保活、关闭握手、超时与队列由库来兜底。

如果你最近在做这些东西，websockets 值得认真补上：

• AI / 自动化任务进度推送；
• 实时日志面板；
• 内部控制台的状态同步；
• 设备遥测、心跳、控制指令；
• 多客户端共享同一个实时状态视图。

2. 先看图：websockets真正改变了什么？

从当前安装包元数据可以直接确认：本文写作时可用的 websockets 版本为 16.0，要求 Python >= 3.10。官方在 README 里的定位也很明确：它是一个 for building WebSocket servers and clients in Python 的库，强调 correctness、simplicity、robustness、performance。

这几个词不是营销话术，基本就是它的产品哲学：

1. Correctness
   ：协议实现要靠谱；
2. Simplicity
   ：核心 API 尽量压缩到 recv() / send() 这个级别；
3. Robustness
   ：要能处理真实网络环境里的连接问题；
4. Performance
   ：内存和收发路径不能太糟。

真正重要的变化，其实是这件事：

3.websockets是什么？

如果只用一句话概括：

websockets = 一套专注于 WebSocket 协议本身的 Python 实现，默认主打 asyncio，也提供 threading 和 Sans-I/O 版本。

官方 README 里点得很直白：

• 默认实现建立在 asyncio 之上；
• 还提供基于 threading 的实现；
• 也提供 Sans-I/O 实现，方便你接入更底层或更特殊的网络栈。

这意味着它不是只能服务某一种写法：

但它同样有明确边界。

官方 README 也明确提醒了：websockets 的目标是把 RFC 6455 WebSocket 协议 和压缩扩展实现好，HTTP 支持只有最小必要程度，甚至文档里直接说如果你要做“HTTP + WebSocket 混合服务”，更适合看基于它构建的服务器，比如 uvicorn、Sanic 这类框架。

这个边界非常重要。

很多人一上来就容易误判：

• 以为它是“FastAPI 的替代品”；
• 以为它负责模板、路由、REST、鉴权全家桶；
• 以为它应该顺手把整站都做了。

其实不是。

它更像一把很锋利的专用刀：

• 专心做双向长连接；
• 把消息收发和连接生命周期处理顺；
• 别的 HTTP 事情，交给更适合的层。

4. 什么时候该用它，什么时候别硬上？

先看判断表：

我的判断是：

• 如果你只是“提交任务后过一会再查结果”，未必需要 WebSocket；
• 如果你已经需要秒级甚至亚秒级更新，并且后端要主动推消息，那就该考虑它；
• 如果你需要的是持续会话，而不是“请求-响应”式交互，websockets 会比你硬拗 HTTP 自然得多；
• 如果你的业务核心其实是 HTTP，只是顺带加一点实时能力，那更适合把 WebSocket 当成旁路，而不是把整个系统都改写成纯 WebSocket 服务。

一句话：

websockets 适合做实时通道，不适合独自扛完整 Web 应用。

5. 3 分钟上手：先写一个最小可用回声服务

先安装：

pip install websockets

很多旧教程还在写顶层 websockets.serve(...)、websockets.connect(...)。当前官方 Quick examples 已经明显更偏向下面这种命名空间写法：

import asyncio
from websockets.asyncio.server import serve


async def echo(websocket):

    async for message in websocket:
        await websocket.send(f"echo: {message}")


async def main() -> None:
    async with serve(echo, "127.0.0.1", 8765):
        await asyncio.Future()


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

再写一个客户端：

import asyncio
from websockets.asyncio.client import connect


async def main() -> None:
    async with connect("ws://127.0.0.1:8765") as websocket:
        await websocket.send("hello")
        print(await websocket.recv())


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

这两段代码的价值不是“能不能做个 demo”，而是它把心智模型讲得很清楚：

• 服务端收到一条连接，就把它交给一个协程处理；
• 客户端拿到连接对象后，就是正常地 send() / recv()；
• async with 负责把连接生命周期包住；
• async for message in websocket 说明消息流本身就是一个异步迭代器。

我这里还用当前安装版本实际跑过一遍最小回声示例，serve(...) + connect(...) + send/recv 这套写法在 websockets 16.0 下是通的。

6. 真正让websockets值钱的 4 个能力

6.1serve()的心智模型非常顺：一条连接就是一个协程

从当前版本的签名可以直接看到，serve(...) 最核心的参数还是一个 handler：

from websockets.asyncio.server import serve


async def handler(websocket):
    async for message in websocket:
        await websocket.send(message)

这个模型很值钱，因为它让你写实时服务时不必先去理解一大坨回调。

你可以把它理解成：

• 每来一个客户端；
• 建立一个 ServerConnection；
• 完成 opening handshake；
• 然后把这个连接交给你的处理协程；
• 协程结束后，再由库完成 closing handshake。

官方 serve() 的文档字符串里也明确写了这套生命周期。

它带来的最大好处就是：连接生命周期、协议握手和你的业务处理，被切成了相对清晰的几层。

6.2connect()不只是连一次：上下文管理和自动重连都给了

connect(...) 也不是“连上就完”的小工具。

当前版本文档字符串里明确提到两种用法：

1. 作为异步上下文管理器：

async with connect("ws://127.0.0.1:8765") as websocket:
    await websocket.send("hello")

1. 作为无限异步迭代器，在错误发生时自动重连：

import websockets
from websockets.asyncio.client import connect


async def listen_forever() -> None:
    async for websocket in connect("ws://127.0.0.1:8765/stream"):
        try:
            async for message in websocket:
                print(message)
        except websockets.exceptions.ConnectionClosed:
            continue

官方说明里还点到了：

• 遇到瞬时错误会按指数退避重试；
• 遇到致命错误则会直接抛异常；
• 每一轮循环结束时连接会自动关闭。

这对真实项目特别有用。

因为现实世界的连接并不会永远稳定：

• 网络会抖；
• 代理会断；
• 服务端会重启；
• 移动端和弱网环境更是家常便饭。

如果客户端重连机制要你自己手搓一整套，工程体验会立刻变差很多。

6.3 它不是玩具：保活、超时、队列和背压都在考虑范围内

只看 serve(...) 和 connect(...) 的签名，你就能感觉到它不是只为 demo 准备的。

当前版本里你能直接看到这些参数：

• ping_interval=20
• ping_timeout=20
• close_timeout=10
• max_size=1048576
• max_queue=16
• write_limit=32768
• origins=...
• subprotocols=...

这些参数背后分别在处理几件现实问题：

官方 keepalive 文档也强调了一点：WebSocket 本来就是长连接，而 HTTP/1.1 基础设施往往并不天然欢迎“长时间空闲但不断着”的连接。

文档里直接提到，很多代理会在 30~120 秒 内关闭空闲连接；也就是说，如果你完全不做保活，你以为自己“连着”，中间网络设备未必同意。

所以别把 ping_interval 和 ping_timeout 看成无关紧要的小参数。

它们关系到的是：

你的实时系统到底能不能尽快发现“其实早就断了”。

6.4 广播能力很实用，但你得知道它的边界

很多实时系统最终都会落到一个经典需求：

• 一个任务状态更新了；
• 把这条消息推给所有订阅者；
• 或者推给某一组当前在线的观察者。

websockets 当前版本直接提供了 broadcast(...)：

from websockets.asyncio.server import ServerConnection, broadcast

CLIENTS: set[ServerConnection] = set()


def push_to_all(message: str) -> None:
    if CLIENTS:
        broadcast(CLIENTS, message)

这很方便。

我这里也实际跑过一个最小广播例子：维护一个连接集合，然后 broadcast(CLIENTS, "tick")，两个客户端都能正常收到消息。

但官方文档字符串同样把边界写得很明白：

• broadcast() 是同步地往所有连接推消息；
• 没有 backpressure
  ；
• 如果你广播得太快，慢客户端的写缓冲会不断堆积；
• 文档明确建议保持较低的 ping_interval / ping_timeout，让慢连接尽快超时退出，避免内存占用失控。

这个提醒非常重要。

也就是说：

• 低频状态广播
  ，很好用；
• 高频大消息广播
  ，要谨慎；
• 真到高吞吐阶段，通常要自己做更细的分组、限流、队列或慢客户端剔除策略。

7. 实战：给播客转码任务做一个实时进度通道

假设你现在有一个播客处理流水线：

• 用户上传音频；
• 后台开始转码、降噪、切片、生成字幕；
• 前端或终端控制台想实时看到每个任务走到哪一步。

这种场景如果继续靠轮询，通常会有几个问题：

• 请求很多；
• 进度刷新延迟大；
• 日志和状态拆成很多接口；
• 同时打开多个观察端时，还得重复查同一批数据。

更顺的方式，是把“任务进度变化”直接推给在线观察者。

7.1 服务端：维护连接集合并推送 JSON 事件

import asyncio
import json
from datetime import datetime

from websockets.asyncio.server import ServerConnection, broadcast, serve

CLIENTS: set[ServerConnection] = set()


def make_event(job_id: str, step: str, progress: int) -> str:
    return json.dumps(
        {
            "type": "job.progress",
            "job_id": job_id,
            "step": step,
            "progress": progress,
            "ts": datetime.now().isoformat(timespec="seconds"),
        },
        ensure_ascii=False,
    )


async def handler(websocket: ServerConnection) -> None:
    CLIENTS.add(websocket)
    try:
        await websocket.send(
            json.dumps(
                {
                    "type": "system.hello",
                    "online_clients": len(CLIENTS),
                },
                ensure_ascii=False,
            )
        )
        async for _ in websocket:
            # 这里可以接收客户端命令，例如暂停、继续、重试
            pass
    finally:
        CLIENTS.discard(websocket)


async def fake_job_stream() -> None:
    for progress, step in [(5, "queued"), (20, "transcoding"), (55, "denoise"), (80, "subtitle"), (100, "done")]:
        await asyncio.sleep(1)
        if CLIENTS:
            broadcast(CLIENTS, make_event("podcast-20260406", step, progress))


async def main() -> None:
    async with serve(handler, "127.0.0.1", 8765, ping_interval=20, ping_timeout=20):
        while True:
            await fake_job_stream()
            await asyncio.sleep(2)


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

这个骨架有几个点值得注意：

• 连接集合 CLIENTS 只负责“谁在线”；
• 真实业务状态仍然应该来自任务系统本身，而不是只存在于 WebSocket 层；
• 推送的单位不是“字符串随便拼”，而是结构化事件；
• broadcast() 很适合这种低频、轻量、多人观察同一状态的场景。

上面这张图真正想表达的是：WebSocket 最值钱的不是“连接能不断着”，而是它把任务事件、连接管理、保活检测和客户端展示串成了一条稳定的实时链路。

7.2 客户端：把连接当成持续事件流处理

import asyncio
import json
import websockets
from websockets.asyncio.client import connect


async def listen() -> None:
    async for websocket in connect("ws://127.0.0.1:8765"):
        try:
            async for raw in websocket:
                event = json.loads(raw)
                print(event)
        except websockets.exceptions.ConnectionClosed:
            continue


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

这段代码很适合做：

• 浏览器页面背后的 Python 中转层；
• 终端面板的数据源；
• AI 任务调度器的状态监听器；
• 本地自动化工具的“实时观察器”。

它背后的思路也很简单：

不要把 WebSocket 当成“某次特殊请求”，而要把它当成“持续到来的事件流”。

8. 生产里怎么把它用稳？

8.1 别拿它硬扛所有 HTTP 需求

这是最容易犯的错误。

websockets 非常适合做实时消息通道，但它不是拿来承包：

• 完整路由系统；
• REST API；
• 模板渲染；
• 文件上传下载；
• 整站鉴权体系。

如果你的项目本来就是 Web 服务，常见的更稳组合其实是：

• HTTP 层：FastAPI / Starlette / Sanic / Django 等；
• 实时层：WebSocket；
• 后台任务层：Celery / asyncio task / queue；
• 持久化层：Redis / DB / 对象存储。

websockets 可以单独用，也可以作为更大系统中的“实时管道”存在。

8.2 消息别乱发字符串，尽早定义事件协议

很多 demo 一开始都这么写：

await websocket.send("progress=40")

演示可以，工程里不建议。

更稳的方式几乎总是：

{
  "type": "job.progress",
  "job_id": "podcast-20260406",
  "progress": 40,
  "step": "transcoding"
}

至少要明确：

• type
• id
• 时间戳
• 业务字段
• 是否可重放 / 可幂等

否则系统一复杂，你很快就会进入“前后端都在猜你这串文本到底什么意思”的状态。

8.3 广播要克制，慢客户端要尽早剔除

broadcast() 好用，但官方已经提醒了它没有背压。

所以生产里最好遵守几个原则：

• 不要把高频日志原封不动广播给所有人；
• 同类事件可以合并后再推；
• 只推关键状态变化，而不是每个内部细节；
• 对慢客户端设置超时和剔除策略；
• 必要时按房间、任务、租户做分组广播。

能用 1 条消息表达的状态，不要拆成 10 条碎片消息乱飞。

8.4ping_interval/ping_timeout不要当默认摆设

官方 keepalive 文档明确指出，很多中间网络设施会在 30~120 秒 关闭空闲连接。

所以生产里至少要想清楚：

• 你的连接多久没消息算异常？
• 弱网环境能接受多高的误判？
• 多久没 pong 就该认为这条连接已经死了？
• 断线后客户端是否自动重连？

如果这些问题完全不设计，系统上线后最常见的现象就是：

用户以为页面还连着，服务端以为连接还活着，但中间网络早就把它掐掉了。

8.5 从第一天就考虑断线重连和幂等

只要是长连接，断线就不是意外，而是常态。

因此你要尽量让这些事情从第一天就有设计：

• 客户端重连后如何恢复订阅；
• 任务状态是否可重发；
• 相同事件重复到达时能否幂等处理；
• 服务端重启后客户端能否平滑接回；
• 是否需要“最后一次状态快照 + 后续增量事件”的组合。

别等到线上出现“刷新一下页面进度没了”“断网再连回来就卡死”才想起这些。

9. 最后总结

websockets 最值得学的，不是“我也能写个 echo demo 了”，而是下面这套判断：

1. 实时问题应该用实时通道解决，不要一直拿轮询硬扛；
2. WebSocket 的本质不是前端小技巧，而是后端主动推送能力；
3. websockets 的价值在于它把协议细节、连接生命周期、保活和收发模型整理得很顺；
4. 它很适合做实时层，但不要拿它替代整个 Web 栈；
5. 一旦进入生产，保活、背压、广播频率、重连和幂等必须一起考虑。

如果你最近正在做：

• AI 任务进度推送；
• 日志流展示；
• 设备状态同步；
• 内部控制台实时刷新；
• 多客户端共享同一个任务状态；

那 websockets 很可能就是那块把系统体验从“能用”拉到“顺手”的关键拼图。