当前位置：首页>python>Python Web开发:WebSocket实时通信

Python Web开发:WebSocket实时通信

2026-06-27 21:56:11

副标题

: 90%的人不知道，WebSocket能让你的应用实现真正的实时交互

痛点：为什么你的应用总是需要不断轮询？

2025年某聊天应用每秒发送1000次HTTP请求获取新消息。问题出在哪？工程师没有使用WebSocket。

真相

：WebSocket建立持久连接，服务器可以主动推送数据。

方案	延迟	带宽	复杂度
HTTP轮询	高	高	低
长轮询	中	中	中
WebSocket	低	低	中
SSE	低	低	低

一、WebSocket基础

1.1 什么是WebSocket

WebSocket是一种双向通信协议，允许服务器主动向客户端推送数据。

HTTP轮询:  客户端 → 请求 → 服务器 → 响应
(重复循环)
WebSocket: 客户端 ←→ 服务器 (持久连接)
(双向实时通信)

1.2 FastAPI WebSocket

from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from typing import List
app = FastAPI()
class ConnectionManager:
def __init__(self):
self.active_connections: List[WebSocket] = []
async def connect(self, websocket: WebSocket):
await websocket.accept()
self.active_connections.append(websocket)
def disconnect(self, websocket: WebSocket):
self.active_connections.remove(websocket)
async def send_personal_message(self, message: str, websocket: WebSocket):
await websocket.send_text(message)
async def broadcast(self, message: str):
for connection in self.active_connections:
await connection.send_text(message)
manager = ConnectionManager()
@app.websocket("/ws/{client_id}")
async def websocket_endpoint(websocket: WebSocket, client_id: int):
await manager.connect(websocket)
try:
while True:
data = await websocket.receive_text()
await manager.send_personal_message(
f"You wrote: {data}", websocket
)
await manager.broadcast(
f"Client #{client_id} says: {data}"
)
except WebSocketDisconnect:
manager.disconnect(websocket)
await manager.broadcast(
f"Client #{client_id} left the chat"
)

1.3 客户端连接

<!DOCTYPE html>
<html>
<head>
<title>WebSocket Chat</title>
</head>
<body>
<h1>WebSocket Chat</h1>
<div id="messages"></div>
<input id="messageInput" placeholder="输入消息">
<button onclick="sendMessage()">发送</button>
<script>
const ws = new WebSocket('ws://localhost:8000/ws/1');
ws.onopen = () => {
console.log('已连接');
};
ws.onmessage = (event) => {
const messagesDiv = document.getElementById('messages');
messagesDiv.innerHTML += <p>${event.data}</p>;
};
ws.onclose = () => {
console.log('已断开');
};
function sendMessage() {
const input = document.getElementById('messageInput');
ws.send(input.value);
input.value = '';
}
</script>
</body>
</html>

二、聊天室实现

2.1 房间管理

from fastapi import FastAPI, WebSocket, WebSocketDisconnect, HTTPException
from typing import Dict, List
from pydantic import BaseModel
import json
app = FastAPI()
class ConnectionManager:
def __init__(self):
# {room_id: {client_id: WebSocket}}
self.rooms: Dict[str, Dict[int, WebSocket]] = {}
async def connect(self, websocket: WebSocket, room_id: str, client_id: int):
await websocket.accept()
if room_id not in self.rooms:
self.rooms[room_id] = {}
self.rooms[room_id][client_id] = websocket
def disconnect(self, room_id: str, client_id: int):
if room_id in self.rooms:
self.rooms[room_id].pop(client_id, None)
if not self.rooms[room_id]:
del self.rooms[room_id]
async def send_to_room(self, room_id: str, message: str):
if room_id in self.rooms:
for websocket in self.rooms[room_id].values():
await websocket.send_text(message)
async def send_to_user(self, room_id: str, client_id: int, message: str):
if room_id in self.rooms and client_id in self.rooms[room_id]:
await self.rooms[room_id][client_id].send_text(message)
manager = ConnectionManager()
class Message(BaseModel):
type: str  # "chat", "join", "leave", "private"
content: str
sender_id: int
target_id: int | None = None
@app.websocket("/ws/{room_id}/{client_id}")
async def websocket_endpoint(
websocket: WebSocket,
room_id: str,
client_id: int
):
await manager.connect(websocket, room_id, client_id)
# 通知房间其他人
await manager.send_to_room(
room_id,
json.dumps({
"type": "system",
"content": f"用户 {client_id} 加入了房间"
})
)
try:
while True:
data = await websocket.receive_text()
message = json.loads(data)
if message["type"] == "chat":
# 广播消息
await manager.send_to_room(
room_id,
json.dumps({
"type": "chat",
"content": message["content"],
"sender_id": client_id
})
)
elif message["type"] == "private" and message.get("target_id"):
# 私聊
await manager.send_to_user(
room_id,
message["target_id"],
json.dumps({
"type": "private",
"content": message["content"],
"sender_id": client_id
})
)
except WebSocketDisconnect:
manager.disconnect(room_id, client_id)
await manager.send_to_room(
room_id,
json.dumps({
"type": "system",
"content": f"用户 {client_id} 离开了房间"
})
)

