1
const express = require('express');
2
const app = express();
3

4
// SSE 路由
5
app.get('/api/sse', (req, res) => {
6
  // 设置 SSE 相关 headers
7
  res.setHeader('Content-Type', 'text/event-stream');
8
  res.setHeader('Cache-Control', 'no-cache');
9
  res.setHeader('Connection', 'keep-alive');
10
  res.setHeader('Access-Control-Allow-Origin', '*');
11

12
  // 发送初始连接消息
13
  res.write('event: connected\n');
14
  res.write('data: {"status": "connected"}\n\n');
15

16
  // 定时发送数据
17
  const intervalId = setInterval(() => {
18
    const data = {
19
      timestamp: Date.now(),
20
      random: Math.random().toFixed(4)
21
    };
22

23
    res.write(`event: heartbeat\n`);
24
    res.write(`data: ${JSON.stringify(data)}\n\n`);
25
  }, 5000);
26

27
  // 模拟事件推送
28
  const eventId = setInterval(() => {
29
    const messages = ['Hello!', 'New notification', 'System update'];
30
    const message = messages[Math.floor(Math.random() * messages.length)];
31

32
    res.write(`event: notification\n`);
33
    res.write(`id: ${Date.now()}\n`);
34
    res.write(`data: {"message": "${message}"}\n\n`);
35
  }, 15000);
36

37
  // 客户端断开连接时清理
38
  req.on('close', () => {
39
    clearInterval(intervalId);
40
    clearInterval(eventId);
41
    console.log('Client disconnected');
42
  });
43
});
44

45
app.listen(3000, () => {
46
  console.log('Server running on http://localhost:3000');
47
});

1
const http = require('http');
2

3
const server = http.createServer((req, res) => {
4
  if (req.url === '/api/events') {
5
    res.writeHead(200, {
6
      'Content-Type': 'text/event-stream',
7
      'Cache-Control': 'no-cache',
8
      'Connection': 'keep-alive',
9
      'Access-Control-Allow-Origin': '*',
10
      'X-Accel-Buffering': 'no' // 禁用 Nginx 缓冲
11
    });
12

13
    const sendEvent = (eventName, data) => {
14
      res.write(`event: ${eventName}\n`);
15
      res.write(`data: ${JSON.stringify(data)}\n\n`);
16
    };
17

18
    const heartbeat = setInterval(() => {
19
      res.write(`: heartbeat\n\n`);
20
    }, 30000);
21

22
    const pushData = setInterval(() => {
23
      sendEvent('update', {
24
        time: new Date().toISOString(),
25
        value: Math.random()
26
      });
27
    }, 2000);
28

29
    req.on('close', () => {
30
      clearInterval(heartbeat);
31
      clearInterval(pushData);
32
      console.log('Connection closed');
33
    });
34
  } else {
35
    res.writeHead(404);
36
    res.end('Not Found');
37
  }
38
});
39

40
server.listen(8080, () => {
41
  console.log('SSE Server running on http://localhost:8080');
42
});

1
const Koa = require('koa');
2
const Router = require('@koa/router');
3

4
const app = new Koa();
5
const router = new Router();
6

7
router.get('/api/sse', async (ctx) => {
8
  ctx.set('Content-Type', 'text/event-stream');
9
  ctx.set('Cache-Control', 'no-cache');
10
  ctx.set('Connection', 'keep-alive');
11
  ctx.set('Access-Control-Allow-Origin', '*');
12
  ctx.set('X-Accel-Buffering', 'no');
13

14
  const encoder = new TextEncoder();
15

16
  const send = (eventName, data, eventId = null) => {
17
    let message = '';
18
    if (eventId) message += `id: ${eventId}\n`;
19
    message += `event: ${eventName}\n`;
20
    message += `data: ${JSON.stringify(data)}\n\n`;
21
    ctx.res.write(message);
22
  };
23

24
  send('connected', { status: 'connected' });
25

26
  const interval = setInterval(() => {
27
    send('heartbeat', { timestamp: Date.now() });
28
  }, 5000);
29

30
  ctx.req.on('close', () => {
31
    clearInterval(interval);
32
  });
33
});
34

35
app.use(router.routes());
36
app.listen(3000);

1
from flask import Flask, Response, request
2
import json
3
import time
4
import random
5
from datetime import datetime
6

7
app = Flask(__name__)
8

9
def generate_events():
10
    """生成 SSE 事件流"""
11
    event_id = 0
12

13
    while True:
14
        event_id += 1
15
        data = {
16
            'id': event_id,
17
            'timestamp': datetime.now().isoformat(),
18
            'value': round(random.random() * 100, 2)
19
        }
20

21
        yield f"event: update\n"
22
        yield f"id: {event_id}\n"
23
        yield f"data: {json.dumps(data)}\n\n"
24

25
        time.sleep(2)
26

27
@app.route('/api/events')
28
def events():
29
    """SSE 端点"""
30
    last_event_id = request.headers.get('Last-Event-ID')
31

32
    if last_event_id:
33
        print(f"Reconnecting from event ID: {last_event_id}")
34

