使用 WebSocket 实时通信

消息回显需要客户端与服务端持续收发。反复发送 HTTP 请求可以完成简单查询，但持续双向通信更适合 WebSocket。本专题先建立一条连接，再处理循环接收、断开和多客户端广播。

WebSocket 路由和普通 HTTP 路由不同。连接建立后，服务端先 accept()，再通过 receive_text() 与 send_text() 交换消息。示例见下文。

单次消息回显

最小示例只收发一次，用来确认握手和消息格式。

先写一个最小版本。

这个版本只做一件事：

text

客户端连接
    ↓
服务端接收一条消息
    ↓
服务端返回一条消息

示例文件：01_websocket_echo.py。运行：uvicorn 01_websocket_echo:app --reload。

01_websocket_echo.py

from starlette.applications import Starlette
from starlette.routing import WebSocketRoute
from starlette.websockets import WebSocket


async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    message = await websocket.receive_text()
    await websocket.send_text(f"服务端收到：{message}")


routes = [
    WebSocketRoute("/ws", websocket_endpoint),
]


app = Starlette(routes=routes)


if __name__ == "__main__":
    import uvicorn

    uvicorn.run("01_websocket_echo:app", host="127.0.0.1", port=8000, reload=True)

启动：

bash

uvicorn 01_websocket_echo:app --reload

这个地址不能直接在浏览器地址栏里当普通页面访问，因为它是 WebSocket 地址：

text

ws://127.0.0.1:8000/ws

核心步骤只有四步：

用 WebSocketRoute 注册 /ws。
在 endpoint 中先 await websocket.accept()。
用 receive_text() 收消息。
用 send_text() 回消息。

WebSocket 握手升级与单次回显时序

单次收发的局限

它只能处理一次收发：

python

message = await websocket.receive_text()

接下来我们把它改成循环版本。

持续收发消息

循环处理消息时，客户端断开会触发 WebSocketDisconnect；它是正常连接结束的一种情况。

如果想连续发多条消息，就要使用循环。

示例文件：02_websocket_loop.py。运行：uvicorn 02_websocket_loop:app --reload。

02_websocket_loop.py

from starlette.applications import Starlette
from starlette.routing import WebSocketRoute
from starlette.websockets import WebSocket, WebSocketDisconnect


async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()

    try:
        while True:
            message = await websocket.receive_text()
            await websocket.send_text(f"服务端回显：{message}")
    except WebSocketDisconnect:
        print("客户端断开连接")


routes = [
    WebSocketRoute("/ws", websocket_endpoint),
]


app = Starlette(routes=routes)


if __name__ == "__main__":
    import uvicorn

    uvicorn.run("02_websocket_loop:app", host="127.0.0.1", port=8000, reload=True)

关键在这段代码：

python

while True:
    message = await websocket.receive_text()
    await websocket.send_text(f"服务端回显：{message}")

它表示：

text

只要连接还在
    ↓
就一直等下一条消息
    ↓
收到后马上回显

客户端关闭页面后，会进入：

python

except WebSocketDisconnect:
    print("客户端断开连接")

循环接收消息

因为 WebSocket 建立连接后，通常不是只通信一次，而是持续通信：

text

收消息 1
回消息 1
收消息 2
回消息 2
收消息 3
回消息 3

所以很多 WebSocket endpoint 都会出现循环。

while True 消息循环与 WebSocketDisconnect 异常控制流

从页面连接 WebSocket

页面通过浏览器 WebSocket API 建立连接；这一节需要页面交互，因此保留浏览器操作说明。

前面只有服务端，还没有浏览器页面。

现在增加一个普通 HTTP 页面：

text

GET /ws-page

页面中通过 JavaScript 创建 WebSocket 连接。

示例文件：03_websocket_page.py。运行：uvicorn 03_websocket_page:app --reload。

03_websocket_page.py

from starlette.applications import Starlette
from starlette.routing import Route, WebSocketRoute
from starlette.templating import Jinja2Templates
from starlette.websockets import WebSocket, WebSocketDisconnect


templates = Jinja2Templates(directory="templates")


async def websocket_page(request):
    return templates.TemplateResponse(
        request, "ws.html", {"title": "WebSocket 消息演示"}
    )


async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    try:
        while True:
            message = await websocket.receive_text()
            await websocket.send_text(f"服务端回显：{message}")
    except WebSocketDisconnect:
        print("客户端断开连接")


app = Starlette(routes=[
    Route("/ws-page", websocket_page),
    WebSocketRoute("/ws", websocket_endpoint),
])

templates/ws.html：

templates/ws.html

html

<!DOCTYPE html>
<html lang="zh-CN">
<head>
    <meta charset="UTF-8">
    <title>{{ title }}</title>
