1
package main
2

3
import (
4
  "fmt"
5
  "log"
6
  "net/http"
7
  "time"
8
)
9

10
// sseHandler 处理 SSE 连接
11
func sseHandler(w http.ResponseWriter, r *http.Request) {
12
  // 设置 SSE 必需的响应头
13
  w.Header().Set("Content-Type", "text/event-stream")
14
  w.Header().Set("Cache-Control", "no-cache")
15
  w.Header().Set("Connection", "keep-alive")
16

17
  // 获取 Flusher，用于立即将数据推送到客户端
18
  flusher, ok := w.(http.Flusher)
19
  if !ok {
20
    http.Error(w, "不支持 Streaming", http.StatusInternalServerError)
21
    return
22
  }
23

24
  // 模拟 LLM 逐 token 生成
25
  tokens := []string{"你好", "，", "我是", "一个", "流式", "输出", "的", "示例", "。"}
26
  for i, token := range tokens {
27
    // 检查客户端是否已断开
28
    select {
29
    case <-r.Context().Done():
30
      log.Println("客户端断开连接，停止发送")
31
      return
32
    default:
33
    }
34

35
    // 写入 SSE 格式的事件
36
    fmt.Fprintf(w, "id: %d\n", i+1)
37
    fmt.Fprintf(w, "data: %s\n\n", token)
38
    flusher.Flush()
39

40
    // 模拟生成延迟
41
    time.Sleep(200 * time.Millisecond)
42
  }
43

44
  // 发送结束标记，客户端据此判断流结束
45
  fmt.Fprintf(w, "event: done\ndata: [DONE]\n\n")
46
  flusher.Flush()
47
}
48

49
func main() {
50
  http.HandleFunc("/stream", sseHandler)
51
  log.Println("SSE 服务启动在 :8080")
52
  log.Fatal(http.ListenAndServe(":8080", nil))
53
}

1
// Node.js SSE 服务端
2
import { createServer } from "http";
3

4
createServer((req, res) => {
5
  if (req.url === "/stream") {
6
    // 设置 SSE 响应头
7
    res.writeHead(200, {
8
      "Content-Type": "text/event-stream",
9
      "Cache-Control": "no-cache",
10
      Connection: "keep-alive",
11
    });
12

13
    const tokens = ["你好", "，", "我是", "流式", "输出", "示例", "。"];
14
    let index = 0;
15

16
    const interval = setInterval(() => {
17
      // 检查是否还有 token 要发送
18
      if (index < tokens.length) {
19
        res.write(`id: ${index + 1}\ndata: ${tokens[index]}\n\n`);
20
        index++;
21
      } else {
22
        // 发送结束事件
23
        res.write("event: done\ndata: [DONE]\n\n");
24
        clearInterval(interval);
25
        res.end();
26
      }
27
    }, 200);
28

29
    // 客户端断开时清理
30
    req.on("close", () => {
31
      clearInterval(interval);
32
    });
33
  }
34
}).listen(8080, () => console.log("SSE 服务启动在 :8080"));

1
// 使用 EventSource 消费 SSE 流
2
function consumeSSE(url: string): void {
3
  const source = new EventSource(url);
4

5
  source.onmessage = (event: MessageEvent) => {
6
    // 每收到一个事件，追加到页面
7
    console.log(`收到事件 #${event.lastEventId}: ${event.data}`);
8
  };
9

10
  // 监听自定义事件类型
11
  source.addEventListener("done", (event: MessageEvent) => {
12
    console.log("流结束:", event.data);
13
    source.close();
14
  });
15

16
  source.onerror = () => {
17
    // EventSource 自动重连，这里只需要记录日志
18
    console.log("连接中断，自动重连中...");
19
  };
20
}
21

22
consumeSSE("http://localhost:8080/stream");

1
// 使用 fetch + ReadableStream 消费 SSE，支持背压
2
async function consumeWithBackpressure(url: string): Promise<void> {
3
  const response = await fetch(url);
4
  const reader = response.body!.getReader();
5
  const decoder = new TextDecoder();
6
  let buffer = "";
7

8
  while (true) {
9
    // read() 返回后才继续拉取，天然背压
10
    const { done, value } = await reader.read();
11
    if (done) break;
12

13
    buffer += decoder.decode(value, { stream: true });
14

15
    // 按 SSE 事件边界（双换行）拆分
16
    const parts = buffer.split("\n\n");
17
    buffer = parts.pop()!; // 最后一段可能不完整，保留到下次
18

19
    for (const part of parts) {
20
      const lines = part.split("\n");
21
      let data = "";
22
      for (const line of lines) {
23
        if (line.startsWith("data: ")) {
24
          data += line.slice(6);
25
        }
26
      }
27
      if (data) {
28
        console.log("收到:", data);
29
        // 这里可以做耗时处理，处理完才继续 read
30
      }
31
    }
32
  }
33
}
34

35
consumeWithBackpressure("http://localhost:8080/stream");