1. 什么是 Server-Sent Events（SSE）
在《WebSocket 从原理到实践》[1]一文中介绍了客户端和服务端进行实时交互的几种方式，包括：
- 轮询（Polling）。这种方案的有点是简单，只需要不断发起 HTTP 请求即可，但是，这种方案最大的问题也正是上述的优点带来的，不断发起的 HTTP 请求会浪费带宽和服务器资源；
- 长轮询（Long Polling）。长轮询在发起 HTTP 请求时会挂起，这样就不会频繁发起请求，但是依然会出现连接断开重连的情况，如果频繁的断开重连，依然存在上述的浪费带宽和服务器资源的问题；
- WebSocket 从根本上避免了断开重连的问题，通过 HTTP 请求完成客户端和服务端的握手后，改由 TCP 连接，同时能够保持连接。
注意到，上面介绍的轮询和长轮询，都是客户端主动去请求服务器端获取最新的数据，服务器端是无法主动给客户端推送消息的。WebSocket 却像个全能选手，客户端和服务端能够互相推送信息，但是 WebSocket 需要单独的开发，是否有更简单的方式，实现服务器端主动向客户端推送信息的方案呢？答案是有的，那就是 SSE（Server-Sent Events），可以翻译成“服务端推送事件”。
SSE（Server-Sent Events）是一种可以主动从服务端推送消息的技术，客户端向服务端发起 HTTP 长连接，服务端返回 stream 响应流。客户端收到 stream 响应流并不会关闭连接而是一直等待服务端发送新的数据流。SSE 的具体过程如下图所示：
2. SSE 的基本原理
2.1. SSE 的工作原理
SSE 分为两个部分，第一个部分是建立连接，第二个部分是发送信息。
- 建立连接。由客户端向服务端发起一个标准的 HTTP 请求开启 SSE 会话，与普通的 HTTP 不同的是，在 HTTP 请求的 header 部分包含了字段
Accept: text/event-stream，这也就相当于告诉服务端，期望接收 SSE 消息流；当服务端接收到信息后，便知道这是个 SSE 的请求，便向客户端返回一个响应头，响应头中只包含字段：Content-Type: text/event-stream。这样便使得客户端和服务端建立了连接。 - 发送信息。客户端和服务端建立连接后，服务端就可以向客户端不断发送信息了。
2.2. 基于 FastAPI 的 SSE 实现
SSE 代码的底层就是 HTTP 服务，因此，服务端就构建一个 HTTP 服务即可，代码如下：
import uvicorn
import asyncio
from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import StreamingResponse
app = FastAPI()
app.add_middleware(
CORSMiddleware,
allow_origins=["*"], # 允许所有来源（生产环境请换成具体域名，如 ["http://localhost:3000"]）
allow_credentials=True,
allow_methods=["*"], # 允许所有方法（GET, POST, OPTIONS 等）
allow_headers=["*"], # 允许所有请求头（包括 Accept, Content-Type 等）
)
@app.post("/test_sse")
async def run_sse(request: Request):
print(f"request: {request.headers}")
return StreamingResponse(generate(), media_type="text/event-stream")
async def generate():
for s in "Hello World":
yield f"data: {s}\n\n"
await asyncio.sleep(0.1)
if __name__ == "__main__":
uvicorn.run('main:app', host='0.0.0.0', port=8001)
注意：在返回时用 StreamingResponse，表示服务器不等所有数据生成完，只要拿到第一个 yield 的数据块，就立刻先发送 HTTP 响应头（状态码 200、Content-Type 等），然后持续往网络管道里塞后续 yield 的数据块，具体过程如下：
- 立即返回响应头：浏览器收到状态码 200 和
Content-Type，连接保持打开（Keep-Alive）。 - 循环调用生成器：不断执行
generate()里的代码，每遇到一个yield，就把这个值打包成一个 HTTP 数据块（Chunk）发送给客户端。 - 直到生成器结束：当
generate()函数执行完毕（或遇到return），Uvicorn 发送一个空块（done: true），告诉浏览器“数据传完了”，连接关闭。
写一个简单的客户端，代码如下：
<!DOCTYPE html>
<html lang="zh-CN">
<head>
<meta charset="UTF-8">
<title>SSE 测试</title>
</head>
<body>
<div class="send_area">
<span>发送信息：<input type="text" id="input_a" /></span>
<button onclick="sendMsg()">发送</button>
</div>
<div class="receive_area">
<textarea id="receive_a"></textarea>
</div>
<script>
async function sendMsg() {
const textarea = document.getElementById("receive_a");
textarea.value = '';
const response = await fetch('http://127.0.0.1:8001/test_sse', {
method: 'POST',
headers: {
'Content-Type': 'application/json',
'Accept': 'text/event-stream'
},
body: JSON.stringify({ userId: 123, prompt: "hello" }) // 你的POST数据
});
const reader = response.body.getReader();
const decoder = new TextDecoder();
let buffer = ''; // 未完成行的缓存
let currentMessageData = []; // 当前消息的 data 行内容（可能有多个 data:）
let fullText = ''
while (true) {
const { value, done } = await reader.read();
console.log(done)
if (done) break;
const chunk = decoder.decode(value);
console.log('收到数据:', chunk);
// 解析
buffer += chunk;
const lines = buffer.split('\n');
buffer = lines.pop() || '';
for (const line of lines) {
if (line.startsWith('data: ')) {
const content = line.slice(6); // 去掉 "data: "
currentMessageData.push(content);
}
// 如果是空行（\n），表示一条消息结束
else if (line === '') {
if (currentMessageData.length > 0) {
const message = currentMessageData.join('\n');
fullText += message;
textarea.value = fullText;
currentMessageData = [];
}
}
}
}
}
</script>
</body>
</html>
这样就能在 textarea 中不断接收到来自服务端的数据，是不是有点像 LLM 的打字机输出效果。至此，我们实现了简单的 SSE 通信。