2.2 消息历史记录

from collections import deque
from datetime import datetime
class ChatRoom:
def __init__(self, max_history: int = 100):
self.clients: Dict[int, WebSocket] = {}
self.history: deque = deque(maxlen=max_history)
async def add_message(self, message: dict):
"""添加消息到历史"""
message["timestamp"] = datetime.utcnow().isoformat()
self.history.append(message)
async def get_history(self, limit: int = 50) -> list:
"""获取历史消息"""
return list(self.history)[-limit:]
async def broadcast(self, message: str):
"""广播消息"""
for websocket in self.clients.values():
await websocket.send_text(message)
房间存储
rooms: Dict[str, ChatRoom] = {}
def get_room(room_id: str) -> ChatRoom:
if room_id not in rooms:
rooms[room_id] = ChatRoom()
return rooms[room_id]

三、实时通知

3.1 通知系统

from fastapi import FastAPI, WebSocket, Depends
from pydantic import BaseModel
from datetime import datetime
from typing import Dict, List
import json
app = FastAPI()
class Notification(BaseModel):
type: str
title: str
message: str
data: dict | None = None
created_at: datetime = None
class NotificationManager:
def __init__(self):
# {user_id: WebSocket}
self.connections: Dict[int, WebSocket] = {}
self.user_rooms: Dict[str, Dict[int, WebSocket]] = {}
async def connect(self, websocket: WebSocket, user_id: int):
await websocket.accept()
self.connections[user_id] = websocket
def disconnect(self, user_id: int):
self.connections.pop(user_id, None)
async def send_notification(self, user_id: int, notification: Notification):
if user_id in self.connections:
await self.connections[user_id].send_text(
json.dumps(notification.model_dump())
)
async def broadcast_notification(self, notification: Notification):
for websocket in self.connections.values():
await websocket.send_text(
json.dumps(notification.model_dump())
)
manager = NotificationManager()
@app.websocket("/ws/notifications/{user_id}")
async def notifications_endpoint(
websocket: WebSocket,
user_id: int
):
await manager.connect(websocket, user_id)
try:
while True:
# 保持连接，等待服务器推送
await websocket.receive_text()
except WebSocketDisconnect:
manager.disconnect(user_id)
@app.post("/api/notifications/send")
async def send_notification(notification: Notification):
"""发送通知给特定用户"""
await manager.send_notification(notification.user_id, notification)
return {"status": "sent"}
@app.post("/api/notifications/broadcast")
async def broadcast_notification(notification: Notification):
"""广播通知给所有用户"""
await manager.broadcast_notification(notification)
return {"status": "broadcasted"}

3.2 前端通知

