Server-Sent Events - ZeMiのBlog

为什么选择 SSE

面向“服务器→浏览器单向推送”的场景，浏览器原生支持 EventSource，无需第三方库
基于标准 HTTP（含 HTTP/2），与现有网关、负载均衡、鉴权、缓存体系兼容度高
自动断线重连与消息续传（Last‑Event‑ID）由浏览器内置完成
只需文本数据，天然适合通知、进度、日志、监控等单向推送

原理与协议

连接与建立
客户端通过 EventSource 发起 GET 请求，请求头隐式包含 Accept: text/event-stream；服务器以 Content-Type: text/event-stream 响应并保持连接打开，通过持续的 HTTP 响应流推送数据。
事件格式
一条事件由多行键值对组成并以连续两个换行符结束。常见字段：
- event: 事件类型，默认是 message
- data: 消息主体数据（可多行，每行一个 data:）
- id: 事件唯一标识（用于断线续传）
- retry: 重连时间间隔（毫秒）示例： data: This is a message

或带事件类型与 JSON 数据：

event: userupdate
data: {"username":"john_doe","status":"online"}

断线重连
浏览器自动重试，若服务器曾发送 id:，浏览器在重连时携带 Last-Event-ID，服务器可从该 ID 之后继续发送。
保活与代理
许多代理会因长时间无数据而断开，服务器可定期发送注释行（以冒号起始，如 : keepalive）作为保活。建议设置 Cache-Control: no-cache、Connection: keep-alive，必要时对部分代理设置 X-Accel-Buffering: no。

SSE 与 WebSocket 对比

通信模型
- SSE：服务器→浏览器单向文本流，浏览器原生 EventSource，自动重连与续传
- WebSocket：在独立的 TCP 连接上建立全双工通信，使用协议升级与帧（frame）进行数据传输，需自行处理重连与续传
基础设施与兼容性
- SSE：走标准 HTTP，易穿透与复用现有网关、鉴权、中间件；适合通知、状态推送、日志流
- WebSocket：适合 IM、协作编辑、游戏等强双向低延时场景
性能与实现
- SSE：实现简单、成本低；消息量大或需要二进制时不适用
- WebSocket：灵活强大，但实现与维护复杂度更高

后端

CodeBlock Loading...

前端

<script>
// 订阅统一流 events，根据不同 event 类型分别处理
const es = new EventSource("http://localhost:8080/events?stream=events");

// 默认事件类型 message：文本通知
es.onmessage = (e) => {
  console.log("message", e.data);
};
// tick：时间戳字符串
es.addEventListener("tick", (e) => {
  console.log("tick", e.data);
});
// stock：行情 JSON
es.addEventListener("stock", (e) => {
  try { console.log("stock", JSON.parse(e.data)); } catch { console.log(e.data); }
});
// score：比分 JSON
es.addEventListener("score", (e) => {
  try { console.log("score", JSON.parse(e.data)); } catch { console.log(e.data); }
});
// location：位置 JSON
es.addEventListener("location", (e) => {
  try { console.log("location", JSON.parse(e.data)); } catch { console.log(e.data); }
});
// progress：进度 JSON/数字
es.addEventListener("progress", (e) => {
  try { console.log("progress", JSON.parse(e.data)); } catch { console.log(e.data); }
});
// system：系统事件文本
es.addEventListener("system", (e) => {
  console.log("system", e.data);
});
</script>

// 组件中使用：将不同事件类型渲染到页面
const es = new EventSource("http://localhost:8080/events?stream=events");
es.onmessage = (e) => {
  const el = document.getElementById("out");
  if (el) el.textContent = e.data;
};
es.addEventListener("tick", (e) => {
  const el = document.getElementById("tick");
  if (el) el.textContent = e.data;
});
es.addEventListener("progress", (e) => {
  const el = document.getElementById("progress");
  try { const p = JSON.parse(e.data); if (el) el.textContent = p.percent + "%"; }
  catch { if (el) el.textContent = e.data; }
});

运行与测试

启动后端：go run main.go，监听 8080 端口
打开页面：在浏览器控制台观察事件输出
发布消息：POST 文本到 http://localhost:8080/publish?type=message 或 GET http://localhost:8080/publish?type=message&msg=hello，浏览器会收到 events 流的推送（根据 type 作为事件类型）

最佳实践与注意事项

保活与缓冲：周期性发送注释行维持连接活跃；必要时设置 X-Accel-Buffering: no
事件整形：多行数据逐行 data: 前缀；使用 id 支持断线续传（配合 Last-Event-ID）
并发与背压：为 client 通道设置有限缓冲并在广播时使用非阻塞写入；根据场景调整缓冲与丢弃策略
安全与鉴权：走标准 HTTP，可复用 Cookie、Token、签名等鉴权方式；若需鉴权刷新，结合短期令牌与重连策略

适用场景

实时通知与社交动态提示
股票行情与金融市场数据更新
体育赛事比分与统计实时更新
地理位置追踪与地图轨迹刷新
系统监控面板：服务器状态、日志流与指标
长任务进度、批处理状态、构建或部署流水输出
面板与监控看板的低频实时更新

