SSE（Server-Sent Events）协议学习指南

📚 适合碎片化学习，预计总学习时长：3-4小时

1. 基础理论

🎯 核心知识点

1.1 什么是 SSE？

SSE（Server-Sent Events） 是一种基于 HTTP 的单向通信协议，允许服务器主动向客户端推送数据。

┌─────────────┐                    ┌─────────────┐
│   客户端     │ ──── HTTP 请求 ────▶│   服务器     │
│  (Browser)  │                    │  (Server)   │
│             │ ◀──── 持续推送 ─────│             │
└─────────────┘    (Event Stream)  └─────────────┘

核心特点：

单向通信：服务器 → 客户端（客户端无法通过同一连接发送数据）
基于 HTTP：使用标准 HTTP 协议，无需特殊端口或协议升级
自动重连：浏览器原生支持断线自动重连
文本协议：数据以 UTF-8 文本格式传输

1.2 SSE 的工作原理

1.3 SSE 的适用场景

场景	说明	典型应用
实时通知	服务器主动推送状态变化	社交通知、系统告警
数据流	持续推送更新数据	股票行情、实时日志
进度更新	长任务进度反馈	文件上传进度、AI 生成进度
实时协作	多用户状态同步	在线文档协作状态

1.4 HTTP 响应头要求

http

HTTP/1.1 200 OK
Content-Type: text/event-stream    # 必须
Cache-Control: no-cache            # 禁用缓存
Connection: keep-alive             # 保持连接

1.5 SSE vs 传统轮询

方式	连接数	实时性	服务器压力	实现复杂度
短轮询	频繁新建	差（取决于轮询间隔）	高	低
长轮询	较少	中等	中等	中等
SSE	单连接	高（毫秒级）	低	低

✅ 实操任务 1：理解 SSE 请求

使用 curl 命令体验 SSE 连接：

bash

# 连接一个公开的 SSE 测试端点
curl -N -H "Accept: text/event-stream" https://sse.dev/test

# 或者使用本地测试（如果有服务运行）
curl -N http://localhost:8080/events

观察点：

响应头中的 Content-Type: text/event-stream
数据以 data: 开头，以 \n\n 结尾
连接保持打开状态，持续接收数据

2. 核心语法

🎯 核心知识点

2.1 事件流格式（Event Stream Format）

SSE 使用简单的文本格式，每个事件由以下字段组成：

[field]: [value]\n
[field]: [value]\n
\n

四个标准字段：

字段	作用	示例
`data`	消息内容（必需）	`data: Hello World`
`event`	事件类型（可选）	`event: update`
`id`	事件 ID（可选）	`id: 12345`
`retry`	重连间隔毫秒（可选）	`retry: 3000`

2.2 消息格式详解

基本消息：

data: 这是一条简单消息

多行消息：

data: 第一行
data: 第二行
data: 第三行

客户端接收后会自动用 \n 连接成：第一行\n第二行\n第三行

JSON 消息：

data: {"user": "张三", "action": "login", "time": 1699999999}

带事件类型：

event: user_login
data: {"user": "张三"}

event: user_logout
data: {"user": "李四"}

带 ID（支持断线重连）：

id: 1001
data: 消息1

id: 1002
data: 消息2

2.3 完整事件示例

:这是注释，会被忽略

retry: 5000

id: msg-001
event: notification
data: {"title": "新消息", "content": "您有一条新通知"}

id: msg-002
event: heartbeat
data: ping

2.4 特殊规则

规则	说明
空行分隔	两个事件之间必须用空行（`\n\n`）分隔
注释	以 `:` 开头的行是注释，常用于心跳保活
UTF-8	必须使用 UTF-8 编码
无 BOM	不能包含 UTF-8 BOM

2.5 客户端 EventSource API

javascript

// 创建 SSE 连接
const evtSource = new EventSource('/events');

// 连接状态
// evtSource.readyState: 0=CONNECTING, 1=OPEN, 2=CLOSED

// 监听默认消息（无 event 字段的消息）
evtSource.onmessage = (event) => {
    console.log('收到消息:', event.data);
    console.log('事件ID:', event.lastEventId);
};