</head>
<body>
    <h1>{{ title }}</h1>
    <p>连接地址：ws://127.0.0.1:8000/ws</p>
    <input id="message-input" type="text" placeholder="请输入消息">
    <button onclick="sendMessage()">发送</button>
    <ul id="message-list"></ul>
    <script>
        const socket = new WebSocket("ws://127.0.0.1:8000/ws");
        socket.onopen = function () { addMessage("连接已建立"); };
        socket.onmessage = function (event) { addMessage(event.data); };
        socket.onclose = function () { addMessage("连接已关闭"); };
        function sendMessage() {
            const input = document.getElementById("message-input");
            if (socket.readyState !== WebSocket.OPEN) return;
            socket.send(input.value);
            input.value = "";
        }
        function addMessage(message) {
            const item = document.createElement("li");
            item.textContent = message;
            document.getElementById("message-list").appendChild(item);
        }
    </script>
</body>
</html>

启动：

bash

uvicorn 03_websocket_page:app --reload

访问：

text

http://127.0.0.1:8000/ws-page

这里同时用了两种路由：

python

Route("/ws-page", websocket_page)
WebSocketRoute("/ws", websocket_endpoint)

/ws-page 是普通 HTTP 页面。 /ws 是 WebSocket 连接地址。

页面加载后会执行：

javascript

const socket = new WebSocket("ws://127.0.0.1:8000/ws");

然后：

text

用户输入消息
    ↓
socket.send(...)
    ↓
服务端 receive_text()
    ↓
服务端 send_text()
    ↓
浏览器 onmessage 收到回显

页面与 WebSocket 接口

可以这样理解：

text

浏览器访问 /ws-page
    ↓
服务端返回 HTML
    ↓
浏览器执行 JavaScript
    ↓
JavaScript 建立 ws://... 连接
    ↓
用户发送消息
    ↓
后端回显消息
    ↓
页面显示结果

本章只用原生 JavaScript，不引入前端框架。

混合路由体系与浏览器前端交互驱动流

向多个客户端广播

广播示例维护进程内连接列表，适合课堂理解机制；跨进程共享和可靠投递不属于这个示例。

前面的例子是“谁发消息，就回给谁”。现在做一个教学版广播：

text

多个浏览器窗口连接同一个地址
任意一个窗口发消息
所有窗口都收到同一条广播消息

示例文件：04_websocket_broadcast.py。运行：uvicorn 04_websocket_broadcast:app --reload。

python

from starlette.applications import Starlette
from starlette.routing import WebSocketRoute
from starlette.websockets import WebSocket, WebSocketDisconnect

connected_websockets = []

async def broadcast_websocket(websocket: WebSocket):
    await websocket.accept()
    connected_websockets.append(websocket)

    try:
        while True:
            message = await websocket.receive_text()
            for client in connected_websockets:
                await client.send_text(f"广播消息：{message}")
    except WebSocketDisconnect:
        connected_websockets.remove(websocket)
        print("客户端断开广播连接")

routes = [
    WebSocketRoute("/ws/broadcast", broadcast_websocket),
]

app = Starlette(routes=routes)

核心思路很简单：

text

新连接进来
    ↓
放进 connected_websockets
    ↓
某个连接发来消息
    ↓
遍历列表，把消息发给所有连接
    ↓
连接断开时，从列表里移除

广播示例的边界

因为它没有处理：

text

用户身份
房间
权限
消息历史
连接异常
多进程或多服务器同步

所以不要把这个版本直接当成生产聊天室。

进程内连接池管理与 1:N 广播扇出流水线

HTTP、后台任务与 WebSocket

现在我们已经学过三类能力：

机制	典型特点	适合场景
普通 HTTP	一次请求，一次响应	页面、表单、普通接口
后台任务	响应返回后执行轻量动作	日志、通知、临时文件清理
WebSocket	建立连接后持续双向通信	回显、实时通知、简单聊天

HTTP 请求、后台任务与 WebSocket 通信模型全景对比

HTTP 请求

例如：

text

GET /articles
POST /api/articles

这些仍然更适合 HTTP。

后台任务

例如：

text

创建文章后写日志
创建文章后发送简单通知
响应返回后清理临时文件

这些适合后台任务。

WebSocket 连接

例如：

text

消息回显
实时通知
简单聊天
浏览器与服务端持续连接

这些更适合 WebSocket。

交互式消息页面

完整案例提供两个页面：GET /ws-page 演示回显，GET /broadcast-page 演示广播。它们分别连接 /ws 与 /ws/broadcast，只保留理解 WebSocket 所需的页面和连接代码。前几节的单次收发与循环收发仍可独立运行。

广播接口把已连接的客户端保存在一个列表中：

python

connected_websockets: list[WebSocket] = []

async def broadcast_websocket(websocket: WebSocket):
    await websocket.accept()
    connected_websockets.append(websocket)
    try:
        while True:
            message = await websocket.receive_text()
            for client in tuple(connected_websockets):
                await client.send_text(f"广播消息：{message}")
    except WebSocketDisconnect:
        print("广播连接已断开")
    finally:
        connected_websockets.remove(websocket)

在 _code_starlette/p03c03 中启动 uvicorn 04_websocket_broadcast:app --reload。访问 /ws-page 发送一条消息，应看到服务端回显；再用两个窗口访问 /broadcast-page，任一窗口发送消息，两边都会显示。也可运行 python websocket_demo.py，用 TestClient 查看四个阶段的状态与消息。

