WebSocket是一种基于TCP的全双工通信协议，允许客户端和服务端建立持久连接后双向实时传输数据，避免了HTTP协议下频繁建立连接的开销，非常适合即时聊天、实时数据推送、在线协作等场景。Python的FastAPI框架内置了对WebSocket的良好支持，无需额外依赖复杂的第三方库就能快速搭建服务。
WebSocket核心工作原理
WebSocket连接建立需要经过一次HTTP握手过程，客户端发送包含Upgrade: websocket头的请求，服务端验证通过后返回101状态码，之后连接就升级为WebSocket协议，双方可以随时向对方发送数据帧。整个通信过程中连接保持打开状态，直到任意一方主动关闭连接。
使用FastAPI搭建WebSocket服务端
首先确保已经安装了FastAPI和uvicorn，若未安装可以通过pip安装：
pip install fastapi uvicorn
下面是完整的FastAPI WebSocket服务端代码，实现接收客户端消息并回传的功能：
from fastapi import FastAPI, WebSocket
from fastapi.responses import HTMLResponse
app = FastAPI()
# 存储所有活跃的WebSocket连接
active_connections = []
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
# 接受客户端的WebSocket连接请求
await websocket.accept()
active_connections.append(websocket)
try:
while True:
# 接收客户端发送的消息
data = await websocket.receive_text()
# 向当前客户端回传接收到的消息
await websocket.send_text(f"服务端收到消息: {data}")
# 可选：向所有活跃连接广播消息
for connection in active_connections:
if connection != websocket:
await connection.send_text(f"其他客户端发送了消息: {data}")
except Exception as e:
print(f"连接异常关闭: {e}")
finally:
# 连接关闭后从活跃列表中移除
if websocket in active_connections:
active_connections.remove(websocket)
# 提供一个简单的测试页面
@app.get("/")
async def get():
html_content = """
<html>
<head>
<title>WebSocket测试</title>
</head>
<body>
<h1>WebSocket实时通信测试</h1>
<input type="text" id="messageInput" placeholder="输入要发送的消息">
<button onclick="sendMessage()">发送消息</button>
<div id="messageArea"></div>
<script>
const ws = new WebSocket("ws://localhost:8000/ws");
ws.onopen = function() {
console.log("WebSocket连接已建立");
};
ws.onmessage = function(event) {
const messageArea = document.getElementById("messageArea");
messageArea.innerHTML += "<p>" + event.data + "</p>";
};
ws.onclose = function() {
console.log("WebSocket连接已关闭");
};
function sendMessage() {
const input = document.getElementById("messageInput");
const message = input.value;
if (message) {
ws.send(message);
input.value = "";
}
}
</script>
</body>
</html>
"""
return HTMLResponse(content=html_content)
运行服务端代码的命令如下：
uvicorn main:app --reload
启动后访问http://127.0.0.1:8000就能打开测试页面，输入消息点击发送即可看到服务端的回传。
编写Python客户端协同通信
除了前端客户端，也可以编写Python客户端与FastAPI服务端进行通信，下面是Python客户端的实现代码：
import asyncio
import websockets
async def websocket_client():
# 连接到FastAPI的WebSocket服务
async with websockets.connect("ws://127.0.0.1:8000/ws") as websocket:
# 发送消息到服务端
await websocket.send("这是Python客户端发送的消息")
# 接收服务端的回传消息
response = await websocket.recv()
print(f"收到服务端响应: {response}")
# 保持连接，持续接收消息
while True:
try:
msg = await websocket.recv()
print(f"收到新消息: {msg}")
except Exception as e:
print(f"连接关闭: {e}")
break
if __name__ == "__main__":
asyncio.run(websocket_client())
运行Python客户端前需要先安装websockets库：
pip install websockets
开发注意事项与优化建议
- 连接管理：实际生产环境中需要对连接做更完善的异常处理，比如网络波动导致的连接断开自动重连机制。
- 消息格式：建议使用JSON格式传输结构化数据，而不是直接传输纯文本，方便扩展消息类型。
- 并发控制：FastAPI的WebSocket处理是异步的，高并发场景下需要注意资源占用，避免单个连接阻塞整个服务。
- 鉴权处理：可以在WebSocket握手阶段添加鉴权逻辑，验证客户端的身份令牌，避免未授权连接接入。
常见问题解答
WebSocket连接握手失败怎么办
首先检查服务端路由是否正确，客户端连接的地址是否和服务端定义的WebSocket路由一致，其次确认服务端已经正常启动，端口没有被其他程序占用。
如何向指定客户端发送消息
可以在客户端连接时给每个连接绑定唯一标识，比如用户ID，发送消息时根据标识找到对应的连接对象再发送数据即可。
FastAPI的WebSocket支持二进制数据传输吗
支持，除了receive_text和send_text方法，还有receive_bytes和send_bytes方法可以处理二进制数据，适合传输图片、文件等内容。