class NotificationClient {
constructor(userId) {
this.userId = userId;
this.ws = null;
this.reconnectDelay = 1000;
}
connect() {
this.ws = new WebSocket(ws://localhost:8000/ws/notifications/${this.userId});
this.ws.onmessage = (event) => {
const notification = JSON.parse(event.data);
this.showNotification(notification);
};
this.ws.onclose = () => {
console.log('连接断开，尝试重连...');
setTimeout(() => this.connect(), this.reconnectDelay);
this.reconnectDelay = Math.min(this.reconnectDelay * 2, 30000);
};
this.ws.onerror = (error) => {
console.error('WebSocket错误:', error);
};
}
showNotification(notification) {
// 显示浏览器通知
if (Notification.permission === 'granted') {
new Notification(notification.title, {
body: notification.message,
data: notification.data
});
}
// 或者显示应用内通知
this.displayInAppNotification(notification);
}
displayInAppNotification(notification) {
const container = document.getElementById('notifications');
const item = document.createElement('div');
item.className = 'notification-item';
item.innerHTML = 
<h4>${notification.title}</h4>
<p>${notification.message}</p>
        ;
container.appendChild(item);
// 自动移除
setTimeout(() => item.remove(), 5000);
}
disconnect() {
if (this.ws) {
this.ws.close();
}
}
}
// 使用
const client = new NotificationClient(123);
client.connect();
// 请求浏览器通知权限
Notification.requestPermission();

四、实时数据同步

4.1 协作编辑

from fastapi import FastAPI, WebSocket
from typing import Dict, List
import json
from datetime import datetime
app = FastAPI()
class Document:
def __init__(self, doc_id: str):
self.doc_id = doc_id
self.content = ""
self.version = 0
self.clients: Dict[int, WebSocket] = {}
self.history: List[dict] = []
async def apply_operation(self, operation: dict, client_id: int):
"""应用操作"""
self.version += 1
if operation["type"] == "insert":
pos = operation["position"]
text = operation["text"]
self.content = self.content[:pos] + text + self.content[pos:]
elif operation["type"] == "delete":
pos = operation["position"]
length = operation["length"]
self.content = self.content[:pos] + self.content[pos + length:]
# 记录历史
self.history.append({
"version": self.version,
"operation": operation,
"client_id": client_id,
"timestamp": datetime.utcnow().isoformat()
})
# 广播给其他客户端
await self.broadcast({
"type": "operation",
"operation": operation,
"version": self.version,
"client_id": client_id
})
async def broadcast(self, message: dict):
for client_id, websocket in self.clients.items():
await websocket.send_text(json.dumps(message))
documents: Dict[str, Document] = {}
def get_document(doc_id: str) -> Document:
if doc_id not in documents:
documents[doc_id] = Document(doc_id)
return documents[doc_id]
@app.websocket("/ws/document/{doc_id}/{client_id}")
async def document_endpoint(
websocket: WebSocket,
doc_id: str,
client_id: int
):
doc = get_document(doc_id)
doc.clients[client_id] = websocket
# 发送当前内容
await websocket.send_text(json.dumps({
"type": "init",
"content": doc.content,
"version": doc.version
}))
try:
while True:
data = await websocket.receive_text()
message = json.loads(data)
if message["type"] == "operation":
await doc.apply_operation(message["operation"], client_id)
elif message["type"] == "cursor":
# 光标位置同步
await doc.broadcast({
"type": "cursor",
"client_id": client_id,
"position": message["position"]
})
except Exception as e:
print(f"Error: {e}")
finally:
doc.clients.pop(client_id, None)

4.2 操作转换（OT）

class OperationTransformer:
"""操作转换"""
@staticmethod
def transform(op1: dict, op2: dict) -> tuple[dict, dict]:
"""转换两个操作"""
# 简化版本，实际OT更复杂
if op1["type"] == "insert" and op2["type"] == "insert":
if op1["position"] <= op2["position"]:
op2["position"] += len(op1["text"])
else:
op1["position"] += len(op2["text"])
elif op1["type"] == "delete" and op2["type"] == "delete":
if op1["position"] < op2["position"]:
op2["position"] -= op1["length"]
return op1, op2

五、游戏实时通信

5.1 多人游戏

from fastapi import FastAPI, WebSocket
from typing import Dict, List
import json
import random
app = FastAPI()
class GameRoom:
def __init__(self, room_id: str):
self.room_id = room_id
self.players: Dict[int, dict] = {}
self.state: dict = {
"players": [],
"score": {},
"status": "waiting"
}
async def add_player(self, player_id: int, websocket: WebSocket):
self.players[player_id] = {"websocket": websocket, "score": 0}
self.state["players"].append(player_id)
# 广播玩家加入
await self.broadcast({
"type": "player_joined",
"player_id": player_id,
"players": list(self.players.keys())
})
# 检查是否开始游戏
if len(self.players) >= 2:
await self.start_game()
async def start_game(self):
self.state["status"] = "playing"
await self.broadcast({
"type": "game_started",
"players": list(self.players.keys())
})
async def update_score(self, player_id: int, points: int):
if player_id in self.players:
self.players[player_id]["score"] += points
self.state["score"][player_id] = self.players[player_id]["score"]
await self.broadcast({
"type": "score_updated",
"player_id": player_id,
"score": self.state["score"]
})
async def broadcast(self, message: dict):
for player_id, player in self.players.items():
try:
await player["websocket"].send_text(json.dumps(message))
except:
pass
rooms: Dict[str, GameRoom] = {}
def get_room(room_id: str) -> GameRoom:
if room_id not in rooms:
rooms[room_id] = GameRoom(room_id)
return rooms[room_id]
@app.websocket("/ws/game/{room_id}/{player_id}")
async def game_endpoint(
websocket: WebSocket,
room_id: str,
player_id: int
):
room = get_room(room_id)
await room.add_player(player_id, websocket)
try:
while True:
data = await websocket.receive_text()
message = json.loads(data)
if message["type"] == "move":
# 处理玩家移动
await room.broadcast({
"type": "player_moved",
"player_id": player_id,
"position": message["position"]
})
elif message["type"] == "action":
# 处理玩家动作
await room.broadcast({
"type": "player_action",
"player_id": player_id,
"action": message["action"]
})
except Exception as e:
print(f"Error: {e}")
finally:
room.players.pop(player_id, None)
await room.broadcast({
"type": "player_left",
"player_id": player_id,
"players": list(room.players.keys())
})

六、性能优化

6.1 连接管理

import asyncio
from contextlib import asynccontextmanager
class ConnectionPool:
def __init__(self, max_connections: int = 10000):
self.max_connections = max_connections
self.connections: Dict[str, WebSocket] = {}
self.semaphore = asyncio.Semaphore(max_connections)
async def connect(self, connection_id: str, websocket: WebSocket):
await self.semaphore.acquire()
self.connections[connection_id] = websocket
def disconnect(self, connection_id: str):
self.connections.pop(connection_id, None)
self.semaphore.release()
async def __aenter__(self):
await self.semaphore.acquire()
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
self.semaphore.release()
使用
pool = ConnectionPool(max_connections=1000)
@app.websocket("/ws/{connection_id}")
async def websocket_endpoint(websocket: WebSocket, connection_id: str):
await pool.connect(connection_id, websocket)
try:
while True:
data = await websocket.receive_text()
await websocket.send_text(f"Echo: {data}")
except:
pass
finally:
pool.disconnect(connection_id)

6.2 消息队列

import asyncio
from collections import deque
class MessageQueue:
def __init__(self, max_size: int = 1000):
self.queue = deque(maxlen=max_size)
self.subscribers: List[asyncio.Queue] = []
async def publish(self, message: dict):
self.queue.append(message)
for subscriber in self.subscribers:
try:
await subscriber.put(message)
except asyncio.QueueFull:
pass
async def subscribe(self) -> asyncio.Queue:
queue = asyncio.Queue(maxsize=100)
self.subscribers.append(queue)
# 发送历史消息
for msg in self.queue:
await queue.put(msg)
return queue
def unsubscribe(self, queue: asyncio.Queue):
if queue in self.subscribers:
self.subscribers.remove(queue)
使用
message_queue = MessageQueue()
@app.websocket("/ws/stream")
async def stream_endpoint(websocket: WebSocket):
queue = await message_queue.subscribe()
try:
while True:
message = await queue.get()
await websocket.send_text(json.dumps(message))
except:
message_queue.unsubscribe(queue)

七、实战案例

7.1 完整实时系统

from fastapi import FastAPI, WebSocket, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
from typing import Dict, List, Optional
from datetime import datetime
import json
import asyncio
app = FastAPI()
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
class Message(BaseModel):
type: str
content: str
sender_id: int
timestamp: datetime = None
class Room(BaseModel):
id: str
name: str
owner_id: int
created_at: datetime
class User(BaseModel):
id: int
username: str
avatar: str | None = None
class RealTimeSystem:
def __init__(self):
self.rooms: Dict[str, Dict[int, WebSocket]] = {}
self.users: Dict[int, dict] = {}
self.message_history: Dict[str, List[Message]] = {}
async def join_room(self, room_id: str, user_id: int, websocket: WebSocket):
if room_id not in self.rooms:
self.rooms[room_id] = {}
self.message_history[room_id] = []
self.rooms[room_id][user_id] = websocket
# 通知其他人
await self.broadcast(room_id, {
"type": "user_joined",
"user_id": user_id,
"timestamp": datetime.utcnow().isoformat()
}, exclude=user_id)
def leave_room(self, room_id: str, user_id: int):
if room_id in self.rooms:
self.rooms[room_id].pop(user_id, None)
if not self.rooms[room_id]:
del self.rooms[room_id]
del self.message_history[room_id]
async def send_message(self, room_id: str, message: Message):
if room_id not in self.rooms:
raise HTTPException(status_code=404, detail="Room not found")
message.timestamp = datetime.utcnow()
self.message_history[room_id].append(message)
await self.broadcast(room_id, {
"type": "message",
"message": message.model_dump()
})
async def broadcast(self, room_id: str, data: dict, exclude: int | None = None):
if room_id not in self.rooms:
return
for user_id, websocket in self.rooms[room_id].items():
if user_id != exclude:
try:
await websocket.send_text(json.dumps(data))
except:
pass
def get_history(self, room_id: str, limit: int = 50) -> List[dict]:
if room_id not in self.message_history:
return []
return [m.model_dump() for m in self.message_history[room_id][-limit:]]
system = RealTimeSystem()
@app.websocket("/ws/{room_id}/{user_id}")
async def websocket_endpoint(
websocket: WebSocket,
room_id: str,
user_id: int
):
await websocket.accept()
await system.join_room(room_id, user_id, websocket)
# 发送历史消息
history = system.get_history(room_id)
await websocket.send_text(json.dumps({
"type": "history",
"messages": history
}))
try:
while True:
data = await websocket.receive_text()
message = json.loads(data)
if message["type"] == "message":
msg = Message(
type="message",
content=message["content"],
sender_id=user_id
)
await system.send_message(room_id, msg)
elif message["type"] == "typing":
await system.broadcast(room_id, {
"type": "typing",
"user_id": user_id
}, exclude=user_id)
except Exception as e:
print(f"Error: {e}")
finally:
system.leave_room(room_id, user_id)
@app.get("/api/rooms/{room_id}/history")
async def get_history(room_id: str, limit: int = 50):
return system.get_history(room_id, limit)

常见坑自查清单

坑	现象	自查方法	修复方案
连接泄露	内存增长	检查断开处理	确保清理连接
消息丢失	数据不一致	检查确认机制	添加ACK
重连风暴	服务器压力	检查重连逻辑	添加指数退避
心跳超时	连接断开	检查心跳间隔	定期发送心跳

结语

关键洞察

：

互动

1.你用WebSocket做过什么实时应用？

2.遇到过连接管理的问题吗？

3.觉得WebSocket比轮询好多少？

版本: V1.0 | 2026-05-26 | Python Web开发系列

📚 推荐阅读

📝 摘要：今天深入学习静态代码分析技术，这是安全审计的核心技能。从 Python AST 模块到检测模式设计，收获满满！

发布于 202603

01-Python 环境搭建与第一个脚本

发布于 202603

【优化】Python代码优化与调试技巧

发布于 202603

KEYWORDS

IL, Python, AI, 循环

💡 如果你觉得这篇文章有帮助，请点个在看，分享给更多需要的人！

📝 关注我，获取更多实用干货～

🤝 有问题欢迎评论区留言交流！

本文来自网友投稿或网络内容，如有侵犯您的权益请联系我们删除，联系邮箱：wyl860211@qq.com 。

Python Web开发:WebSocket实时通信

痛点：为什么你的应用总是需要不断轮询？

一、WebSocket基础

1.1 什么是WebSocket

1.2 FastAPI WebSocket

1.3 客户端连接

二、聊天室实现

2.1 房间管理

2.2 消息历史记录

房间存储

三、实时通知

3.1 通知系统

3.2 前端通知

四、实时数据同步

4.1 协作编辑

4.2 操作转换（OT）

五、游戏实时通信

5.1 多人游戏

六、性能优化

6.1 连接管理

使用

6.2 消息队列

使用

七、实战案例

7.1 完整实时系统

常见坑自查清单

结语

互动

最新文章

热门文章

随机文章

Python Web开发:WebSocket实时通信

痛点：为什么你的应用总是需要不断轮询？

一、WebSocket基础

1.1 什么是WebSocket

1.2 FastAPI WebSocket

1.3 客户端连接

二、聊天室实现

2.1 房间管理

2.2 消息历史记录

房间存储

三、实时通知

3.1 通知系统

3.2 前端通知

四、实时数据同步

4.1 协作编辑

4.2 操作转换（OT）

五、游戏实时通信

5.1 多人游戏

六、性能优化

6.1 连接管理

使用

6.2 消息队列

使用

七、实战案例

7.1 完整实时系统

常见坑自查清单

结语

互动

远洋课堂—Python AI基础:Matplotlib与Seaborn数据可视化

教程丨AI赋能Python-GEE遥感云大数据分析、可视化与Satellite Embedding创新

最新文章

热门文章

随机文章