FastAPI-WebSocket、测试与部署

这篇讲什么

• 三块内容：WebSocket（含鉴权、广播、关闭码、--ws-max-size）、测试（TestClient 与 httpx.AsyncClient 两条路）、部署（uvicorn / gunicorn / nginx / Docker，含代理头与并发上限的实测坑）。
• 部署侧最值得记的一条实测结论：--limit-concurrency 是按「连接」算的，空闲的 keep-alive 连接也占额度——3 条闲置连接就能把第 4 个请求顶成 503，而且它是硬拒、不排队。
• WebSocket 走 nginx 必须显式转发 Upgrade/Connection，实测配了是 101、没配是 404。

WebSocket 路由

• 最小可用的房间广播：

  main.py

  from fastapi import FastAPI, Query, WebSocket, WebSocketDisconnect, status
  
  app = FastAPI()
  rooms: dict[str, set[WebSocket]] = {}
  
  
  @app.websocket("/ws/{room}")
  async def ws_room(websocket: WebSocket, room: str,
                    token: str | None = Query(default=None)):
      if token != "ok":
          await websocket.close(code=status.WS_1008_POLICY_VIOLATION)
          return
      await websocket.accept()
      rooms.setdefault(room, set()).add(websocket)
      try:
          while True:
              text = await websocket.receive_text()
              for other in list(rooms[room]):
                  await other.send_text("echo:" + text)
      except WebSocketDisconnect:
          rooms[room].discard(websocket)

  • 注意顺序：鉴权不通过要在 accept() 之前 close()，否则连接已经建立、客户端以为成功了。
  • 房间成员用 set 而不是 list，discard 不会因为「不在这个集合里」抛 ValueError（这条和 Tornado 那篇在线聊天 的结论一致）。

• 实测收发与广播（TestClient 自带 websocket_connect）：

  test_ws.py

  from fastapi.testclient import TestClient
  
  from main import app
  
  client = TestClient(app)
  with client.websocket_connect("/ws/room1?token=ok") as ws:
      ws.send_text("hi")
      print(ws.receive_text())

  广播回来: echo:hi
  两个人进同一个房间，一人发消息 -> 两人都收到 ['echo:hello-two', 'echo:hello-two']
  第二个人断开后，第一人再发 -> echo:after-leave      # 断开的成员已被摘掉，不会报错
  token 不对 -> WebSocketDisconnect                    # 连接被拒，客户端侧表现为断开

  • receive_text() / receive_json() / send_text() / send_json() / send_bytes() 是客户端侧的方法；服务端对应 websocket.receive_* / send_*。
  • 服务端里抛未捕获异常时，TestClient 会把异常原样抛回测试代码（实测 RuntimeError: 服务端里炸了），真实服务里则是记日志并断开连接。

• 关闭码与 WebSocketDisconnect 实测：

  close_codes.py

  with client.websocket_connect("/ws") as ws:
      ws.send_json({"n": 1})
      ws.close(code=4001, reason="客户端主动关")

  正常结束      -> WS 断开 code=1000 reason=''
  客户端带码关闭 -> WS 断开 code=4001 reason='客户端主动关'
  服务端 receive_json 收到非 JSON -> JSONDecodeError: Expecting value: line 1 column 1

  • WebSocketDisconnect 有 code 和 reason，用它区分「用户主动关」和「网络掉」很实用。
  • receive_json() 不做兜底，脏数据直接抛 JSONDecodeError；对外服务要在 while 外面套 try/except 再回一条错误消息，否则整个连接被异常打断。

• 子协议（Sec-WebSocket-Protocol）实测两种拿法：

  服务端用 Header(alias="Sec-WebSocket-Protocol") 声明 -> 能拿到客户端传来的 'chat'
  客户端 websockets.connect(..., subprotocols=["chat"]) 后 a.subprotocol -> None

  • 也就是说 FastAPI/Starlette 不会自动把子协议「协商回去」，要回给客户端得在 accept(subprotocol="chat") 里显式指定，否则客户端看到的 subprotocol 是 None。

生产开关

• check_origin 这类钩子在 FastAPI 里是自己写中间件或依赖，框架没有 Tornado 那样的 check_origin 方法；跨源握手默认是放行的，所以浏览器侧不受同源保护，鉴权必须在 accept() 前做完。

• 帧大小限制在服务器参数上，不在代码里：

  python -m uvicorn main:app --ws-max-size 1000

  发 5000 字节 -> ConnectionClosedError: received 1009 (message too big)
                 frame exceeds limit of 1000 bytes

  超限是 1009 关闭码直接断连，不是回调通知，客户端要能识别这种关闭码。

