GET /events HTTP/1.1
Host: example.com
Accept: text/event-stream

HTTP/1.1 200 OK
Content-Type: text/event-stream
Cache-Control: no-cache
Connection: keep-alive

data: {"message": "第一条消息"}

data: {"message": "第二条消息"}

curl -N http://localhost:8080/events

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

event: notification
data: {"title": "告警", "body": "CPU 超过 90%"}

event: log
data: {"level": "INFO", "message": "任务完成"}

id: 42
data: {"message": "第42条消息"}

retry: 5000

package main

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

// EventBroker 负责管理所有 SSE 客户端连接
type EventBroker struct {
    clients    map[chan []byte]struct{}
    register   chan chan []byte
    unregister chan chan []byte
    broadcast  chan []byte
    mu         sync.RWMutex
}

func NewEventBroker() *EventBroker {
    return &EventBroker{
        clients:    make(map[chan []byte]struct{}),
        register:   make(chan chan []byte),
        unregister: make(chan chan []byte),
        broadcast:  make(chan []byte),
    }
}

// Run 在单个 goroutine 中串行处理所有客户端操作，避免锁竞争
func (b *EventBroker) Run() {
    for {
        select {
        case client := <-b.register:
            b.clients[client] = struct{}{}

        case client := <-b.unregister:
            if _, ok := b.clients[client]; ok {
                delete(b.clients, client)
                close(client)
            }

        case msg := <-b.broadcast:
            for client := range b.clients {
                // 非阻塞发送：慢客户端不会拖死其他客户端
                select {
                case client <- msg:
                default:
                    // 客户端消费太慢，跳过这条消息
                }
            }
        }
    }
}

// sseHandler 处理单个 SSE 客户端连接
func (b *EventBroker) 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")
    w.Header().Set("X-Accel-Buffering", "no") // 防止 Nginx 缓冲

    flusher, ok := w.(http.Flusher)
    if !ok {
        http.Error(w, "不支持 Streaming", http.StatusInternalServerError)
        return
    }

    // 读取客户端最后收到的事件 ID（断线重连场景）
    lastEventID := r.Header.Get("Last-Event-ID")

    client := make(chan []byte, 256) // 带缓冲，减少阻塞
    b.register <- client
    defer func() {
        b.unregister <- client
    }()

    // 心跳 ticker：每 30 秒发一次注释行，防止代理断开空闲连接
    heartbeat := time.NewTicker(30 * time.Second)
    defer heartbeat.Stop()

    // 通知客户端连接已建立
    fmt.Fprintf(w, "retry: 3000\n\n")
    flusher.Flush()

    // 如果有断线重连的需求，补推丢失的事件
    if lastEventID != "" {
        fmt.Fprintf(w, "id: %s\ndata: 重连成功，从 ID %s 继续\n\n", lastEventID, lastEventID)
        flusher.Flush()
    }

    ctx := r.Context()

    for {
        select {
        case <-ctx.Done():
            // 客户端断开连接
            return

        case <-heartbeat.C:
            // 心跳：SSE 注释行，浏览器忽略但能保活
            fmt.Fprintf(w, ": heartbeat\n\n")
            flusher.Flush()

        case msg, ok := <-client:
            if !ok {
                return
            }
            fmt.Fprintf(w, "data: %s\n\n", msg)
            flusher.Flush()
        }
    }
}

func main() {
    broker := NewEventBroker()
    go broker.Run()

    http.HandleFunc("/events", broker.sseHandler)

    // 模拟推送：每 2 秒广播一条消息
    go func() {
        ticker := time.NewTicker(2 * time.Second)
        for range ticker.C {
            msg := fmt.Sprintf("服务器时间: %s", time.Now().Format("15:04:05"))
            broker.broadcast <- []byte(msg)
        }
    }()

    log.Println("SSE 服务启动在 :8080")
    log.Fatal(http.ListenAndServe(":8080", nil))
}

const es = new EventSource('/events')

es.onmessage = (event) => {
  const data = JSON.parse(event.data)
  console.log('收到消息:', data)
}

es.addEventListener('notification', (event) => {
  // 处理特定类型的事件
  console.log('通知:', JSON.parse(event.data))
})

es.onerror = (event) => {
  // EventSource 会自动重连，这里只做日志
  console.warn('SSE 连接异常，正在自动重连...')
}

// 需要断开时
// es.close()

// 支持自定义 Header 的 SSE 客户端（手动解析）
const response = await fetch('/events', {
  headers: { 'Authorization': 'Bearer ' + token }
})
const reader = response.body.getReader()
const decoder = new TextDecoder()
let buffer = ''

while (true) {
  const { done, value } = await reader.read()
  if (done) break

  buffer += decoder.decode(value, { stream: true })
  const parts = buffer.split('\n\n')
  buffer = parts.pop() // 最后一个可能不完整，留着等下一次

  for (const part of parts) {
    const match = part.match(/^data: (.+)$/m)
    if (match) {
      console.log('收到:', JSON.parse(match[1]))
    }
  }
}