为什么选择 SSE

面向“服务器→浏览器单向推送”的场景，浏览器原生支持 EventSource，无需第三方库
基于标准 HTTP（含 HTTP/2），与现有网关、负载均衡、鉴权、缓存体系兼容度高
自动断线重连与消息续传（Last‑Event‑ID）由浏览器内置完成
只需文本数据，天然适合通知、进度、日志、监控等单向推送

原理与协议

连接与建立
客户端通过 EventSource 发起 GET 请求，请求头隐式包含 Accept: text/event-stream；服务器以 Content-Type: text/event-stream 响应并保持连接打开，通过持续的 HTTP 响应流推送数据。
事件格式
一条事件由多行键值对组成并以连续两个换行符结束。常见字段：
- event: 事件类型，默认是 message
- data: 消息主体数据（可多行，每行一个 data:）
- id: 事件唯一标识（用于断线续传）
- retry: 重连时间间隔（毫秒）示例： data: This is a message

或带事件类型与 JSON 数据：

event: userupdate
data: {"username":"john_doe","status":"online"}

断线重连
浏览器自动重试，若服务器曾发送 id:，浏览器在重连时携带 Last-Event-ID，服务器可从该 ID 之后继续发送。
保活与代理
许多代理会因长时间无数据而断开，服务器可定期发送注释行（以冒号起始，如 : keepalive）作为保活。建议设置 Cache-Control: no-cache、Connection: keep-alive，必要时对部分代理设置 X-Accel-Buffering: no。

SSE 与 WebSocket 对比

通信模型
- SSE：服务器→浏览器单向文本流，浏览器原生 EventSource，自动重连与续传
- WebSocket：在独立的 TCP 连接上建立全双工通信，使用协议升级与帧（frame）进行数据传输，需自行处理重连与续传
基础设施与兼容性
- SSE：走标准 HTTP，易穿透与复用现有网关、鉴权、中间件；适合通知、状态推送、日志流
- WebSocket：适合 IM、协作编辑、游戏等强双向低延时场景
性能与实现
- SSE：实现简单、成本低；消息量大或需要二进制时不适用
- WebSocket：灵活强大，但实现与维护复杂度更高

后端

package main

import (
    "encoding/json"
    "io"
    "log"
    "net/http"
    "strconv"
    "strings"
    "time"

    "github.com/r3labs/sse/v2"
)

func main() {
    // 创建 SSE 服务器与单一流（建议统一使用一个流，通过 event 类型区分）
    server := sse.New()
    server.CreateStream("events")

    mux := http.NewServeMux()
    mux.HandleFunc("/events", func(w http.ResponseWriter, r *http.Request) {
        w.Header().Set("Access-Control-Allow-Origin", "*")
        server.ServeHTTP(w, r)
    })

    mux.HandleFunc("/publish", func(w http.ResponseWriter, r *http.Request) {
        // 支持通过 type 指定事件类型（默认 message），便于后端统一发布
        // 事件类型建议：
        // - message：文本通知，适合短消息或提示
        // - tick：时间戳字符串，适合心跳与时钟
        // - userupdate：JSON 用户状态，如 {"username":"john","status":"online"}
        // - stock：JSON 行情，如 {"symbol":"ACME","price":123.45}
        // - score：JSON 比分，如 {"home":1,"away":2}
        // - location：JSON 位置，如 {"lat":31.23,"lng":121.47}
        // - progress：JSON/数字进度，如 {"percent":42}
        // - system：文本系统事件，如 "ok" 或日志片段
        body, _ := io.ReadAll(r.Body)
        msg := strings.TrimSpace(string(body))
        if msg == "" {
            msg = r.URL.Query().Get("msg")
        }
        if msg == "" {
            http.Error(w, "msg required", http.StatusBadRequest)
            return
        }
        eventType := r.URL.Query().Get("type")
        if eventType == "" {
            eventType = "message"
        }
        server.Publish("events", &sse.Event{
            ID:    []byte(strconv.FormatInt(time.Now().UnixNano(), 10)),
            Event: []byte(eventType),
            Data:  []byte(msg),
        })
        w.WriteHeader(http.StatusNoContent)
    })

    go func() {
        // tick：时间戳字符串，适合心跳与时钟
        ticker := time.NewTicker(5 * time.Second)
        for t := range ticker.C {
            server.Publish("events", &sse.Event{
                ID:    []byte(strconv.FormatInt(time.Now().UnixNano(), 10)),
                Event: []byte("tick"),
                Data:  []byte(t.Format(time.RFC3339)),
            })
        }
    }()

    go func() {
        // stock：行情 JSON，适合实时价格或指标
        type Stock struct {
            Symbol string  `json:"symbol"`
            Price  float64 `json:"price"`
        }
        t := time.NewTicker(7 * time.Second)
        for tt := range t.C {
            _ = tt
            payload, _ := json.Marshal(Stock{Symbol: "ACME", Price: 123.45})
            server.Publish("events", &sse.Event{
                ID:    []byte(strconv.FormatInt(time.Now().UnixNano(), 10)),
                Event: []byte("stock"),
                Data:  payload,
            })
        }
    }()

    go func() {
        // score：比分 JSON，适合体育赛事
        type Score struct {
            Home int `json:"home"`
            Away int `json:"away"`
        }
        t := time.NewTicker(9 * time.Second)
        h, a := 0, 0
        for range t.C {
            h++
            payload, _ := json.Marshal(Score{Home: h, Away: a})
            server.Publish("events", &sse.Event{
                ID:    []byte(strconv.FormatInt(time.Now().UnixNano(), 10)),
                Event: []byte("score"),
                Data:  payload,
            })
        }
    }()

    go func() {
        // location：位置 JSON，适合地图轨迹
        type Location struct {
            Lat float64 `json:"lat"`
            Lng float64 `json:"lng"`
        }
        t := time.NewTicker(11 * time.Second)
        lat, lng := 31.2304, 121.4737
        for range t.C {
            lat += 0.0003
            lng += 0.0002
            payload, _ := json.Marshal(Location{Lat: lat, Lng: lng})
            server.Publish("events", &sse.Event{
                ID:    []byte(strconv.FormatInt(time.Now().UnixNano(), 10)),
                Event: []byte("location"),
                Data:  payload,
            })
        }
    }()

    go func() {
        // progress：进度（数字或 JSON），适合长任务状态
        type Progress struct {
            Percent int `json:"percent"`
        }
        for {
            for p := 0; p <= 100; p += 20 {
                payload, _ := json.Marshal(Progress{Percent: p})
                server.Publish("events", &sse.Event{
                    ID:    []byte(strconv.FormatInt(time.Now().UnixNano(), 10)),
                    Event: []byte("progress"),
                    Data:  payload,
                })
                time.Sleep(3 * time.Second)
            }
        }
    }()

    go func() {
        // system：系统事件文本，适合日志片段或健康检查
        t := time.NewTicker(30 * time.Second)
        for range t.C {
            server.Publish("events", &sse.Event{
                ID:    []byte(strconv.FormatInt(time.Now().UnixNano(), 10)),
                Event: []byte("system"),
                Data:  []byte("ok"),
            })
        }
    }()

    log.Fatal(http.ListenAndServe(":8080", mux))
}