• WebSocket 相关实现可选：--ws auto|websockets|websockets-sansio|wsproto，本机默认装的是 websockets 17.4。

SSE（服务端推流）

• 只需要「服务器单向推」时，SSE 比 WebSocket 简单得多，用 StreamingResponse 就够：

  sse.py

  import asyncio
  
  from fastapi import FastAPI
  from fastapi.responses import StreamingResponse
  
  app = FastAPI()
  
  
  @app.get("/sse")
  async def sse():
      async def gen():
          for i in range(3):
              yield "data: %d\n\n" % i
              await asyncio.sleep(0.05)
      return StreamingResponse(gen(), media_type="text/event-stream")

  content-type: text/event-stream; charset=utf-8
  收到的行: ['data: 0', 'data: 1', 'data: 2']

  • 每条消息以 data: 开头、以空行结尾，少一个 \n 前端就收不到完整事件。
  • 它复用 HTTP，nginx 侧要关掉缓冲（proxy_buffering off;）才拿得到实时流，这点和 WebSocket 的 Upgrade 是两类配置。
  • 对比 Tornado 那边的长轮询/SSE 方案，见 Tornado-同步与异步以及使用WebSocket实现在线聊天。

测试

• 两条路都能测，官方推荐 TestClient（同步、基于 httpx），异步场景用 httpx.AsyncClient + ASGITransport：

  test_app.py

  import httpx
  import pytest
  from fastapi.testclient import TestClient
  
  from main import app
  
  
  def test_sync():
      client = TestClient(app)
      assert client.get("/slow").status_code == 200
  
  
  def test_ws():
      client = TestClient(app)
      with client.websocket_connect("/ws/room1?token=ok") as ws:
          ws.send_text("ping")
          assert ws.receive_text() == "echo:ping"
  
  
  @pytest.mark.anyio
  async def test_asgi_transport():
      transport = httpx.ASGITransport(app=app)
      async with httpx.AsyncClient(transport=transport, base_url="http://test") as ac:
          assert (await ac.get("/slow")).status_code == 200

  4 passed in 0.33s

  • ASGITransport 不走 HTTP 栈，所以中间件里的网络层行为（如 --proxy-headers、真实端口、TLS）测不出来；要验这些得起真服务再用真实地址请求。
  • TestClient 默认会把服务端异常抛进测试；想验 500 的响应体就 TestClient(app, raise_server_exceptions=False)。
  • 生命周期（lifespan）只在 with TestClient(app) as client: 里触发，直接 TestClient(app).get(...) 不会跑 startup（见 FastAPI-依赖注入与异常中间件）。

• 异步测试的插件选择，实测两种都能跑，但别同时开 auto 模式：

  pytest.ini

  [pytest]
  asyncio_mode = auto

  只配 asyncio_mode=auto            -> 4 passed
  anyio_mode=auto 与 asyncio_mode=auto 同时开
    -> PytestConfigWarning: AnyIO auto mode has been enabled together with
       pytest-asyncio auto mode. This may cause unexpected behavior.

  • 环境里同时装了 pytest-asyncio 与 anyio 自带的 pytest 插件，二选一：用 @pytest.mark.anyio + anyio_backend fixture，或者用 pytest-asyncio，别混。
  • 还会看到一条 PytestDeprecationWarning: asyncio_default_fixture_loop_scope is unset，显式配上 asyncio_default_fixture_loop_scope = function 就干净了。

• 依赖打桩用 app.dependency_overrides，比 monkeypatch 稳（用法与「返回 Response 不会短路」这个坑见 FastAPI-依赖注入与异常中间件）。

• 本机版本：pytest 9.2.0 + pytest-asyncio 1.4.1，httpx 0.28.1。

部署：uvicorn 与 gunicorn

• FastAPI 自己不监听端口，进程模型有三种，本机都实测过：

  方式	命令	实测结果
  单进程	uvicorn main:app	正常，200
  uvicorn 自带多进程	uvicorn main:app --workers 4	Windows 上也能起（2026 的 uvicorn 0.54 已支持），Linux 同样
  gunicorn + UvicornWorker	gunicorn main:app -k uvicorn.workers.UvicornWorker -w 4	WSL 里正常，200

  # 开发
  fastapi dev main.py
  # 单机生产（Linux/Windows 都行）
  uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4
  # 需要 gunicorn 的进程管理时
  gunicorn main:app -k uvicorn.workers.UvicornWorker -w 4 -b 127.0.0.1:8000

  • --workers 的默认值来自 WEB_CONCURRENCY 环境变量（--help 里写的是 “Defaults to the $WEB_CONCURRENCY environment variable if available”），所以容器里设了这个变量就不用再传 --workers。
  • 现在更推荐直接用 uvicorn --workers，少一层进程管理；gunicorn 的价值在于它的 --max-requests、--preload、成熟的信号处理与生态。

