扫描二维码
关注或者微信搜一搜：编程智域 前端至全栈交流与成长
发现1000+提升效率与开发的AI工具和实用程序：https://tools.cmdragon.cn/
通过对比传统HTTP协议与WebSocket协议的核心特性，我们可以从以下维度理解二者的差异：
| 对比维度 | HTTP 1.1 | WebSocket |
|---|---|---|
| 通信模式 | 请求-响应模式（半双工） | 全双工双向通信 |
| 连接持久性 | 短连接（默认关闭） | 长连接（保持激活状态） |
| 数据传输效率 | 每次请求需携带完整Header | 初始握手后仅传输有效数据 |
| 数据格式 | 基于文本（支持二进制分帧） | 原生支持二进制帧和文本帧 |
| 服务端推送能力 | 依赖长轮询/SSE实现 | 原生支持服务端主动推送 |
类比理解：HTTP协议如同收发纸质信件（每次通信需重新建立连接），WebSocket则像是电话通话（建立连接后可随时双向对话）
FastAPI通过简洁的装饰器语法支持WebSocket端点定义：
from fastapi import FastAPI, WebSocket
from pydantic import BaseModel
app = FastAPI()
class MessageModel(BaseModel):
content: str
type: str = "text"
@app.websocket("/ws-chat")
async def websocket_endpoint(websocket: WebSocket):
await websocket.accept()
try:
while True:
data = await websocket.receive_json()
message = MessageModel(**data) # Pydantic数据验证
await websocket.send_json({"status": "received", "data": message.dict()})
except Exception as e:
await websocket.close(code=1011, reason=str(e))
（使用环境：Python 3.8+, fastapi0.109.0, uvicorn0.27.0, pydantic==2.6.4）
# 存储活跃连接的全局字典
active_connections = {}
@app.websocket("/chat/{room_id}")
async def chat_room(websocket: WebSocket, room_id: str):
await websocket.accept()
active_connections[websocket] = room_id
try:
while True:
message = await websocket.receive_text()
# 广播消息到同房间用户
for conn, r_id in active_connections.items():
if r_id == room_id:
await conn.send_text(f"Room {room_id}: {message}")
finally:
del active_connections[websocket]
from datetime import datetime
@app.websocket("/system-monitor")
async def monitor_feed(websocket: WebSocket):
await websocket.accept()
while True:
# 模拟获取系统指标
metrics = {
"timestamp": datetime.now().isoformat(),
"cpu_load": os.getloadavg()[0],
"memory_usage": psutil.virtual_memory().percent
}
await websocket.send_json(metrics)
await asyncio.sleep(1) # 每秒推送一次
import redis
from fastapi import BackgroundTasks
redis_conn = redis.Redis(host='localhost', port=6379)
async def notification_listener(websocket: WebSocket, user_id: str):
pubsub = redis_conn.pubsub()
await pubsub.subscribe(f"notifications:{user_id}")
while True:
message = await pubsub.get_message(ignore_subscribe_messages=True)
if message:
await websocket.send_text(message['data'])
@app.websocket("/notifications")
async def notification_endpoint(websocket: WebSocket, user_id: str):
await websocket.accept()
background_tasks = BackgroundTasks()
background_tasks.add_task(notification_listener, websocket, user_id)
await background_tasks()
Q1：WebSocket连接建立过程中，客户端首先发送的特殊HTTP Header是什么？
A) Upgrade: websocket
B) Connection: keep-alive
C) Accept: text/websocket
D) Protocol: ws
正确答案：A
解析：WebSocket握手阶段要求客户端发送Upgrade: websocket头，表示请求协议升级
错误现象：
WebSocket连接失败 400 Bad Request
排查步骤：
ws://或wss://协议前缀proxy_http_version 1.1;
proxy_set_header Upgrade $http_upgrade;
proxy_set_header Connection "upgrade";
余下文章内容请点击跳转至 个人博客页面 或者 扫码关注或者微信搜一搜：编程智域 前端至全栈交流与成长
，阅读完整的文章：WebSocket与HTTP协议究竟有何不同？FastAPI如何让长连接变得如此简单？
登录查看全部
参与评论
手机查看
返回顶部