服务端事件推送处理器

功能说明

服务端事件推送适合部署进度、通知和实时指标这类单向更新，比 WebSocket 简单，浏览器原生支持自动重连。处理器把响应的内容类型设为 text/event-stream，写完响应头立即刷新，之后每产生一条事件就写入 data 行并再次刷新，客户端才能实时收到。

Broker 维护所有订阅者的通道，订阅和取消订阅都加锁；广播时向每个通道做非阻塞发送，缓冲已满的客户端直接丢弃这条事件，避免一个慢客户端拖住整个广播。

关键代码

ServeHTTP 先断言响应实现 http.Flusher，不支持刷新的包装器会直接返回错误，否则消息会滞留在缓冲区。订阅后立刻注册 defer 取消订阅，循环里同时监听请求上下文结束和事件通道；请求断开时上下文取消，处理器退出并释放资源。main 用 httptest 启动内嵌服务，先启动广播协程再发起请求，逐行扫描响应，读取两条 data 事件后关闭连接。

使用注意

响应头一旦写出就不能再修改，认证或限流必须在此之前完成。反向代理可能缓冲响应，需要关闭缓冲或增加心跳注释行。示例未实现 Last-Event-ID 和历史补发，重连后只能收到新事件。每个连接占用一个协程和一个通道，连接数需要限额。丢弃慢客户端是策略而非缺陷，若要求不丢事件应改为持久化队列。

验证方式

输出第一行为内容类型 text/event-stream，随后依次打印 deploy started 与 deploy finished。客户端读取两条事件后主动关闭，服务端上下文取消，处理器退出。可以同时启动两个客户端，确认广播会分别送达，而某个客户端停止读取后另一个仍能继续收到事件。

代码示例

package main

import (
	"bufio"
	"fmt"
	"net/http"
	"net/http/httptest"
	"strings"
	"sync"
	"time"
)

type Broker struct {
	mu      sync.Mutex
	clients map[chan string]struct{}
}

func NewBroker() *Broker {
	return &Broker{clients: make(map[chan string]struct{})}
}

func (b *Broker) Subscribe() chan string {
	events := make(chan string, 8)
	b.mu.Lock()
	b.clients[events] = struct{}{}
	b.mu.Unlock()
	return events
}

func (b *Broker) Unsubscribe(events chan string) {
	b.mu.Lock()
	delete(b.clients, events)
	b.mu.Unlock()
}

func (b *Broker) Broadcast(event string) {
	b.mu.Lock()
	defer b.mu.Unlock()
	for events := range b.clients {
		select {
		case events <- event:
		default:
		}
	}
}

func (b *Broker) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	flusher, ok := w.(http.Flusher)
	if !ok {
		http.Error(w, "streaming unsupported", http.StatusInternalServerError)
		return
	}
	w.Header().Set("Content-Type", "text/event-stream")
	w.Header().Set("Cache-Control", "no-cache")
	w.Header().Set("Connection", "keep-alive")
	w.WriteHeader(http.StatusOK)
	flusher.Flush()

	events := b.Subscribe()
	defer b.Unsubscribe(events)
	for {
		select {
		case <-r.Context().Done():
			return
		case event := <-events:
			fmt.Fprintf(w, "data: %s\n\n", event)
			flusher.Flush()
		}
	}
}

func main() {
	broker := NewBroker()
	server := httptest.NewServer(broker)
	defer server.Close()

	go func() {
		time.Sleep(20 * time.Millisecond)
		broker.Broadcast("deploy started")
		time.Sleep(10 * time.Millisecond)
		broker.Broadcast("deploy finished")
	}()

	response, err := http.Get(server.URL)
	if err != nil {
		panic(err)
	}
	defer response.Body.Close()
	fmt.Println("content-type:", response.Header.Get("Content-Type"))

	scanner := bufio.NewScanner(response.Body)
	received := 0
	for scanner.Scan() && received < 2 {
		line := scanner.Text()
		if strings.HasPrefix(line, "data: ") {
			fmt.Println("event:", strings.TrimPrefix(line, "data: "))
			received++
		}
	}
}