228 lines
5.8 KiB
Markdown
228 lines
5.8 KiB
Markdown
---
|
||
tags: [sse, event-stream, pub-sub, real-time, websockets-alternative, go]
|
||
create time: 2026-06-03 10:35
|
||
---
|
||
|
||
# 08. SSE 实时推送
|
||
|
||
## 概述
|
||
|
||
内存 EventBus 发布/订阅，SSE 推送管线进度到浏览器，让用户实时看到生成过程。
|
||
|
||
---
|
||
|
||
## 正文
|
||
|
||
```mermaid
|
||
flowchart LR
|
||
P["Pipeline Callback"] -->|"Publish"| EB["EventBus Broker"]
|
||
EB -->|"Subscribe taskID"| H1["SSE Handler /tasks/:id/stream"]
|
||
EB -->|"SubscribeAll global"| H2["SSE Handler /projects/:id/stream"]
|
||
H1 -->|"text/event-stream"| B1["Browser EventSource"]
|
||
H2 -->|"text/event-stream"| B2["Browser EventSource"]
|
||
|
||
style P fill:#e8f5e9,stroke:#388e3c
|
||
style EB fill:#fff3e0,stroke:#f57c00
|
||
style H1 fill:#e3f2fd,stroke:#1976d2
|
||
style H2 fill:#e3f2fd,stroke:#1976d2
|
||
style B1 fill:#f3e5f5,stroke:#7b1fa2
|
||
style B2 fill:#f3e5f5,stroke:#7b1fa2
|
||
```
|
||
|
||
---
|
||
|
||
## EventBus 架构
|
||
|
||
EventBus 是 Gen2D 的内存事件总线，负责在 Pipeline 执行过程中发布进度事件，并由 SSE Handler 订阅推送给客户端。
|
||
|
||
### 核心设计
|
||
|
||
```go
|
||
type Broker struct {
|
||
mu sync.RWMutex
|
||
subs map[string][]chan TaskEvent // per-task 订阅
|
||
allSubs []chan TaskEvent // 全局订阅
|
||
}
|
||
```
|
||
|
||
| 组件 | 作用 |
|
||
|------|------|
|
||
| `subs` | 按 taskID 索引的订阅者列表 |
|
||
| `allSubs` | 全局订阅者（接收所有事件） |
|
||
| `sync.RWMutex` | 读写锁保护并发访问 |
|
||
|
||
### TaskEvent 数据结构
|
||
|
||
```go
|
||
type TaskEvent struct {
|
||
TaskID string `json:"task_id"`
|
||
ProjectID string `json:"project_id,omitempty"`
|
||
Status string `json:"status"` // pending|running|saving|completed|failed
|
||
Stage string `json:"stage,omitempty"` // prompt_builder|asset_generator|...
|
||
Progress int `json:"progress"` // 0-100
|
||
Error string `json:"error,omitempty"`
|
||
}
|
||
```
|
||
|
||
---
|
||
|
||
## 订阅模式
|
||
|
||
### Subscribe(taskID) — 任务级订阅
|
||
|
||
```go
|
||
func (b *Broker) Subscribe(taskID string) <-chan TaskEvent {
|
||
ch := make(chan TaskEvent, 16) // 有界缓冲，容量 16
|
||
b.mu.Lock()
|
||
b.subs[taskID] = append(b.subs[taskID], ch)
|
||
b.mu.Unlock()
|
||
return ch
|
||
}
|
||
```
|
||
|
||
- 用于 `GET /api/v1/tasks/:taskId/stream`
|
||
- 仅接收指定任务的状态变更
|
||
- Buffer 容量 **16**，足够应对正常进度更新频率
|
||
|
||
### SubscribeAll() — 全局订阅
|
||
|
||
```go
|
||
func (b *Broker) SubscribeAll() <-chan TaskEvent {
|
||
ch := make(chan TaskEvent, 64) // 有界缓冲，容量 64
|
||
b.mu.Lock()
|
||
b.allSubs = append(b.allSubs, ch)
|
||
b.mu.Unlock()
|
||
return ch
|
||
}
|
||
```
|
||
|
||
- 用于 `GET /api/v1/projects/:projectId/stream`
|
||
- 接收所有任务的事件，在 Handler 层按 projectID 过滤
|
||
- Buffer 容量 **64**，因为全局事件量更大
|
||
|
||
---
|
||
|
||
## Publish — 扇出分发
|
||
|
||
```go
|
||
func (b *Broker) Publish(taskID string, event TaskEvent) {
|
||
// 1. 发送到任务级订阅者
|
||
for _, ch := range subs {
|
||
select {
|
||
case ch <- event:
|
||
default: // 满则丢弃，非阻塞
|
||
slog.Warn("subscriber buffer full, dropping event")
|
||
}
|
||
}
|
||
// 2. 发送到全局订阅者
|
||
for _, ch := range allSubs {
|
||
select { ... }
|
||
}
|
||
}
|
||
```
|
||
|
||
**关键特性**：
|
||
- **Fan-out**：同时发送到 task 级和 global 级订阅者
|
||
- **Non-blocking send**：使用 `select default` 防止慢消费者阻塞发布方
|
||
- **慢消费者丢弃**：缓冲满时静默丢弃，保证 Pipeline 不被 SSE 拖慢
|
||
|
||
---
|
||
|
||
## SSE Handler
|
||
|
||
### Stream — 任务级流
|
||
|
||
```
|
||
GET /api/v1/tasks/:taskId/stream
|
||
```
|
||
|
||
```go
|
||
func (h *SSEHandler) Stream(c *gin.Context) {
|
||
// 1. 设置 SSE 响应头
|
||
c.Header("Content-Type", "text/event-stream")
|
||
c.Header("Cache-Control", "no-cache")
|
||
c.Header("X-Accel-Buffering", "no") // 禁用 nginx 缓冲
|
||
|
||
// 2. 订阅事件
|
||
ch := h.broker.Subscribe(taskID)
|
||
defer h.broker.Unsubscribe(taskID, ch)
|
||
|
||
// 3. 事件循环
|
||
for {
|
||
select {
|
||
case event := <-ch:
|
||
c.SSEvent("status", event)
|
||
c.Writer.Flush()
|
||
// 终态自动关闭
|
||
if event.Status == "completed" || event.Status == "failed" {
|
||
return
|
||
}
|
||
case <-c.Request.Context().Done():
|
||
return // 客户端断开
|
||
}
|
||
}
|
||
}
|
||
```
|
||
|
||
**SSE 响应头**：
|
||
|
||
| Header | 值 | 作用 |
|
||
|--------|---|------|
|
||
| `Content-Type` | `text/event-stream` | 标识 SSE 流 |
|
||
| `Cache-Control` | `no-cache` | 禁用缓存 |
|
||
| `X-Accel-Buffering` | `no` | 禁用 nginx 代理缓冲 |
|
||
|
||
### StreamProject — 工程级流
|
||
|
||
```
|
||
GET /api/v1/projects/:projectId/stream
|
||
```
|
||
|
||
- 使用 `SubscribeAll()` 订阅全局事件
|
||
- 在 Handler 层按 `event.ProjectID != projectID` 过滤
|
||
- 工程级流不会因单个任务完成而关闭，持续监听新任务
|
||
|
||
---
|
||
|
||
## 数据流全景
|
||
|
||
```mermaid
|
||
sequenceDiagram
|
||
participant P as Pipeline
|
||
participant DB as Database
|
||
participant EB as EventBus
|
||
participant SSE as SSE Handler
|
||
participant B as Browser
|
||
|
||
P->>DB: updateTaskInDB(status, progress)
|
||
P->>EB: Publish(taskID, event)
|
||
EB->>SSE: ch <- event
|
||
SSE->>B: data: {"status":"running","progress":45}
|
||
Note over B: EventSource.onmessage()
|
||
P->>DB: updateTaskInDB(completed)
|
||
P->>EB: Publish(taskID, terminal event)
|
||
EB->>SSE: ch <- event
|
||
SSE->>B: data: {"status":"completed","progress":100}
|
||
Note over SSE: 终态，关闭连接
|
||
```
|
||
|
||
---
|
||
|
||
## 容错设计
|
||
|
||
| 场景 | 处理方式 |
|
||
|------|---------|
|
||
| 慢消费者 | Buffer 满时丢弃事件，Pipeline 不阻塞 |
|
||
| 客户端断开 | `c.Request.Context().Done()` 触发，自动 Unsubscribe |
|
||
| 终态到达 | completed/failed 后自动关闭 SSE 连接 |
|
||
| 无订阅者 | Publish 静默返回，不报错 |
|
||
| Broker 关闭 | Close() 关闭所有 channel，SSE 循环退出 |
|
||
|
||
---
|
||
|
||
## 关联文档
|
||
|
||
- [[05-生成管线]] — Pipeline 中的进度回调
|
||
- [[07-可观测性]] — SSE 连接的监控
|
||
- [[10-中间件链]] — SSE 端点的中间件配置
|