当前位置：首页 > news >正文

python--杂识--20 sse与websocket区别

news 2025/7/17 10:34:42

Server-Sent Events (SSE) 和 WebSocket 是两种不同的技术，用于实现实时通信。它们各有优缺点，适用于不同的场景。以下是 SSE 和 WebSocket 的主要区别：

1. 通信方向

SSE：单向通信。服务器可以向客户端推送数据，但客户端不能向服务器发送数据。
WebSocket：双向通信。客户端和服务器都可以向对方发送数据。

2. 协议基础

SSE：基于 HTTP 协议。SSE 使用 HTTP 1.1 的 text/event-stream MIME 类型。
WebSocket：基于 TCP 协议。WebSocket 使用自己的协议，通过 ws:// 或 wss:// URL 方案。

3. 数据格式

SSE：文本协议，通常使用 UTF-8 编码。SSE 消息格式简单，每条消息以 \n 分隔。
WebSocket：二进制协议。WebSocket 可以传输二进制数据，也可以传输文本数据。

4. 连接管理

SSE：连接是 HTTP 连接，通常由浏览器自动管理。如果连接断开，浏览器会自动重新连接。
WebSocket：连接是持久的双向连接，需要手动管理连接的打开、关闭和错误处理。

5. 性能和资源消耗

SSE：相对轻量级，适用于简单的实时数据推送场景。
WebSocket：更复杂，但提供了更强大的功能，适用于需要双向通信的复杂场景。

6. 浏览器支持

SSE：现代浏览器普遍支持，但不如 WebSocket 普遍。
WebSocket：广泛支持，几乎所有现代浏览器都支持 WebSocket。

7. 适用场景

SSE：适用于服务器向客户端推送数据的场景，如实时通知、股票价格更新等。
WebSocket：适用于需要双向通信的场景，如实时聊天、在线游戏、协同编辑等。

示例代码

SSE 示例

服务端（使用 FastAPI）：

from fastapi import FastAPI, Request
from sse_starlette.sse import EventSourceResponse
import asyncioapp = FastAPI()@app.get('/stream')
def message_stream(request: Request):async def event_generator():for i in range(10):  # 发送10条消息if await request.is_disconnected():breakyield {"event": "message","id": f"message_id_{i}","retry": 15000,  # 重连时间，单位为毫秒"data": f"Message {i}"}await asyncio.sleep(1)  # 每秒发送一条消息return EventSourceResponse(event_generator())if __name__ == '__main__':import uvicornuvicorn.run(app, host="0.0.0.0", port=8000)

客户端（使用 JavaScript）：

import requestsdef sse_client(url):response = requests.get(url, stream=True)if response.status_code == 200:for line in response.iter_lines():if line:decoded_line = line.decode('utf-8')print(decoded_line)else:print(f"Failed to connect to SSE stream. Status code: {response.status_code}")if __name__ == "__main__":sse_client('http://127.0.0.1:8000/stream')

WebSocket 示例

服务端（使用 FastAPI）：

import uvicorn
from fastapi import FastAPI, WebSocket, WebSocketDisconnectapp = FastAPI()@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):await websocket.accept()try:while True:data = await websocket.receive_text()await websocket.send_text(f"Message text was: {data}")except WebSocketDisconnect:passif __name__ == '__main__':uvicorn.run(app, host="0.0.0.0", port=8000)

客户端（使用 JavaScript）：

import asyncio
import websocketsasync def connect_to_websocket():uri = "ws://127.0.0.1:8000/ws"async with websockets.connect(uri) as websocket:while True:message = input("Enter a message to send: ")await websocket.send(message)response = await websocket.recv()print(f"Received: {response}")if __name__ == "__main__":asyncio.run(connect_to_websocket())