FastAPI Backend Development Bootcamp · 课时

使用 WebSockets 实现实时通信

使用 WebSockets 在 FastAPI 中构建实时功能：接受连接、交换消息、向多个客户端广播，并使用异步模式妥善处理断开连接。

第 4 / 4 课13 个步骤

使用 WebSockets 实现实时通信 是 CoddyKit 上的免费 FastAPI Backend Development Bootcamp 课时。这是第 4 节课，共 4 节。你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 AI 导师进行实践。这是 FastAPI Backend Development Bootcamp 学习路径的一部分，你的进度在网页和 CoddyKit 应用中同步。 FastAPI Backend Development Bootcamp 课程共包含 4 节课。

本课时的部分内容尚未翻译，以英文显示。

From Request/Response to Real-Time

Regular HTTP is request/response: the client asks, the server answers, the connection closes. WebSockets keep a single connection open for full-duplex, real-time messaging. Perfect for chat, live dashboards, and notifications.

Why Async Fits WebSockets

A WebSocket connection lives a long time and is mostly idle waiting for messages. FastAPI's async model lets one worker handle thousands of concurrent connections without blocking.

A Minimal WebSocket Endpoint

Declare a @app.websocket route. Accept the connection, then loop receiving and sending messages.

from fastapi import FastAPI, WebSocket

app = FastAPI()

@app.websocket('/ws')
async def ws(websocket: WebSocket):
    await websocket.accept()
    while True:
        msg = await websocket.receive_text()
        await websocket.send_text(f'echo: {msg}')

Sending JSON

You can exchange structured data with receive_json and send_json instead of raw text.

data = await websocket.receive_json()
await websocket.send_json({'received': data, 'status': 'ok'})

Handling Disconnects

Clients drop off. Catch WebSocketDisconnect to clean up resources instead of crashing the handler.

from fastapi import WebSocketDisconnect

try:
    while True:
        await websocket.receive_text()
except WebSocketDisconnect:
    print('client left')

A Connection Manager

To broadcast, track active connections in a manager class. Add on connect, remove on disconnect.

class Manager:
    def __init__(self):
        self.active = []
    async def connect(self, ws):
        await ws.accept()
        self.active.append(ws)
    def disconnect(self, ws):
        self.active.remove(ws)

Broadcasting to Everyone

Loop the active connections and send each one the message. This is the core of a chat room.

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

Broadcast Logic in Plain Python

The manager is just list bookkeeping. Here is the connect/disconnect/broadcast flow simulated synchronously.

active = []
def connect(c): active.append(c)
def disconnect(c): active.remove(c)
def broadcast(msg): return [f'{c}<-{msg}' for c in active]
connect('alice'); connect('bob')
print(broadcast('hi'))
disconnect('alice')
print(broadcast('bye'))

A Chat Room Endpoint

Combine the pieces: connect, broadcast each received message, and disconnect on drop.

manager = Manager()

@app.websocket('/chat')
async def chat(ws: WebSocket):
    await manager.connect(ws)
    try:
        while True:
            text = await ws.receive_text()
            await manager.broadcast(text)
    except WebSocketDisconnect:
        manager.disconnect(ws)

Scaling Beyond One Process

An in-memory manager only knows connections on its worker. To broadcast across multiple workers or servers, use a pub/sub backend like Redis to fan out messages.

Security and Limits

Protect WebSocket endpoints:

Authenticate during the handshake (e.g. token in query or header).
Validate and size-limit incoming messages.
Heartbeat/ping to detect dead connections.

Quick Check

Why does an in-memory connection manager fail to broadcast correctly when you run several Uvicorn workers?

在交互式课程中回答

Recap

You added real-time communication:

Accepted WebSocket connections and exchanged text/JSON.
Handled WebSocketDisconnect cleanly.
Built a connection manager to broadcast to many clients.
Learned to scale with Redis pub/sub and to secure connections.

免费开始

用 AI 导师学习 FastAPI Backend Development Bootcamp — 免费

在浏览器中编写并运行真实代码，获得全天候 AI 导师的即时帮助，并在网页或应用中继续学习。

课程: 21
课程: 84

免费开始学习查看完整课程

常见问题解答

「使用 WebSockets 实现实时通信」课时是免费的吗？

是的 — 「使用 WebSockets 实现实时通信」的完整文本可在网页上免费阅读。要进行交互式练习（内置代码编辑器和全天候 AI 导师）并解锁 FastAPI Backend Development Bootcamp 课程的其余内容，请升级到 CoddyKit PRO。 FastAPI Backend Development Bootcamp 课程共包含 4 节课。

「使用 WebSockets 实现实时通信」这节课中我会学到什么？

使用 WebSockets 在 FastAPI 中构建实时功能：接受连接、交换消息、向多个客户端广播，并使用异步模式妥善处理断开连接。你通过在浏览器中直接运行的动手代码来练习 FastAPI Backend Development Bootcamp，全天候 AI 导师会在你学习这节课的过程中回答你的问题。

学习 FastAPI Backend Development Bootcamp 需要有经验吗？

无需任何先前经验。CoddyKit 上的 FastAPI Backend Development Bootcamp 课程适合初学者到高级学习者，你可以从这里开始或从头开始，按照自己的节奏学习。这是第 4 节课，共 4 节。

「使用 WebSockets 实现实时通信」课时需要多长时间？

大多数 CoddyKit 课程大约需要 5–10 分钟。每节课都很精短且互动，所以你能稳步进步，并在网页和应用中从离开的地方继续。

我能在这节 FastAPI Backend Development Bootcamp 课中编写并运行代码吗？

能。每节 FastAPI Backend Development Bootcamp 课都包含内置代码编辑器，你可以在浏览器中直接编写并运行真实代码，并获得即时 AI 反馈 — 无需本地设置。

此课程中的所有课时

← 返回 FastAPI Backend Development Bootcamp