SSE：被低估的实时推送“轻骑兵”
在现代 Web 开发的江湖里，提到“实时通信”，大家往往脱口而出：“上 WebSocket！”仿佛它是解决一切延迟问题的银弹。
但你是否想过，如果你只是想让服务器给客户端发个通知、推个进度条，或者让 AI 像打字机一样逐字蹦出回答，真的需要动用 WebSocket 这种“重型武器”吗？
今天，我们要为一位被长期忽视的“轻骑兵”正名——SSE (Server-Sent Events)。它简单、优雅，且基于你最熟悉的 HTTP 协议。
🎭 场景模拟：餐厅里的故事
为了理解 SSE 和 WebSocket 的区别，我们走进一家餐厅。
1. 普通 HTTP：传统的“点餐模式”
你（客户端）走到柜台（服务器）：“有新品吗？”
服务员：“没有。”（连接断开）
过了 5 分钟，你又跑过去：“有新品吗？”
服务员：“还是没有。”（连接断开）
…
痛点：你累得半死（带宽浪费），服务员烦得要命（服务器负载高），而且新品出炉了你还要等下一轮询问才能知道（延迟高）。这就是轮询 (Polling)。
2. WebSocket：VIP 包厢的“对讲机”
你和服务员各拿一个对讲机。
你：“喂喂，听得到吗？”
服务员：“听得到。随时待命。”
一旦厨房有新菜，服务员立刻喊：“新菜出炉！”
你想加水，也立刻喊：“加杯水！”
特点：双向即时沟通，极其高效。
代价：你需要维护这套对讲机系统（复杂的握手、心跳、断线重连逻辑）。如果网络不好，对讲机坏了，你还得自己想办法重新配对。
3. SSE：餐厅的“广播喇叭” 📢
你坐在座位上，不用跑腿，也不用拿对讲机。
餐厅安装了一个广播喇叭（SSE 连接）。
服务员只需要在喇叭里喊：“今日特价：红烧肉！”、“3号桌的菜好了！”
你只管听着就行。
如果你想点菜（上行数据）？没关系，你举手示意一下（发送一个普通的 HTTP POST 请求）即可。
特点：
- 单向推送：服务员只管喊，你只管听。
- 极简：不需要对讲机配对，只要有耳朵（浏览器原生支持）就能听。
- 自动重连：如果喇叭信号中断了一下，恢复后它会自动接着播，不用你操心。
⚔️ SSE vs WebSocket：谁是你的菜？
| 特性 | SSE (广播喇叭) | WebSocket (对讲机) |
|---|---|---|
| 通信方向 | 单向 (服务器 → 客户端) | 双向 (互相喊话) |
| 底层协议 | HTTP/HTTPS (老熟人，防火墙不拦) | ws:// (新协议，偶尔被拦截) |
| 数据类型 | 仅文本 (JSON, 字符串) | 文本 + 二进制 (图片, 音频流) |
| 开发难度 | ⭐ (极简) | ⭐⭐⭐ (需处理心跳、重连) |
| 自动重连 | ✅ 内置 (浏览器自带“复活甲”) | ❌ 无 (得自己写代码复活) |
| 最佳场景 | AI 流式输出、股票行情、通知、日志 | 在线游戏、即时聊天、协同编辑 |
一句话总结：
- 如果主要是“服务器说，客户端听”，选 SSE。
- 如果双方要“频繁互喷”（如打游戏、协作画图），选 WebSocket。
🛠️ 实战：用 Python (FastAPI) 搭建你的“广播站”
SSE 的魅力在于简单。在 FastAPI 中，实现一个 SSE 接口比写一个普通 Hello World 难不了多少。
1. 服务端：搭建广播喇叭
# sse_server.py
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import asyncio
import time
import json
app = FastAPI()
async def event_generator():
"""
这是一个永动机生成器，只要客户端连着，它就不停说话。
"""
count = 0
try:
while True:
count += 1
# 1. 准备数据
data = {
"time": time.strftime("%H:%M:%S"),
"msg": f"第 {count} 次广播",
"status": "online"
}
# 2. 按照 SSE 格式打包： "data: 内容\n\n"
# 注意：两个换行符是必须的，代表一条消息结束
yield f"data: {json.dumps(data)}\n\n"
# 3. 心跳包：每 5 秒发个注释，防止中间商（Nginx）切断连接
if count % 5 == 0:
yield ": heartbeat\n\n"
await asyncio.sleep(1) # 模拟每秒产生一条数据
except asyncio.CancelledError:
print("📻 客户端断开连接，广播停止。")
raise
@app.get("/broadcast")
def start_broadcast():
return StreamingResponse(
event_generator(),
media_type="text/event-stream", # 告诉浏览器：这是流式数据
headers={
"Cache-Control": "no-cache", # 别缓存，我要实时的
"Connection": "keep-alive" # 保持连接不断
}
)
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)2. 客户端：竖起耳朵听
🌐 前端 (JavaScript) – 简单到哭
浏览器原生支持 EventSource，三行代码搞定：
const radio = new EventSource("http://localhost:8000/broadcast");
radio.onmessage = (event) => {
const data = JSON.parse(event.data);
console.log(`📢 收到广播: ${data.msg} at ${data.time}`);
// 在这里更新你的 UI，比如把文字追加到页面上
};
radio.onerror = () => {
console.log("📡 信号中断，浏览器正在尝试自动重连...");
};🐍 后端 (Python) – 模拟监听者
如果你想在另一个 Python 服务里监听这个广播：
# sse_client.py
import requests
import json
def listen_to_radio(url):
print("📻 正在调频...")
with requests.get(url, stream=True) as r:
for line in r.iter_lines():
if line:
decoded = line.decode('utf-8')
if decoded.startswith("data:"):
data = json.loads(decoded[5:])
print(f"📩 收到: {data}")
listen_to_radio("http://localhost:8000/broadcast")💡 避坑指南：老司机的经验之谈
虽然 SSE 很简单，但有几个坑容易踩：
1. Nginx 是个“急性子”
默认情况下，Nginx 会缓冲响应。这意味着服务器发了 10 条消息，Nginx 可能攒够了一包才发给客户端，导致“实时性”丧失。
解药：在 Nginx 配置里加上：
proxy_buffering off;
proxy_cache off;2. 沉默是金？不，沉默是死
如果服务器半天不说话，中间的防火墙或负载均衡器会认为：“这连接是不是死了？”然后咔嚓切断。
解药：发送心跳包。
yield ": heartbeat\n\n" 以冒号开头的行是注释，客户端会忽略它，但连接会被激活，告诉中间设备：“我还活着！”
3. 只能传文本？
是的，SSE 只能传 UTF-8 文本。如果你想传图片或二进制视频流，请出门左转找 WebSocket。但对于 JSON 数据来说，这完全不是问题。
🏁 结语
SSE 就像是一位低调的实干家。它没有 WebSocket 那么炫技，也不像 HTTP 轮询那么笨重。它在“实时性”和“复杂性”之间找到了完美的平衡点。
下次当你需要实现：
- ✨ AI 聊天的打字机效果
- 📈 股票价格的实时跳动
- 🔔 网页右上角的消息红点
- 📊 后台任务的进度条
请记得，你不需要架设复杂的 WebSocket 服务器。打开你的 HTTP 接口，加上 text/event-stream，SSE 就会为你工作。
简单，才是最高级的复杂。