35
    return Response(
36
        generate_events(),
37
        mimetype='text/event-stream',
38
        headers={
39
            'Cache-Control': 'no-cache',
40
            'Connection': 'keep-alive',
41
            'Access-Control-Allow-Origin': '*',
42
            'X-Accel-Buffering': 'no'
43
        }
44
    )
45

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

1
from fastapi import FastAPI, Request
2
from fastapi.responses import StreamingResponse
3
import asyncio
4
import json
5
import random
6
from datetime import datetime
7
from typing import AsyncGenerator
8

9
app = FastAPI()
10

11
@app.get("/api/events")
12
async def sse_endpoint(request: Request):
13
    """SSE 端点"""
14
    last_event_id = request.headers.get("last-event-id")
15

16
    async def event_generator() -> AsyncGenerator[str, None]:
17
        event_id = int(last_event_id) if last_event_id else 0
18

19
        while True:
20
            event_id += 1
21
            data = {
22
                "id": event_id,
23
                "timestamp": datetime.now().isoformat(),
24
                "value": round(random.random() * 100, 2)
25
            }
26

27
            yield f"event: update\n"
28
            yield f"id: {event_id}\n"
29
            yield f"data: {json.dumps(data)}\n\n"
30

31
            await asyncio.sleep(2)
32

33
    return StreamingResponse(
34
        event_generator(),
35
        media_type="text/event-stream",
36
        headers={
37
            "Cache-Control": "no-cache",
38
            "Connection": "keep-alive",
39
            "Access-Control-Allow-Origin": "*",
40
            "X-Accel-Buffering": "no"
41
        }
42
    )

1
package main
2

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

10
func sseHandler(w http.ResponseWriter, r *http.Request) {
11
  w.Header().Set("Content-Type", "text/event-stream")
12
  w.Header().Set("Cache-Control", "no-cache")
13
  w.Header().Set("Connection", "keep-alive")
14
  w.Header().Set("Access-Control-Allow-Origin", "*")
15

16
  flusher, ok := w.(http.Flusher)
17
  if !ok {
18
    http.Error(w, "SSE not supported", http.StatusInternalServerError)
19
    return
20
  }
21

22
  eventID := 0
23
  ticker := time.NewTicker(2 * time.Second)
24
  defer ticker.Stop()
25

26
  for {
27
    select {
28
    case <-r.Context().Done():
29
      log.Println("Client disconnected")
30
      return
31
    case <-ticker.C:
32
      eventID++
33
      data := map[string]interface{}{
34
        "id":        eventID,
35
        "timestamp": time.Now().Format(time.RFC3339),
36
        "value":     float64(eventID) * 1.5,
37
      }
38

39
      fmt.Fprintf(w, "id: %d\n", eventID)
40
      fmt.Fprintf(w, "event: update\n")
41
      fmt.Fprintf(w, "data: %s\n\n", toJSON(data))
42
      flusher.Flush()
43
    }
44
  }
45
}
46

47
func toJSON(data map[string]interface{}) string {
48
  return fmt.Sprintf(`{"id":%d,"timestamp":"%s","value":%.2f}`,
49
    data["id"], data["timestamp"], data["value"])
50
}
51

52
func main() {
53
  http.HandleFunc("/api/events", sseHandler)
54
  log.Println("Server running on http://localhost:8080")
55
  log.Fatal(http.ListenAndServe(":8080", nil))
56
}

1
@RestController
2
@RequestMapping("/api")
3
public class SseController {
4

5
    @GetMapping("/events")
6
    public SseEmitter streamEvents() {
7
        SseEmitter emitter = new SseEmitter(0L);
8

9
        ExecutorService executor = Executors.newCachedThreadPool();
10

11
        executor.execute(() -> {
12
            try {
13
                int[] eventId = {0};
14

15
                while (true) {
16
                    eventId[0]++;
17
                    Map<String, Object> data = new HashMap<>();
18
                    data.put("id", eventId[0]);
19
                    data.put("timestamp", Instant.now().toString());
20
                    data.put("value", Math.random() * 100);
21

22
                    emitter.send(SseEmitter.event()
23
                        .name("update")
24
                        .id(String.valueOf(eventId[0]))
25
                        .data(data));
26

27
                    Thread.sleep(2000);
28
                }
29
            } catch (Exception e) {
30
                emitter.completeWithError(e);
31
            }
32
        });
33

34
        emitter.onCompletion(() -> log.info("SSE stream completed"));
35
        emitter.onError(e -> log.error("SSE error", e));
36
        emitter.onTimeout(() -> log.warning("SSE timeout"));
37

38
        return emitter;
39
    }
40
}

1
server {
2
    location /api/sse {
3
        # 禁用缓冲
4
        proxy_buffering off;
5
        proxy_cache off;
6

7
        # 禁用 gzip (SSE 需要原始流)
8
        gzip off;
9

10
        proxy_pass http://backend;
11
        proxy_http_version 1.1;
12
        proxy_set_header Connection '';
13
        proxy_read_timeout 86400;
14
        proxy_send_timeout 86400;
15
    }
16
}