这个广播列表只存在于当前进程。多个工作进程之间不会共享连接；一个发送动作也可能因某个连接异常而中断。它适合解释广播机制，不应直接作为可靠消息系统。

案例源码

04_websocket_broadcast.pywebsocket_demo.py

from starlette.applications import Starlette
from starlette.routing import Route, WebSocketRoute
from starlette.templating import Jinja2Templates
from starlette.websockets import WebSocket, WebSocketDisconnect


templates = Jinja2Templates(directory="templates")
connected_websockets: list[WebSocket] = []


async def websocket_page(request):
    return templates.TemplateResponse(
        request, "ws.html", {"title": "WebSocket 消息演示"}
    )


async def broadcast_page(request):
    return templates.TemplateResponse(
        request, "broadcast.html", {"title": "多客户端广播演示"}
    )


async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    try:
        while True:
            message = await websocket.receive_text()
            await websocket.send_text(f"服务端回显：{message}")
    except WebSocketDisconnect:
        print("回显连接已断开")


async def broadcast_websocket(websocket: WebSocket):
    await websocket.accept()
    connected_websockets.append(websocket)
    try:
        while True:
            message = await websocket.receive_text()
            for client in tuple(connected_websockets):
                await client.send_text(f"广播消息：{message}")
    except WebSocketDisconnect:
        print("广播连接已断开")
    finally:
        connected_websockets.remove(websocket)


app = Starlette(routes=[
    Route("/ws-page", websocket_page),
    Route("/broadcast-page", broadcast_page),
    WebSocketRoute("/ws", websocket_endpoint),
    WebSocketRoute("/ws/broadcast", broadcast_websocket),
])

from starlette.testclient import TestClient

from importlib import import_module

echo_app = import_module("01_websocket_echo")
loop_app = import_module("02_websocket_loop")
page_app = import_module("03_websocket_page")
broadcast_app = import_module("04_websocket_broadcast")


def test_single_message():
    with TestClient(echo_app.app) as client:
        with client.websocket_connect("/ws") as websocket:
            websocket.send_text("你好")
            print("单次回显：", websocket.receive_text())


def test_multiple_messages():
    with TestClient(loop_app.app) as client:
        with client.websocket_connect("/ws") as websocket:
            websocket.send_text("第一条")
            print("第一次：", websocket.receive_text())
            websocket.send_text("第二条")
            print("第二次：", websocket.receive_text())


def test_websocket_page():
    with TestClient(page_app.app) as client:
        response = client.get("/ws-page")
        print("页面状态码：", response.status_code)
        print("页面标题存在：", "WebSocket 消息演示" in response.text)


def test_broadcast():
    with TestClient(broadcast_app.app) as client:
        with client.websocket_connect("/ws/broadcast") as first:
            with client.websocket_connect("/ws/broadcast") as second:
                first.send_text("课堂消息")
                print("客户端一：", first.receive_text())
                print("客户端二：", second.receive_text())


if __name__ == "__main__":
    test_single_message()
    test_multiple_messages()
    test_websocket_page()
    test_broadcast()

页面模板

templates/broadcast.html

html

<!DOCTYPE html>
<html lang="zh-CN">
<head>
    <meta charset="UTF-8">
    <title>{{ title }}</title>
</head>
<body>
    <h1>{{ title }}</h1>
    <p>在两个窗口打开此页，任一窗口发送的消息会出现在两边。</p>
    <input id="message-input" type="text" placeholder="请输入消息">
    <button onclick="sendMessage()">广播</button>
    <ul id="message-list"></ul>
    <p><a href="/ws-page">返回回显演示</a></p>
    <script>
        const socket = new WebSocket("ws://127.0.0.1:8000/ws/broadcast");
        socket.onmessage = function (event) {
            const item = document.createElement("li");
            item.textContent = event.data;
            document.getElementById("message-list").appendChild(item);
        };
        function sendMessage() {
            const input = document.getElementById("message-input");
            if (socket.readyState !== WebSocket.OPEN) return;
            socket.send(input.value);
            input.value = "";
        }
    </script>
</body>
</html>

使用 WebSocket 实时通信 ​

单次消息回显 ​

单次收发的局限 ​

持续收发消息 ​

循环接收消息 ​

从页面连接 WebSocket ​

页面与 WebSocket 接口 ​

向多个客户端广播 ​

广播示例的边界 ​

HTTP、后台任务与 WebSocket ​

HTTP 请求 ​

后台任务 ​

WebSocket 连接 ​

交互式消息页面 ​

案例源码 ​

页面模板 ​

使用 WebSocket 实时通信

单次消息回显

单次收发的局限

持续收发消息

循环接收消息

从页面连接 WebSocket

页面与 WebSocket 接口

向多个客户端广播

广播示例的边界

HTTP、后台任务与 WebSocket

HTTP 请求

后台任务

WebSocket 连接

交互式消息页面

案例源码

页面模板