第 17 章 · WebSockets 实时通信

本章目标：用 @app.websocket 建立双向长连接，实现收发循环、连接管理器广播与断开处理，并了解 WebSocket 端点里依赖的用法与限制。

17.1 HTTP 请求 vs WebSocket 连接

HTTP 是一问一答：客户端不发请求，服务端永远没机会说话。聊天室、行情推送、协同编辑这类"服务端要主动推"的场景需要 WebSocket：一次握手升级后，双方在同一条 TCP 长连接上随时互发消息。

FastAPI（Starlette）原生支持 WebSocket。先安装官方示例使用的客户端库（服务端本身不需要额外依赖）：

bash

uv add websockets

17.2 最小回声服务器

python

from fastapi import FastAPI, WebSocket

app = FastAPI()


@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()          # 完成握手，正式建立连接
    while True:                       # 一个连接上可以循环收发任意多条消息
        data = await websocket.receive_text()
        await websocket.send_text(f"Message text was: {data}")
    # 循环退出（客户端断开抛异常）后连接结束

除了文本，还有三组对称方法：send_json/receive_json（自动序列化/反序列化）、send_bytes/receive_bytes（二进制）。

17.3 浏览器端对接

写一个自带测试页面的完整应用：

python

from fastapi import FastAPI, WebSocket
from fastapi.responses import HTMLResponse

app = FastAPI()

html = """
<!DOCTYPE html>
<html>
<body>
  <input id="msg" placeholder="输入消息"/>
  <button onclick="send()">发送</button>
  <ul id="log"></ul>
  <script>
    const ws = new WebSocket("ws://127.0.0.1:8000/ws");
    ws.onmessage = (event) => {           // 收到服务端推送
      document.getElementById("log")
        .insertAdjacentHTML("beforeend", `<li>\${event.data}</li>`);
    };
    function send() {
      ws.send(document.getElementById("msg").value);   // 发给服务端
    }
  </script>
</body>
</html>
"""


@app.get("/")
async def page():
    return HTMLResponse(html)


@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    while True:
        data = await websocket.receive_text()
        await websocket.send_text(f"Echo: {data}")

注意前端 URL 用的是 ws:// 协议（HTTPS 下是 wss://）。打开多个标签页，每个都是独立连接，各自独立收发。

17.4 断开处理与多客户端广播

客户端关闭页面时，阻塞中的 await websocket.receive_text() 会抛出 WebSocketDisconnect 异常——这就是你清理连接的信号。配合一个"连接管理器"类即可实现聊天室广播：

python

from fastapi import FastAPI, WebSocket, WebSocketDisconnect

app = FastAPI()


class ConnectionManager:
    def __init__(self):
        self.active: list[WebSocket] = []

    async def connect(self, ws: WebSocket):
        await ws.accept()
        self.active.append(ws)

    def disconnect(self, ws: WebSocket):
        self.active.remove(ws)

    async def broadcast(self, message: str):
        for conn in self.active:
            await conn.send_text(message)


manager = ConnectionManager()


@app.websocket("/ws/{client_id}")
async def chat(websocket: WebSocket, client_id: int):
    await manager.connect(websocket)
    try:
        while True:
            data = await websocket.receive_text()
            await manager.broadcast(f"Client #{client_id} says: {data}")
    except WebSocketDisconnect:
        manager.disconnect(websocket)
        await manager.broadcast(f"Client #{client_id} left the chat")

生产提示

这个管理器是单进程内存版。多 worker / 多机部署时，各进程的连接列表互相不可见，广播需要借助 Redis Pub/Sub 等外部通道转发；同时给 broadcast 加异常保护，避免某个半死连接拖垮整个循环。

17.5 WebSocket 里能用哪些依赖？

WebSocket 端点支持一部分 FastAPI 设施：Depends、Security、Cookie、Header、Path、Query 都可用，用法和普通端点一致——鉴权通常靠握手请求里的 Cookie 或查询参数完成：

python

from fastapi import Depends, Query, WebSocket, WebSocketException, status


async def check_token(websocket: WebSocket, token: Annotated[str, Query()]):
    if token != "secret-token":
        # WebSocket 场景没有 HTTPException 的意义，要用专用异常
        raise WebSocketException(code=status.WS_1008_POLICY_VIOLATION)
    return token


@app.websocket("/items/{item_id}/ws")
async def item_ws(
    websocket: WebSocket,
    item_id: int,                                    # Path 参数照常解析
    _: str = Depends(check_token),                   # 依赖鉴权
):
    await websocket.accept()
    ...

限制也要记住：WebSocket 没有 HTTP 语义，所以不能抛 HTTPException、不能用返回值定义响应模型，错误用 WebSocketException 并携带 RFC 6455 规定的关闭码（如 WS_1008_POLICY_VIOLATION）；中间件对 WebSocket 也只部分生效。

17.6 本章小结

WebSocket 握手后是双向长连接，accept() 后在 while True 中循环收发；
文本/JSON/二进制各有对称的 send/receive 方法；
客户端断开会让 receive 抛 WebSocketDisconnect，在此做清理和广播告别语；
单进程管理器只能单机广播，分布式部署需要 Redis Pub/Sub 之类的外部总线；
WebSocket 支持 Depends/Query/Cookie 等依赖，但不能用 HTTPException 和响应模型。

🧪 随堂测验

点击你认为正确的选项。答错时会展示正确答案与原因解析。

1. 客户端突然关闭浏览器后，服务端的哪个现象标志着断开？

2. 在 WebSocket 端点中校验失败想主动拒绝连接，正确的做法是？

3. 把第 17.4 节的聊天应用以 --workers 4 启动，广播功能会出现什么问题？

4. 下列哪项在 WebSocket 端点中是不可用的？

🛠️ 动手实践

把 17.4 的聊天室跑起来，开三个标签页验证广播与离开通知；然后给 broadcast 加 try/except，保证一个坏连接不影响其他人。
实现 /ws/count：每秒向所有连接推送一次服务器时间（asyncio.create_task 定时任务），前端实时显示，体会"服务端主动推"。
给聊天 WebSocket 加 token 校验依赖（查询参数传 token），用错误的 token 连接并观察浏览器 Console 里的关闭码是否为 1008。

恭喜！FastAPI 主线教程到此收官，接下来请进入下一章：模板渲染与静态文件。

第 17 章 · WebSockets 实时通信 ​

17.1 HTTP 请求 vs WebSocket 连接 ​

17.2 最小回声服务器 ​

17.3 浏览器端对接 ​

17.4 断开处理与多客户端广播 ​

17.5 WebSocket 里能用哪些依赖？ ​

17.6 本章小结 ​

🧪 随堂测验

🛠️ 动手实践 ​

第 17 章 · WebSockets 实时通信

17.1 HTTP 请求 vs WebSocket 连接

17.2 最小回声服务器

17.3 浏览器端对接

17.4 断开处理与多客户端广播

17.5 WebSocket 里能用哪些依赖？

17.6 本章小结

🛠️ 动手实践