Python FastAPI WebSocket详解

Python FastAPI WebSocket详解

什么是WebSocket

WebSocket是一种在单个TCP连接上进行全双工通信的协议。它允许客户端和服务器之间进行实时通信，从而可以实现实时更新、聊天功能等互动性强的功能。与传统的HTTP协议相比，WebSocket具有更低的延迟和更高的效率，使得它成为开发实时网络应用的理想选择。

FastAPI

FastAPI是一个现代的Python web框架，具有高性能和易用性。它基于Starlette框架，使用Pydantic进行数据验证，提供了自动生成API文档的功能。FastAPI支持异步请求处理，可以在性能要求苛刻的应用中发挥出色的性能。

在FastAPI中使用WebSocket

在FastAPI中使用WebSocket非常方便，只需要使用WebSockets库即可。下面我们来演示如何在FastAPI应用中实现WebSocket功能。

首先，我们需要安装WebSockets库：

pip install websockets

接着我们创建一个FastAPI应用，并添加WebSocket路由：

from fastapi import FastAPI
import websockets

app = FastAPI()

@app.websocket("/ws")
async def websocket_endpoint(websocket: websockets.WebSocket):
    await websocket.send_text("Hello, WebSocket!")
    while True:
        data = await websocket.receive_text()
        await websocket.send_text(f"Received: {data}")

在上面的代码中，我们定义了一个WebSocket路由/ws，并在其处理函数中实现了简单的消息收发逻辑。

现在我们可以运行FastAPI应用，并测试WebSocket功能。启动应用：

uvicorn main:app --reload

然后通过浏览器或者WebSocket客户端连接到ws://localhost:8000/ws，你会收到消息Hello, WebSocket!。然后你可以输入消息发送给服务器，服务器会原样返回。

WebSocket消息广播

除了简单的消息收发外，WebSocket还支持消息的广播功能，即向所有连接的客户端发送消息。下面我们来演示如何实现WebSocket消息广播。

from fastapi import FastAPI
import websockets

app = FastAPI()
connected_users = set()

async def notify_users(message):
    if connected_users:
        await asyncio.gather([user.send(message) for user in connected_users])

@app.websocket("/ws")
async def websocket_endpoint(websocket: websockets.WebSocket):
    connected_users.add(websocket)
    await notify_users("A new user has connected.")

    try:
        while True:
            data = await websocket.receive_text()
            await notify_users(f"User: {data}")
    finally:
        connected_users.remove(websocket)
        await notify_users("A user has disconnected.")

在上面的代码中，我们维护了一个connected_users集合来保存所有连接的客户端。当有新的客户端连接时，向所有客户端发送消息A new user has connected.；当有客户端断开连接时，向所有客户端发送消息A user has disconnected.。

结语

通过本文的介绍，我们学习了如何在FastAPI应用中使用WebSocket实现实时通信功能。WebSocket是一种非常有用的协议，可以为网络应用提供实时交互功能。结合FastAPI框架，我们可以快速地搭建出高性能的Web应用，满足各种应用场景的需求。

本文链接：http://so.lmcjl.com/news/7807/