• worker 分布实测，这条容易得出错误结论：

  同一个客户端（keep-alive 复用）发 30 个请求 -> 全部落在 1 个 worker
  每次新建连接 + Connection: close 发 30 个 -> 落到 3 个 worker（分布 27/2/1）

  • 连接是粘在 worker 上的：一个长连接的所有请求都由同一个进程处理，所以「压测时只看到一个 worker 忙」通常不是 bug，是客户端复用了连接。
  • 就算每请求新建连接，分布也不均匀（27/2/1），别拿「四个 worker 应该各 25%」当验收标准。
  • 推论：任何进程内状态（WebSocket 房间表、内存缓存、限流计数）在多 worker 下都不共享，要跨进程就放 Redis（见 Redis入门）。

部署：nginx 反代

• 关键配置就三段（本机用第二个 nginx 实例验的，不碰系统那份）：

  nginx.conf

  upstream fastapi_app { server 127.0.0.1:8000; }
  
  server {
    listen 80;
    client_max_body_size 1m;
  
    location / {
      proxy_pass http://fastapi_app;
      proxy_set_header Host $host;
      proxy_set_header X-Real-IP $remote_addr;
      proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
      proxy_set_header X-Forwarded-Proto $scheme;
    }
  
    location /ws {
      proxy_pass http://fastapi_app/ws;
      proxy_http_version 1.1;
      proxy_set_header Upgrade $http_upgrade;
      proxy_set_header Connection "upgrade";
      proxy_read_timeout 60s;
    }
  }

  经 nginx GET /url        -> 200 | scheme: http | client: 127.0.0.1
  2MB 请求体               -> 413 <html><title>413 Request Entity Too Large</title>
                             error.log: client intended to send too large body: 2097160 bytes
  经 nginx 连 /ws          -> 101 Switching Protocols（access.log 记的是 "GET /ws" 101）
  没配 Upgrade 的 /no-upgrade -> HTTP 404，握手直接被拒

  • WebSocket 必须 proxy_http_version 1.1 + 两个 Upgrade 头，缺一就在握手阶段失败（实测 404），前端只会看到「连不上」。
  • client_max_body_size 是 nginx 侧拦的，跟应用无关；上传大文件要同时调它和 uvicorn（见 FastAPI-请求体与Pydantic校验 的文件一节）。
  • 反代必须自己设 X-Forwarded-Proto，否则应用以为自己是 http，生成的重定向、OAuth 回调、request.url 全是 http://。
  • proxy_read_timeout 对 SSE/长连接要放大，默认 60s 会把静默久的流掐掉。
  • 完整安装与站点配置见 Nginx。

部署：代理头与并发上限

• uvicorn 的代理头是默认开启、但只信任 127.0.0.1，实测三种组合：

  默认（不带任何参数）+ X-Forwarded-Proto/For -> scheme=https client=203.0.113.9
  --no-proxy-headers + 同样的头              -> scheme=http  client=127.0.0.1
  --proxy-headers --forwarded-allow-ips='*'   -> 任意来源的转发头都信，
                                                  访问日志里客户端被写成 203.0.113.9:0

  • 也就是说：本地测出来的「转发头生效」不是因为你加了参数，而是默认就信本机。
  • --forwarded-allow-ips '*' 等于把客户端 IP 交给调用方随便填，限流、审计、日志全被污染，生产要写成真实代理 IP 或网段。
  • 命令行里那个 * 一定要引号包住（--forwarded-allow-ips='*'），否则 shell 会把它展成当前目录的文件名，uvicorn 直接报 Got unexpected extra arguments (...)——本机踩过。