// 监听特定事件类型
evtSource.addEventListener('user_login', (event) => {
    const data = JSON.parse(event.data);
    console.log('用户登录:', data.user);
});

// 监听连接打开
evtSource.onopen = () => {
    console.log('SSE 连接已建立');
};

// 监听错误
evtSource.onerror = (error) => {
    console.error('SSE 错误:', error);
    if (evtSource.readyState === EventSource.CLOSED) {
        console.log('连接已关闭');
    }
};

// 关闭连接
evtSource.close();

✅ 实操任务 2：解析 SSE 消息格式

手动解析以下 SSE 数据流，写出客户端会触发的事件：

:heartbeat

id: 1
event: chat
data: {"from": "Alice", "msg": "Hi"}

id: 2
data: 这是默认消息

id: 3
event: chat
data: {"from": "Bob", "msg": "Hello"}
data: {"from": "Bob", "msg": "How are you?"}

预期答案：

注释被忽略
触发 chat 事件，data = {"from": "Alice", "msg": "Hi"}
触发 message 事件（默认），data = 这是默认消息
触发 chat 事件，data = {"from": "Bob", "msg": "Hello"}\n{"from": "Bob", "msg": "How are you?"}

3. 示例代码

🎯 核心知识点

3.1 Python 服务端（Flask）

python

# sse_server.py
from flask import Flask, Response
import time
import json

app = Flask(__name__)

def generate_events():
    """生成 SSE 事件流"""
    event_id = 0
    while True:
        event_id += 1
        
        # 构造 SSE 消息
        data = json.dumps({
            "id": event_id,
            "time": time.strftime("%H:%M:%S"),
            "message": f"这是第 {event_id} 条消息"
        })
        
        # SSE 格式：id + event + data + 空行
        yield f"id: {event_id}\n"
        yield f"event: update\n"
        yield f"data: {data}\n"
        yield "\n"  # 重要：空行表示事件结束
        
        time.sleep(2)  # 每2秒推送一次

@app.route('/events')
def sse_endpoint():
    """SSE 端点"""
    return Response(
        generate_events(),
        mimetype='text/event-stream',
        headers={
            'Cache-Control': 'no-cache',
            'Connection': 'keep-alive',
            'X-Accel-Buffering': 'no'  # 禁用 Nginx 缓冲
        }
    )

@app.route('/')
def index():
    """测试页面"""
    return '''
    <!DOCTYPE html>
    <html>
    <head><title>SSE Demo</title></head>
    <body>
        <h1>SSE 实时消息</h1>
        <div id="messages"></div>
        <script>
            const evtSource = new EventSource('/events');
            const messagesDiv = document.getElementById('messages');
            
            evtSource.addEventListener('update', (e) => {
                const data = JSON.parse(e.data);
                messagesDiv.innerHTML += `<p>[${data.time}] ${data.message}</p>`;
            });
            
            evtSource.onerror = () => {
                messagesDiv.innerHTML += '<p style="color:red">连接断开，正在重连...</p>';
            };
        </script>
    </body>
    </html>
    '''

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=8080, threaded=True)

3.2 Python 服务端（FastAPI + 异步）

python

# sse_fastapi.py
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from sse_starlette.sse import EventSourceResponse
import asyncio
import json

app = FastAPI()

async def event_generator():
    """异步事件生成器"""
    event_id = 0
    while True:
        event_id += 1
        data = {
            "id": event_id,
            "message": f"异步消息 #{event_id}"
        }
        
        yield {
            "event": "update",
            "id": str(event_id),
            "data": json.dumps(data)
        }
        
        await asyncio.sleep(1)

@app.get('/events')
async def sse_endpoint():
    """SSE 端点"""
    return EventSourceResponse(event_generator())

# 安装依赖：pip install fastapi uvicorn sse-starlette
# 运行：uvicorn sse_fastapi:app --host 0.0.0.0 --port 8080

3.3 Go 服务端

// sse_server.go
package main

import (
    "fmt"
    "net/http"
    "time"
)

