[!info] related notes

• 所属 MOC: AI Agent Application MOC
• 相关: SSE Client, SSE

Stream Resumption

一句话定义

Stream Resumption 是 SSE 连接断开后从断点继续接收事件的机制。SSE 协议原生支持通过 Last-Event-ID 实现断线续传。

核心原理

SSE 断线续传机制

1. 服务端发送事件时附带 id:
   event: text_delta
   id: 42
   data: {"delta": "你好"}

2. 连接断开

3. 客户端重连时携带 Last-Event-ID:
   GET /api/chat/stream
   Last-Event-ID: 42

4. 服务端从 id 43 继续发送

服务端实现

func sseHandler(w http.ResponseWriter, r *http.Request) {
    // 获取客户端最后收到的事件 ID
    lastID := r.Header.Get("Last-Event-ID")
    startSeq := 0
    if lastID != "" {
        startSeq, _ = strconv.Atoi(lastID)
    }

    // 从断点继续发送
    for event := range eventStream {
        if event.Sequence <= startSeq {
            continue // 跳过已发送的事件
        }
        fmt.Fprintf(w, "id: %d\nevent: %s\ndata: %s\n\n",
            event.Sequence, event.Type, event.JSON())
        w.(http.Flusher).Flush()
    }
}

客户端实现

// EventSource 原生支持断线续传
const es = new EventSource('/api/chat/stream');

// 浏览器自动处理 Last-Event-ID
es.onmessage = (event) => {
  // event.lastEventId 包含事件 ID
  console.log('Received event:', event.lastEventId);
};

常见坑

1. 不设置事件 id: 无法实现断线续传
2. id 不递增: 无法判断哪些事件已发送
3. 不缓存历史事件: 断线后无法重发旧事件
4. 缓存太大: 保留所有历史事件占用内存

参考资料

• SSE
• SSE Client