• root_path 的两侧要配套，实测：

  FastAPI(root_path="/api/v1")
    /url 响应 -> {'path': '/url', 'root_path': '/api/v1'}
    openapi.json 的 servers -> [{'url': '/api/v1'}]

  • 应用侧 root_path 只影响「生成的 URL / 文档里的 servers」，不改真实路由；nginx 侧如果用 location /api/v1/ { proxy_pass http://app/; } 剥了前缀，就正好对上。

  • 命令行 uvicorn --root-path /api/v1 实测不等价，它会让应用看到的 URL 带上前缀：

    uvicorn rp_app:app --root-path /api/v1
    GET /ping -> {'path': '/api/v1/ping', 'root_path': '/api/v1',
                  'url': 'http://127.0.0.1:8999/api/v1/ping'}
    openapi servers -> [{'url': '/api/v1'}]

    路由仍然按 /ping 匹配（请求要打到 /ping 才 200），但 request.url 与 url_path_for 生成的地址会多出前缀。所以规则是：nginx 剥前缀就只在 nginx 侧配，别两边都设 root_path，否则前缀会被加两次。

• --limit-concurrency 是按连接数算的，实测：

  --limit-concurrency 3，先建 3 条空闲 keep-alive 连接
  第 4 条连接发请求 -> 503
  把那 3 条关掉再发 -> 200
  
  10 个并发打 /slow（每个 0.5s）-> 只有 1~3 个 200，其余全 503

  • 结论：空闲 keep-alive 连接照样占额度，所以这个参数不能按「QPS」去估，要按「同时打开的连接数」估；前端连接池一大就容易误伤。
  • 超限时是直接 503，不排队，配合 --backlog（内核队列长度）才是「排队 + 拒绝」的组合。
  • 相关参数实测存在：--backlog、--timeout-keep-alive、--max-requests-per-connection、--h11-max-incomplete-event-size。

• 可观测性：0.162 起 FastAPI(telemetry=...) 是原生 OpenTelemetry 配置入口（实测传 {"tracing": False} 甚至乱写的键都不报错，宽松配置），配套的 opentelemetry-api 1.45.2 已随 fastapi[standard] 装好；接入方式与指标口径见 Django-接入监控告警 的同类做法。

部署：Docker

• 标准镜像（本机没有 Docker，这一段是按官方写法给的配置，未在本机跑过，落地前请自己 docker build 验一遍）：

  Dockerfile

  FROM python:3.14-slim
  
  ENV PYTHONUNBUFFERED=1 \
      PYTHONDONTWRITEBYTECODE=1 \
      PIP_NO_CACHE_DIR=1
  
  WORKDIR /app
  COPY requirements.txt .
  RUN pip install -r requirements.txt
  
  COPY . .
  
  EXPOSE 8000
  CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

  # requirements.txt
  fastapi[standard]>=0.162

  • PYTHONUNBUFFERED=1 一定要有，否则容器里 print 的日志被缓冲，docker logs 半天看不到东西。
  • 进程数交给编排层（--workers 或 replicas），不要在镜像里写死；容器就是一进程一容器更好扩。
  • 生产别在容器里跑 fastapi dev（它开 reload、只监听 127.0.0.1）。
  • 多阶段构建 + uv 能显著瘦身，思路与 Django 的容器化一致，见 Django-云环境中的部署。

上线检查清单

• 进程与协议：
  • 关掉 debug、关掉 --reload，fastapi dev 只用于开发。
  • --workers 数量按 CPU 核数起，再压测调；确认没有依赖进程内状态。
  • --forwarded-allow-ips 填真实代理，不要 *。
• 应用层：
  • /docs、/redoc、/openapi.json 在公网要么关干净，要么挂鉴权。
  • CORS 名单、allow_credentials 与 allow_origins 的组合复核一遍。
  • 所有 async def 里没有阻塞调用（time.sleep、requests、同步驱动），否则线程池 40 会先撑不住。
  • 数据库连接池上限 × worker 数 < 数据库 max_connections，MySQL 侧的默认值见 MySQL入门。
• 可观测与兜底：
  • 访问日志在 nginx 侧留一份、应用侧留一份，字段里要有请求 ID。
  • 未捕获异常必须有 Exception handler + 上报，否则线上只看到 500 空页。
  • 慢接口与 WebSocket 的超时（proxy_read_timeout、--timeout-keep-alive）要对齐，别让网关先断。

参考

• FastAPI 官方文档 - WebSockets
• FastAPI 官方文档 - Hypertext Middleware Server (nginx)
• FastAPI 官方文档 - Manually Deploying FastAPI
• uvicorn 部署文档 - Workers / Bind Socket / Proxy Headers
• Starlette - TestClient
• httpx - Async transports (ASGITransport)
• Server-Sent Events - MDN