func sseHandler(w http.ResponseWriter, r *http.Request) {
    // 设置 SSE 响应头
    w.Header().Set("Content-Type", "text/event-stream")
    w.Header().Set("Cache-Control", "no-cache")
    w.Header().Set("Connection", "keep-alive")
    
    // 获取 Flusher 接口
    flusher, ok := w.(http.Flusher)
    if !ok {
        http.Error(w, "Streaming not supported", http.StatusInternalServerError)
        return
    }
    
    eventID := 0
    for {
        eventID++
        
        // 发送 SSE 事件
        fmt.Fprintf(w, "id: %d\n", eventID)
        fmt.Fprintf(w, "event: update\n")
        fmt.Fprintf(w, "data: {\"id\": %d, \"time\": \"%s\"}\n\n", 
            eventID, time.Now().Format("15:04:05"))
        
        flusher.Flush()  // 立即发送
        
        time.Sleep(2 * time.Second)
    }
}

func main() {
    http.HandleFunc("/events", sseHandler)
    fmt.Println("SSE Server running on :8080")
    http.ListenAndServe(":8080", nil)
}

3.4 JavaScript 客户端（完整示例）

html

<!-- sse_client.html -->
<!DOCTYPE html>
<html lang="zh-CN">
<head>
    <meta charset="UTF-8">
    <title>SSE 客户端示例</title>
    <style>
        body { font-family: Arial, sans-serif; max-width: 800px; margin: 50px auto; }
        #status { padding: 10px; margin: 10px 0; border-radius: 5px; }
        .connected { background: #d4edda; color: #155724; }
        .disconnected { background: #f8d7da; color: #721c24; }
        .connecting { background: #fff3cd; color: #856404; }
        #messages { border: 1px solid #ddd; padding: 15px; height: 300px; overflow-y: auto; }
        .message { padding: 8px; margin: 5px 0; background: #f8f9fa; border-radius: 3px; }
        button { padding: 10px 20px; margin: 5px; cursor: pointer; }
    </style>
</head>
<body>
    <h1>🔴 SSE 实时消息演示</h1>
    
    <div id="status" class="disconnected">未连接</div>
    
    <button onclick="connect()">连接</button>
    <button onclick="disconnect()">断开</button>
    
    <h3>消息列表：</h3>
    <div id="messages"></div>
    
    <script>
        let eventSource = null;
        const statusEl = document.getElementById('status');
        const messagesEl = document.getElementById('messages');
        
        function updateStatus(status, text) {
            statusEl.className = status;
            statusEl.textContent = text;
        }
        
        function addMessage(type, data) {
            const div = document.createElement('div');
            div.className = 'message';
            div.innerHTML = `<strong>[${type}]</strong> ${JSON.stringify(data)}`;
            messagesEl.appendChild(div);
            messagesEl.scrollTop = messagesEl.scrollHeight;
        }
        
        function connect() {
            if (eventSource) {
                eventSource.close();
            }
            
            updateStatus('connecting', '正在连接...');
            
            // 创建 SSE 连接
            eventSource = new EventSource('/events');
            
            // 连接成功
            eventSource.onopen = () => {
                updateStatus('connected', '✅ 已连接');
                addMessage('系统', '连接已建立');
            };
            
            // 监听默认消息
            eventSource.onmessage = (event) => {
                addMessage('默认', event.data);
            };
            
            // 监听自定义事件
            eventSource.addEventListener('update', (event) => {
                const data = JSON.parse(event.data);
                addMessage('update', data);
            });
            
            eventSource.addEventListener('notification', (event) => {
                const data = JSON.parse(event.data);
                addMessage('通知', data);
                // 可以触发浏览器通知
                if (Notification.permission === 'granted') {
                    new Notification('新通知', { body: data.message });
                }
            });
            
            // 错误处理
            eventSource.onerror = (error) => {
                if (eventSource.readyState === EventSource.CONNECTING) {
                    updateStatus('connecting', '🔄 重新连接中...');
                    addMessage('系统', '连接断开，正在自动重连...');
                } else if (eventSource.readyState === EventSource.CLOSED) {
                    updateStatus('disconnected', '❌ 连接已关闭');
                    addMessage('系统', '连接已关闭');
                }
            };
        }
        
        function disconnect() {
            if (eventSource) {
                eventSource.close();
                eventSource = null;
                updateStatus('disconnected', '已断开');
                addMessage('系统', '主动断开连接');
            }
        }
        
        // 页面关闭时清理连接
        window.onbeforeunload = () => {
            if (eventSource) {
                eventSource.close();
            }
        };
    </script>
</body>
</html>

3.5 Python 客户端

python

# sse_client.py
import requests
import sseclient  # pip install sseclient-py

def listen_sse(url):
    """监听 SSE 事件流"""
    headers = {'Accept': 'text/event-stream'}
    
    response = requests.get(url, headers=headers, stream=True)
    client = sseclient.SSEClient(response)
    
    print(f"已连接到 {url}")
    
    for event in client.events():
        print(f"事件类型: {event.event}")
        print(f"事件ID: {event.id}")
        print(f"数据: {event.data}")
        print("-" * 40)

if __name__ == '__main__':
    listen_sse('http://localhost:8080/events')

✅ 实操任务 3：搭建 SSE 服务

复制上面的 Flask 示例代码
安装依赖：pip install flask
运行服务：python sse_server.py
浏览器访问：http://localhost:8080
观察实时消息推送

进阶任务：修改代码，实现以下功能：

添加一个 heartbeat 事件，每 10 秒发送一次
添加一个 alert 事件，当消息数量达到 10 的倍数时触发

4. 常见问题

🎯 核心知识点

4.1 连接数限制问题

问题描述：浏览器对同一域名的 HTTP/1.1 连接数有限制（通常 6 个）

┌──────────────────────────────────────────┐
│ 浏览器同域名最大连接数限制（HTTP/1.1）      │
├──────────────────────────────────────────┤
│ Chrome:     6 个                          │
│ Firefox:    6 个                          │
│ Safari:     6 个                          │
│ Edge:       6 个                          │
└──────────────────────────────────────────┘

影响：如果用户打开多个标签页，每个页面都建立 SSE 连接，可能很快耗尽连接数

解决方案：

javascript

// 方案1：使用 SharedWorker（多标签页共享连接）
// shared_worker.js
const clients = [];
let eventSource = null;

self.onconnect = (e) => {
    const port = e.ports[0];
    clients.push(port);
    
    if (!eventSource) {
        eventSource = new EventSource('/events');
        eventSource.onmessage = (event) => {
            clients.forEach(client => {
                client.postMessage(event.data);
            });
        };
    }
    
    port.onmessage = (e) => {
        if (e.data === 'close') {
            const index = clients.indexOf(port);
            if (index > -1) clients.splice(index, 1);
        }
    };
};

// 方案2：使用 HTTP/2（多路复用，无连接数限制）
// 需要服务器支持 HTTP/2

// 方案3：使用 BroadcastChannel（同源页面广播）
// 主页面
const bc = new BroadcastChannel('sse_channel');
const evtSource = new EventSource('/events');
evtSource.onmessage = (e) => bc.postMessage(e.data);

// 其他页面
const bc = new BroadcastChannel('sse_channel');
bc.onmessage = (e) => console.log('收到广播:', e.data);

4.2 断线重连与消息丢失

问题描述：网络断开期间，服务器发送的消息会丢失

解决方案：

python

# 服务端：支持 Last-Event-ID
from flask import Flask, Response, request

# 消息存储（生产环境用 Redis）
message_history = []

@app.route('/events')
def sse_endpoint():
    # 获取客户端上次收到的事件ID
    last_event_id = request.headers.get('Last-Event-ID', '0')
    last_id = int(last_event_id) if last_event_id.isdigit() else 0
    
    def generate():
        # 先发送丢失的历史消息
        for msg in message_history:
            if msg['id'] > last_id:
                yield f"id: {msg['id']}\n"
                yield f"data: {msg['data']}\n\n"
        
        # 然后发送新消息
        # ...
    
    return Response(generate(), mimetype='text/event-stream')

javascript

// 客户端会自动在重连时带上 Last-Event-ID 请求头
// 无需额外代码，EventSource 原生支持

4.3 代理/负载均衡器缓冲问题

问题描述：Nginx、CDN 等可能会缓冲响应，导致消息延迟

Nginx 配置：

nginx

location /events {
    proxy_pass http://backend;
    
    # SSE 必需配置
    proxy_http_version 1.1;
    proxy_set_header Connection '';
    proxy_buffering off;           # 关闭代理缓冲
    proxy_cache off;               # 关闭缓存
    
    # 超时配置
    proxy_read_timeout 86400s;     # 长连接超时
    proxy_send_timeout 86400s;
    
    # 禁用 gzip（某些情况下会导致缓冲）
    gzip off;
}

代码层面：

python

# 添加响应头禁用缓冲
headers = {
    'X-Accel-Buffering': 'no',     # Nginx
    'Cache-Control': 'no-cache, no-transform',
}

4.4 跨域（CORS）问题

python

# Flask 服务端
from flask_cors import CORS

app = Flask(__name__)
CORS(app, resources={r"/events": {"origins": "*"}})

# 或手动添加响应头
@app.after_request
def add_cors_headers(response):
    response.headers['Access-Control-Allow-Origin'] = '*'
    response.headers['Access-Control-Allow-Credentials'] = 'true'
    return response

javascript

// 客户端：带凭证的跨域请求
const evtSource = new EventSource('/events', { 
    withCredentials: true  // 携带 Cookie
});

4.5 内存泄漏问题

javascript

// ❌ 错误：未清理事件监听器和连接
function initSSE() {
    const evtSource = new EventSource('/events');
    evtSource.onmessage = (e) => {
        document.getElementById('msg').innerHTML += e.data;
    };
}

// ✅ 正确：组件卸载时清理
class SSEComponent {
    constructor() {
        this.evtSource = null;
    }
    
    connect() {
        this.evtSource = new EventSource('/events');
        this.handleMessage = this.handleMessage.bind(this);
        this.evtSource.addEventListener('message', this.handleMessage);
    }
    
    handleMessage(e) {
        console.log(e.data);
    }
    
    disconnect() {
        if (this.evtSource) {
            this.evtSource.removeEventListener('message', this.handleMessage);
            this.evtSource.close();
            this.evtSource = null;
        }
    }
}

// React 中的清理
useEffect(() => {
    const evtSource = new EventSource('/events');
    evtSource.onmessage = (e) => setMessages(prev => [...prev, e.data]);
    
    return () => {
        evtSource.close();  // 组件卸载时关闭连接
    };
}, []);

✅ 实操任务 4：排查 SSE 问题

创建一个简单的 SSE 服务，然后故意制造以下问题并排查：

模拟 Nginx 缓冲：不设置 X-Accel-Buffering: no，观察消息是否延迟
模拟断线重连：停止服务器，观察客户端重连行为
检查连接数：在同一浏览器打开 7+ 个标签页，观察连接阻塞

5. 选型对比

🎯 核心知识点

5.1 SSE vs WebSocket vs 轮询

特性	SSE	WebSocket	长轮询	短轮询
通信方向	单向（服务器→客户端）	双向	单向	单向
协议	HTTP	ws:// / wss://	HTTP	HTTP
连接	持久连接	持久连接	短连接	短连接
浏览器支持	IE外全支持	全支持	全支持	全支持
自动重连	✅ 原生支持	❌ 需手动实现	❌	❌
二进制数据	❌ 仅文本	✅ 支持	✅ 支持	✅ 支持
防火墙友好	✅ 标准 HTTP	❌ 可能被阻止	✅	✅
服务器实现	简单	较复杂	简单	最简单
适用场景	通知、实时数据流	聊天、游戏	兼容性要求高	低频更新

5.2 决策流程图

5.3 何时选择 SSE？

✅ 推荐使用 SSE 的场景：

单向数据推送：
- 实时通知系统
- 股票/加密货币行情
- 社交媒体 Feed 更新
- 日志流监控
AI/LLM 流式输出：
- ChatGPT 类应用的打字机效果
- 代码生成进度
- 流式翻译
进度跟踪：
- 文件上传/下载进度
- 批处理任务状态
- CI/CD 构建日志
简单实时更新：
- 新闻推送
- 评论实时刷新
- 在线人数统计

❌ 不推荐使用 SSE 的场景：

需要双向通信：聊天室、多人游戏
需要传输二进制数据：音视频流、文件传输
需要支持 IE：使用长轮询或 WebSocket + polyfill
超高频消息（>100/s）：考虑 WebSocket

5.4 SSE 的优势

优势	说明
简单易用	基于 HTTP，无需特殊协议，服务端实现简单
原生重连	浏览器自动处理断线重连
事件ID	原生支持消息 ID，方便断点续传
防火墙友好	使用标准 HTTP 端口（80/443）
HTTP/2 兼容	在 HTTP/2 下效率更高
调试方便	纯文本协议，易于抓包分析

5.5 SSE 的局限

局限	解决方案
单向通信	另开 HTTP 请求发送数据
连接数限制	使用 HTTP/2 或 SharedWorker
仅支持文本	Base64 编码二进制数据
IE 不支持	使用 polyfill 或降级到轮询
无二进制帧	需要二进制时改用 WebSocket

✅ 实操任务 5：技术选型练习

针对以下场景，选择最合适的技术方案并说明理由：

场景	你的选择	理由
ChatGPT 类 AI 对话（流式输出）
多人协作白板
股票行情实时推送
网页版微信聊天
后台任务进度条

参考答案：

AI 对话 → SSE（单向推送，文本数据，简单易实现）
协作白板 → WebSocket（需要双向实时同步绘图数据）
股票行情 → SSE（单向推送，更新频率中等）
网页微信 → WebSocket（双向聊天，需要发送和接收消息）
进度条 → SSE（单向推送进度百分比）

6. 学习资源与工具

📚 权威学习资源

资源	链接	说明
MDN Web Docs	MDN SSE 文档	最权威的 Web API 文档
HTML Living Standard	WHATWG SSE 规范	官方规范文档
Stream Handbook	Stream Handbook	流式处理概念理解
SSE vs WebSocket	Web.dev 对比文章	Google 的技术选型指南
Real-time Web 技术	Real-time Web Technologies	Ably 的深度技术文章

🛠 常用调试工具

1. 浏览器开发者工具

Chrome DevTools:
Network 面板 → 筛选 "EventStream" 
→ 点击请求 → EventStream 标签页

![Chrome SSE Debug](示意图：Network → EventStream)

使用步骤：

打开 Chrome DevTools（F12）
切换到 Network 面板
在筛选器中选择 "Other" 或搜索 "EventStream"
点击 SSE 请求，查看 "EventStream" 标签页
实时查看收到的事件

2. curl 命令行

bash

# 基本连接
curl -N http://localhost:8080/events

# 带请求头
curl -N -H "Accept: text/event-stream" http://localhost:8080/events

# 模拟断线重连
curl -N -H "Last-Event-ID: 100" http://localhost:8080/events

# 显示响应头
curl -N -i http://localhost:8080/events

# 超时设置
curl -N --max-time 60 http://localhost:8080/events

3. httpie（更友好的 curl 替代品）

bash

# 安装
pip install httpie

# SSE 请求
http --stream GET localhost:8080/events Accept:text/event-stream

4. 在线 SSE 测试工具

工具	链接	用途
SSE.dev	https://sse.dev/	在线 SSE 测试服务
Postman	支持 SSE 请求	API 调试
Insomnia	支持 SSE 请求	API 调试

5. VS Code 插件

REST Client：支持 SSE 请求
Thunder Client：轻量级 API 测试

http

# .http 文件示例
GET http://localhost:8080/events
Accept: text/event-stream

📋 学习检查清单

完成本文档学习后，你应该能够：

[ ] 理解 SSE 的工作原理和适用场景
[ ] 手写 SSE 消息格式（data、event、id、retry）
[ ] 使用 Python/Go/Node.js 实现 SSE 服务端
[ ] 使用 EventSource API 实现客户端
[ ] 处理断线重连和消息丢失问题
[ ] 解决 Nginx 代理缓冲问题
[ ] 正确选择 SSE/WebSocket/轮询

🎯 综合实操项目

创建一个「实时系统监控面板」，要求：

后端（Python/Go）：
- 每秒推送 CPU、内存使用率
- 支持 metrics 和 alert 两种事件类型
- 当 CPU > 80% 时发送 alert 事件
前端：
- 使用 EventSource 接收数据
- 实时更新图表（可用 Chart.js）
- 弹出告警通知
进阶：
- 实现断线重连提示
- 使用 Last-Event-ID 恢复丢失消息
- 多标签页共享连接（SharedWorker）

📝 文档版本：v1.0
最后更新：2026-02-09
适用人群：Web 中级开发者

SSE（Server-Sent Events）协议学习指南 ​

目录 ​

1. 基础理论 ​

🎯 核心知识点 ​

1.1 什么是 SSE？ ​

1.2 SSE 的工作原理 ​

1.3 SSE 的适用场景 ​

1.4 HTTP 响应头要求 ​

1.5 SSE vs 传统轮询 ​

✅ 实操任务 1：理解 SSE 请求 ​

2. 核心语法 ​

🎯 核心知识点 ​

2.1 事件流格式（Event Stream Format） ​

2.2 消息格式详解 ​

2.3 完整事件示例 ​

2.4 特殊规则 ​

2.5 客户端 EventSource API ​

✅ 实操任务 2：解析 SSE 消息格式 ​

3. 示例代码 ​

🎯 核心知识点 ​

3.1 Python 服务端（Flask） ​

3.2 Python 服务端（FastAPI + 异步） ​

3.3 Go 服务端 ​

3.4 JavaScript 客户端（完整示例） ​

3.5 Python 客户端 ​

✅ 实操任务 3：搭建 SSE 服务 ​

4. 常见问题 ​

🎯 核心知识点 ​

4.1 连接数限制问题 ​

4.2 断线重连与消息丢失 ​

4.3 代理/负载均衡器缓冲问题 ​

4.4 跨域（CORS）问题 ​

4.5 内存泄漏问题 ​

✅ 实操任务 4：排查 SSE 问题 ​

5. 选型对比 ​

🎯 核心知识点 ​

5.1 SSE vs WebSocket vs 轮询 ​

5.2 决策流程图 ​

5.3 何时选择 SSE？ ​

5.4 SSE 的优势 ​

5.5 SSE 的局限 ​

✅ 实操任务 5：技术选型练习 ​

6. 学习资源与工具 ​

📚 权威学习资源 ​

🛠 常用调试工具 ​

1. 浏览器开发者工具 ​

2. curl 命令行 ​

3. httpie（更友好的 curl 替代品） ​

4. 在线 SSE 测试工具 ​

5. VS Code 插件 ​

📋 学习检查清单 ​

🎯 综合实操项目 ​

SSE（Server-Sent Events）协议学习指南

目录

1. 基础理论

🎯 核心知识点

1.1 什么是 SSE？

1.2 SSE 的工作原理

1.3 SSE 的适用场景

1.4 HTTP 响应头要求

1.5 SSE vs 传统轮询

✅ 实操任务 1：理解 SSE 请求

2. 核心语法

🎯 核心知识点

2.1 事件流格式（Event Stream Format）

2.2 消息格式详解

2.3 完整事件示例

2.4 特殊规则

2.5 客户端 EventSource API

✅ 实操任务 2：解析 SSE 消息格式

3. 示例代码

🎯 核心知识点

3.1 Python 服务端（Flask）

3.2 Python 服务端（FastAPI + 异步）

3.3 Go 服务端

3.4 JavaScript 客户端（完整示例）

3.5 Python 客户端

✅ 实操任务 3：搭建 SSE 服务

4. 常见问题

🎯 核心知识点

4.1 连接数限制问题

4.2 断线重连与消息丢失

4.3 代理/负载均衡器缓冲问题

4.4 跨域（CORS）问题

4.5 内存泄漏问题

✅ 实操任务 4：排查 SSE 问题

5. 选型对比

🎯 核心知识点

5.1 SSE vs WebSocket vs 轮询

5.2 决策流程图

5.3 何时选择 SSE？

5.4 SSE 的优势

5.5 SSE 的局限

✅ 实操任务 5：技术选型练习

6. 学习资源与工具

📚 权威学习资源

🛠 常用调试工具

1. 浏览器开发者工具

2. curl 命令行

3. httpie（更友好的 curl 替代品）

4. 在线 SSE 测试工具

5. VS Code 插件

📋 学习检查清单

🎯 综合实操项目