CodeBlock Loading...

前端

<script>
// 订阅统一流 events，根据不同 event 类型分别处理
const es = new EventSource("http://localhost:8080/events?stream=events");

// 默认事件类型 message：文本通知
es.onmessage = (e) => {
  console.log("message", e.data);
};
// tick：时间戳字符串
es.addEventListener("tick", (e) => {
  console.log("tick", e.data);
});
// stock：行情 JSON
es.addEventListener("stock", (e) => {
  try { console.log("stock", JSON.parse(e.data)); } catch { console.log(e.data); }
});
// score：比分 JSON
es.addEventListener("score", (e) => {
  try { console.log("score", JSON.parse(e.data)); } catch { console.log(e.data); }
});
// location：位置 JSON
es.addEventListener("location", (e) => {
  try { console.log("location", JSON.parse(e.data)); } catch { console.log(e.data); }
});
// progress：进度 JSON/数字
es.addEventListener("progress", (e) => {
  try { console.log("progress", JSON.parse(e.data)); } catch { console.log(e.data); }
});
// system：系统事件文本
es.addEventListener("system", (e) => {
  console.log("system", e.data);
});
</script>

// 组件中使用：将不同事件类型渲染到页面
const es = new EventSource("http://localhost:8080/events?stream=events");
es.onmessage = (e) => {
  const el = document.getElementById("out");
  if (el) el.textContent = e.data;
};
es.addEventListener("tick", (e) => {
  const el = document.getElementById("tick");
  if (el) el.textContent = e.data;
});
es.addEventListener("progress", (e) => {
  const el = document.getElementById("progress");
  try { const p = JSON.parse(e.data); if (el) el.textContent = p.percent + "%"; }
  catch { if (el) el.textContent = e.data; }
});

运行与测试

启动后端：go run main.go，监听 8080 端口
打开页面：在浏览器控制台观察事件输出
发布消息：POST 文本到 http://localhost:8080/publish?type=message 或 GET http://localhost:8080/publish?type=message&msg=hello，浏览器会收到 events 流的推送（根据 type 作为事件类型）

最佳实践与注意事项

保活与缓冲：周期性发送注释行维持连接活跃；必要时设置 X-Accel-Buffering: no
事件整形：多行数据逐行 data: 前缀；使用 id 支持断线续传（配合 Last-Event-ID）
并发与背压：为 client 通道设置有限缓冲并在广播时使用非阻塞写入；根据场景调整缓冲与丢弃策略
安全与鉴权：走标准 HTTP，可复用 Cookie、Token、签名等鉴权方式；若需鉴权刷新，结合短期令牌与重连策略

适用场景

实时通知与社交动态提示
股票行情与金融市场数据更新
体育赛事比分与统计实时更新
地理位置追踪与地图轨迹刷新
系统监控面板：服务器状态、日志流与指标
长任务进度、批处理状态、构建或部署流水输出
面板与监控看板的低频实时更新