447 lines
9.0 KiB
Markdown
447 lines
9.0 KiB
Markdown
|
# SSE 流式传输
|
||
|
|
|
||
|
|
## 概述
|
||
|
|
|
||
|
|
Server-Sent Events (SSE) 是 PR-Helper 的核心通信机制，用于实时推送克隆进度、PR 生成和代码审查结果。
|
||
|
|
|
||
|
|
## 为什么选择 SSE
|
||
|
|
|
||
|
|
### 对比 WebSocket
|
||
|
|
|
||
|
|
| 特性 | SSE | WebSocket |
|
||
|
|
|------|-----|-----------|
|
||
|
|
| 方向 | 单向（服务器→客户端） | 双向 |
|
||
|
|
| 协议 | HTTP | 独立协议 |
|
||
|
|
| 实现 | 简单 | 复杂 |
|
||
|
|
| 重连 | 自动 | 手动 |
|
||
|
|
| 兼容性 | 好 | 需要升级 |
|
||
|
|
|
||
|
|
### 适用场景
|
||
|
|
|
||
|
|
- 服务器推送数据到客户端
|
||
|
|
- 不需要客户端频繁发送数据
|
||
|
|
- 需要 HTTP 兼容性
|
||
|
|
- 需要自动重连
|
||
|
|
|
||
|
|
## SSE 协议
|
||
|
|
|
||
|
|
### 基本格式
|
||
|
|
|
||
|
|
```
|
||
|
|
event: message_type
|
||
|
|
data: {"key": "value"}
|
||
|
|
|
||
|
|
```
|
||
|
|
|
||
|
|
- `event:` 事件类型（可选）
|
||
|
|
- `data:` 事件数据（JSON 字符串）
|
||
|
|
- 空行分隔事件
|
||
|
|
|
||
|
|
### 多行数据
|
||
|
|
|
||
|
|
```
|
||
|
|
event: content
|
||
|
|
data: {"content": "第一行\n第二行"}
|
||
|
|
```
|
||
|
|
|
||
|
|
## 后端实现
|
||
|
|
|
||
|
|
### 设置响应头
|
||
|
|
|
||
|
|
```go
|
||
|
|
c.Header("Content-Type", "text/event-stream")
|
||
|
|
c.Header("Cache-Control", "no-cache")
|
||
|
|
c.Header("Connection", "keep-alive")
|
||
|
|
c.Header("X-Accel-Buffering", "no")
|
||
|
|
c.Status(http.StatusOK)
|
||
|
|
```
|
||
|
|
|
||
|
|
### 发送事件
|
||
|
|
|
||
|
|
```go
|
||
|
|
sendEvent := func(event string, data interface{}) {
|
||
|
|
jsonData, err := json.Marshal(data)
|
||
|
|
if err != nil {
|
||
|
|
jsonData = []byte(`{"error":"序列化事件数据失败"}`)
|
||
|
|
}
|
||
|
|
fmt.Fprintf(c.Writer, "event: %s\ndata: %s\n\n", event, jsonData)
|
||
|
|
flusher.Flush()
|
||
|
|
}
|
||
|
|
```
|
||
|
|
|
||
|
|
### 获取 Flusher
|
||
|
|
|
||
|
|
```go
|
||
|
|
flusher, ok := c.Writer.(http.Flusher)
|
||
|
|
if !ok {
|
||
|
|
c.JSON(http.StatusInternalServerError, gin.H{"error": "服务器不支持流式传输"})
|
||
|
|
return
|
||
|
|
}
|
||
|
|
```
|
||
|
|
|
||
|
|
## 前端实现
|
||
|
|
|
||
|
|
### SSE 客户端
|
||
|
|
|
||
|
|
```javascript
|
||
|
|
const SSE = {
|
||
|
|
async post(url, body, handlers = {}) {
|
||
|
|
const controller = new AbortController();
|
||
|
|
|
||
|
|
const run = async () => {
|
||
|
|
try {
|
||
|
|
const resp = await fetch(url, {
|
||
|
|
method: 'POST',
|
||
|
|
headers: { 'Content-Type': 'application/json' },
|
||
|
|
credentials: 'same-origin',
|
||
|
|
body: JSON.stringify(body),
|
||
|
|
signal: controller.signal,
|
||
|
|
});
|
||
|
|
|
||
|
|
if (!resp.ok) {
|
||
|
|
const errText = await resp.text();
|
||
|
|
// 处理错误...
|
||
|
|
return;
|
||
|
|
}
|
||
|
|
|
||
|
|
const reader = resp.body.getReader();
|
||
|
|
const decoder = new TextDecoder();
|
||
|
|
let buffer = '';
|
||
|
|
let currentEvent = '';
|
||
|
|
|
||
|
|
while (true) {
|
||
|
|
const { done, value } = await reader.read();
|
||
|
|
if (done) break;
|
||
|
|
|
||
|
|
buffer += decoder.decode(value, { stream: true });
|
||
|
|
const lines = buffer.split('\n');
|
||
|
|
buffer = lines.pop() || '';
|
||
|
|
|
||
|
|
for (const line of lines) {
|
||
|
|
if (line.startsWith('event: ')) {
|
||
|
|
currentEvent = line.slice(7).trim();
|
||
|
|
} else if (line.startsWith('data: ')) {
|
||
|
|
const raw = line.slice(6);
|
||
|
|
let data;
|
||
|
|
try {
|
||
|
|
data = JSON.parse(raw);
|
||
|
|
} catch (_) {
|
||
|
|
data = raw;
|
||
|
|
}
|
||
|
|
|
||
|
|
if (currentEvent && handlers[currentEvent]) {
|
||
|
|
handlers[currentEvent](data);
|
||
|
|
}
|
||
|
|
}
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
if (handlers.done) handlers.done();
|
||
|
|
} catch (err) {
|
||
|
|
if (err.name === 'AbortError') return;
|
||
|
|
if (handlers.error) handlers.error({ message: err.message });
|
||
|
|
}
|
||
|
|
};
|
||
|
|
|
||
|
|
run();
|
||
|
|
return { abort: () => controller.abort() };
|
||
|
|
}
|
||
|
|
};
|
||
|
|
```
|
||
|
|
|
||
|
|
### 使用示例
|
||
|
|
|
||
|
|
```javascript
|
||
|
|
SSE.post('/api/repos/1/review', {
|
||
|
|
base: 'main',
|
||
|
|
head: 'feature',
|
||
|
|
top_n: 20,
|
||
|
|
concurrency: 5
|
||
|
|
}, {
|
||
|
|
// 开始事件
|
||
|
|
start: (data) => {
|
||
|
|
console.log(`开始审查 ${data.total_files} 个文件`);
|
||
|
|
updateProgress(0, data.reviewed_files);
|
||
|
|
},
|
||
|
|
|
||
|
|
// 文件开始
|
||
|
|
file_start: (data) => {
|
||
|
|
console.log(`正在审查 ${data.file} (${data.index}/${data.total})`);
|
||
|
|
highlightCurrentFile(data.file);
|
||
|
|
},
|
||
|
|
|
||
|
|
// 内容流
|
||
|
|
content: (data) => {
|
||
|
|
appendToOutput(data.content);
|
||
|
|
},
|
||
|
|
|
||
|
|
// 审查建议
|
||
|
|
suggestion: (data) => {
|
||
|
|
showSuggestion(data.file, data.severity, data.content);
|
||
|
|
},
|
||
|
|
|
||
|
|
// 文件结束
|
||
|
|
file_end: (data) => {
|
||
|
|
markFileComplete(data.file);
|
||
|
|
},
|
||
|
|
|
||
|
|
// 汇总
|
||
|
|
summary: (data) => {
|
||
|
|
showSummary(data.score, data.overall, data.findings, data.recommendations);
|
||
|
|
},
|
||
|
|
|
||
|
|
// 分析保存
|
||
|
|
analysis_saved: (data) => {
|
||
|
|
console.log(`分析已保存，ID: ${data.analysis_id}`);
|
||
|
|
},
|
||
|
|
|
||
|
|
// 进度
|
||
|
|
progress: (data) => {
|
||
|
|
if (data.step === 'generating_summary') {
|
||
|
|
showSpinner('生成汇总中...');
|
||
|
|
}
|
||
|
|
},
|
||
|
|
|
||
|
|
// 完成
|
||
|
|
done: () => {
|
||
|
|
console.log('审查完成');
|
||
|
|
hideSpinner();
|
||
|
|
},
|
||
|
|
|
||
|
|
// 错误
|
||
|
|
error: (data) => {
|
||
|
|
console.error('错误:', data.message);
|
||
|
|
showError(data.message);
|
||
|
|
}
|
||
|
|
});
|
||
|
|
```
|
||
|
|
|
||
|
|
## 事件类型
|
||
|
|
|
||
|
|
### PR 生成事件
|
||
|
|
|
||
|
|
| 事件 | 数据 | 说明 |
|
||
|
|
|------|------|------|
|
||
|
|
| `content` | `{content: string}` | Markdown 片段 |
|
||
|
|
| `done` | `{content: ""}` | 完成 |
|
||
|
|
|
||
|
|
### 代码审查事件
|
||
|
|
|
||
|
|
| 事件 | 数据 | 说明 |
|
||
|
|
|------|------|------|
|
||
|
|
| `start` | `{total_files, reviewed_files, top_n}` | 开始 |
|
||
|
|
| `file_start` | `{file, index, total}` | 文件开始 |
|
||
|
|
| `content` | `{content: string}` | LLM 输出 |
|
||
|
|
| `suggestion` | `{file, severity, content}` | 建议 |
|
||
|
|
| `file_end` | `{file}` | 文件结束 |
|
||
|
|
| `summary` | `{score, overall, findings, recommendations}` | 汇总 |
|
||
|
|
| `progress` | `{step: string}` | 进度 |
|
||
|
|
| `error` | `{message: string}` | 错误 |
|
||
|
|
| `analysis_saved` | `{analysis_id}` | 保存完成 |
|
||
|
|
| `done` | `{content: ""}` | 完成 |
|
||
|
|
|
||
|
|
### 克隆事件
|
||
|
|
|
||
|
|
| 事件 | 数据 | 说明 |
|
||
|
|
|------|------|------|
|
||
|
|
| `progress` | `{step, current, total}` | 进度 |
|
||
|
|
| `error` | `{message: string}` | 错误 |
|
||
|
|
| `done` | `{repo_id, branches, tags, commit_num}` | 完成 |
|
||
|
|
|
||
|
|
## 错误处理
|
||
|
|
|
||
|
|
### 后端错误
|
||
|
|
|
||
|
|
```go
|
||
|
|
if err != nil {
|
||
|
|
sendEvent("error", map[string]interface{}{
|
||
|
|
"message": err.Error(),
|
||
|
|
})
|
||
|
|
return
|
||
|
|
}
|
||
|
|
```
|
||
|
|
|
||
|
|
### 前端错误
|
||
|
|
|
||
|
|
```javascript
|
||
|
|
error: (data) => {
|
||
|
|
// 显示错误消息
|
||
|
|
showErrorToast(data.message);
|
||
|
|
|
||
|
|
// 重置 UI
|
||
|
|
resetProgressBar();
|
||
|
|
hideSpinner();
|
||
|
|
}
|
||
|
|
```
|
||
|
|
|
||
|
|
### 网络错误
|
||
|
|
|
||
|
|
```javascript
|
||
|
|
catch (err) {
|
||
|
|
if (err.name === 'AbortError') {
|
||
|
|
// 用户中断，忽略
|
||
|
|
return;
|
||
|
|
}
|
||
|
|
if (handlers.error) {
|
||
|
|
handlers.error({ message: err.message });
|
||
|
|
}
|
||
|
|
}
|
||
|
|
```
|
||
|
|
|
||
|
|
## 中断请求
|
||
|
|
|
||
|
|
### 前端中断
|
||
|
|
|
||
|
|
```javascript
|
||
|
|
const controller = SSE.post(url, body, handlers);
|
||
|
|
|
||
|
|
// 用户点击取消按钮
|
||
|
|
cancelButton.onclick = () => {
|
||
|
|
controller.abort();
|
||
|
|
};
|
||
|
|
```
|
||
|
|
|
||
|
|
### 后端处理
|
||
|
|
|
||
|
|
- 客户端断开连接时，Gin 会检测到
|
||
|
|
- 服务层应检查 context 取消
|
||
|
|
- 长时间运行的操作应支持取消
|
||
|
|
|
||
|
|
## 性能优化
|
||
|
|
|
||
|
|
### 缓冲控制
|
||
|
|
|
||
|
|
```
|
||
|
|
X-Accel-Buffering: no
|
||
|
|
```
|
||
|
|
|
||
|
|
- 禁用 Nginx 缓冲
|
||
|
|
- 确保事件立即发送
|
||
|
|
|
||
|
|
### 批量发送
|
||
|
|
|
||
|
|
- 避免频繁发送小事件
|
||
|
|
- 合并相关数据
|
||
|
|
|
||
|
|
### 压缩
|
||
|
|
|
||
|
|
- SSE 不支持 gzip 压缩
|
||
|
|
- 数据量大时考虑压缩正文
|
||
|
|
|
||
|
|
## 并发安全
|
||
|
|
|
||
|
|
### 线程安全回调
|
||
|
|
|
||
|
|
```go
|
||
|
|
safeCallback := callback
|
||
|
|
if callback != nil {
|
||
|
|
safeCallback = func(event string, data interface{}) {
|
||
|
|
mu.Lock()
|
||
|
|
defer mu.Unlock()
|
||
|
|
callback(event, data)
|
||
|
|
}
|
||
|
|
}
|
||
|
|
```
|
||
|
|
|
||
|
|
- 使用互斥锁保护回调
|
||
|
|
- 避免并发写入
|
||
|
|
|
||
|
|
### Goroutine 管理
|
||
|
|
|
||
|
|
```go
|
||
|
|
var wg sync.WaitGroup
|
||
|
|
|
||
|
|
for i, file := range files {
|
||
|
|
wg.Add(1)
|
||
|
|
go func(idx int, f FileDiff) {
|
||
|
|
defer wg.Done()
|
||
|
|
// ...
|
||
|
|
}(i, file)
|
||
|
|
}
|
||
|
|
|
||
|
|
wg.Wait()
|
||
|
|
```
|
||
|
|
|
||
|
|
- 等待所有 goroutine 完成
|
||
|
|
- 避免资源泄漏
|
||
|
|
|
||
|
|
## 测试
|
||
|
|
|
||
|
|
### 手动测试
|
||
|
|
|
||
|
|
```bash
|
||
|
|
curl -N -X POST http://localhost:8080/api/repos/1/review \
|
||
|
|
-H "Content-Type: application/json" \
|
||
|
|
-d '{"base":"main","head":"feature"}'
|
||
|
|
```
|
||
|
|
|
||
|
|
### 自动化测试
|
||
|
|
|
||
|
|
```go
|
||
|
|
func TestSSEStream(t *testing.T) {
|
||
|
|
// 创建测试服务器
|
||
|
|
// 发送请求
|
||
|
|
// 读取事件流
|
||
|
|
// 验证事件序列
|
||
|
|
}
|
||
|
|
```
|
||
|
|
|
||
|
|
## 监控
|
||
|
|
|
||
|
|
### 关键指标
|
||
|
|
|
||
|
|
- 连接数
|
||
|
|
- 事件发送速率
|
||
|
|
- 错误率
|
||
|
|
- 响应时间
|
||
|
|
|
||
|
|
### 日志
|
||
|
|
|
||
|
|
```go
|
||
|
|
log.Printf("SSE connected: %s", c.ClientIP())
|
||
|
|
log.Printf("SSE event: %s", event)
|
||
|
|
log.Printf("SSE disconnected: %s", c.ClientIP())
|
||
|
|
```
|
||
|
|
|
||
|
|
## 安全考虑
|
||
|
|
|
||
|
|
### 认证
|
||
|
|
|
||
|
|
- 所有 SSE 端点需要认证
|
||
|
|
- 使用会话 Cookie
|
||
|
|
|
||
|
|
### 速率限制
|
||
|
|
|
||
|
|
- 限制并发 SSE 连接数
|
||
|
|
- 限制事件发送频率
|
||
|
|
|
||
|
|
### 数据验证
|
||
|
|
|
||
|
|
- 验证输入参数
|
||
|
|
- 防止注入攻击
|
||
|
|
|
||
|
|
## 故障排查
|
||
|
|
|
||
|
|
### 常见问题
|
||
|
|
|
||
|
|
1. **事件不发送**
|
||
|
|
- 检查 Flusher 是否可用
|
||
|
|
- 确认响应头设置正确
|
||
|
|
- 检查 Nginx 缓冲配置
|
||
|
|
|
||
|
|
2. **连接断开**
|
||
|
|
- 检查超时设置
|
||
|
|
- 确认网络稳定
|
||
|
|
- 查看错误日志
|
||
|
|
|
||
|
|
3. **数据乱码**
|
||
|
|
- 确认 JSON 序列化正确
|
||
|
|
- 检查字符编码
|
||
|
|
- 验证事件格式
|
||
|
|
|
||
|
|
### 调试工具
|
||
|
|
|
||
|
|
- 浏览器开发者工具 Network 面板
|
||
|
|
- curl 命令行测试
|
||
|
|
- Wireshark 抓包
|