当前位置：首页 > news >正文

gin中sse流式服务

news 2025/9/17 8:02:08

例子

首先安装必要的依赖

go get -u github.com/gin-gonic/gin

实现代码

package mainimport ("fmt""math/rand""net/http""time""github.com/gin-gonic/gin"
)// SSEHeaders 设置SSE所需的响应头
func SSEHeaders() gin.HandlerFunc {return func(c *gin.Context) {c.Writer.Header().Set("Content-Type", "text/event-stream")c.Writer.Header().Set("Cache-Control", "no-cache")c.Writer.Header().Set("Connection", "keep-alive")c.Writer.Header().Set("Access-Control-Allow-Origin", "*")c.Next()}
}func main() {r := gin.Default()// SSE 流式接口r.GET("/sse", SSEHeaders(), func(c *gin.Context) {// 获取客户端传递的参数（可选）message := c.Query("message")if message == "" {message = "默认消息"}// 创建通道用于监听客户端断开连接clientGone := c.Writer.CloseNotify()// 设置响应头c.Writer.Flush()// 循环发送事件for i := 1; i <= 10; i++ {select {case <-clientGone:// 客户端断开连接fmt.Println("客户端断开连接")returndefault:// 生成随机数据作为示例data := fmt.Sprintf("%s - 事件 #%d: 随机数: %d", message, i, rand.Intn(100))// 发送事件，格式为: "data: <内容>\n\n"// 可以添加事件类型: "event: <事件类型>\n"// 可以添加事件ID: "id: <ID>\n"c.SSEvent("message", map[string]interface{}{"data":      data,"timestamp": time.Now().Format(time.RFC3339),"count":     i,})// 刷新缓冲区，确保数据立即发送到客户端c.Writer.Flush()// 等待一段时间time.Sleep(2 * time.Second)}}// 发送结束事件c.SSEvent("end", "流式传输结束")c.Writer.Flush()})// 提供测试页面r.GET("/", func(c *gin.Context) {c.HTML(http.StatusOK, "index.html", nil)})// 加载HTML模板r.LoadHTMLFiles("index.html")fmt.Println("服务器启动在 :8080")r.Run(":8080")
}

创建测试页面

<!DOCTYPE html>
<html>
<head><title>SSE 测试</title>
</head>
<body><h1>SSE 流式数据测试</h1><div><input type="text" id="message" placeholder="输入消息" value="测试消息"><button onclick="startSSE()">开始接收数据</button><button onclick="stopSSE()">停止接收</button></div><div id="output" style="margin-top: 20px; border: 1px solid #ccc; padding: 10px; height: 300px; overflow-y: scroll;"></div><script>let eventSource = null;function startSSE() {const message = document.getElementById('message').value;const output = document.getElementById('output');// 如果已有连接，先关闭if (eventSource) {eventSource.close();}// 创建新的 EventSource 连接eventSource = new EventSource(`/sse?message=${encodeURIComponent(message)}`);output.innerHTML += '开始连接...<br>';// 监听 message 事件eventSource.addEventListener('message', function(e) {const data = JSON.parse(e.data);output.innerHTML += `收到消息: ${data.data} (时间: ${data.timestamp})<br>`;output.scrollTop = output.scrollHeight;});// 监听 end 事件eventSource.addEventListener('end', function(e) {output.innerHTML += '服务器通知: 传输结束<br>';eventSource.close();});// 错误处理eventSource.onerror = function(e) {output.innerHTML += '连接错误或已关闭<br>';eventSource.close();};}function stopSSE() {if (eventSource) {eventSource.close();document.getElementById('output').innerHTML += '已手动停止连接<br>';}}</script>
</body>
</html>

高级用法 - 支持多个客户端

// 全局通道用于广播消息
var broadcast = make(chan string)// 广播消息的goroutine
func broadcaster() {for {msg := <-broadcast// 这里可以维护客户端连接列表并广播消息fmt.Println("广播消息:", msg)}
}// 多个客户端的SSE处理
r.GET("/sse-broadcast", SSEHeaders(), func(c *gin.Context) {clientGone := c.Writer.CloseNotify()c.Writer.Flush()// 监听广播通道for {select {case <-clientGone:fmt.Println("客户端断开连接")returncase msg := <-broadcast:c.SSEvent("broadcast", msg)c.Writer.Flush()}}
})

注意事项
连接管理: SSE 连接是长连接，需要妥善管理连接的生命周期
错误处理: 需要处理客户端意外断开的情况
性能考虑: 大量并发连接时需要考虑服务器资源
超时设置: 可以设置适当的超时时间避免资源浪费

Gin SSE 流式服务中添加错误处理

package mainimport ("fmt""math/rand""net/http""time""github.com/gin-gonic/gin"
)// SSEHeaders 设置SSE所需的响应头
func SSEHeaders() gin.HandlerFunc {return func(c *gin.Context) {c.Writer.Header().Set("Content-Type", "text/event-stream")c.Writer.Header().Set("Cache-Control", "no-cache")c.Writer.Header().Set("Connection", "keep-alive")c.Writer.Header().Set("Access-Control-Allow-Origin", "*")c.Next()}
}func main() {r := gin.Default()// SSE 流式接口r.GET("/sse", SSEHeaders(), func(c *gin.Context) {// 获取客户端传递的参数message := c.Query("message")countStr := c.Query("count")// 参数验证if message == "" {// 发送错误事件并返回c.SSEvent("error", "参数'message'不能为空")c.Writer.Flush()return}// 转换并验证count参数var count intif countStr != "" {if _, err := fmt.Sscanf(countStr, "%d", &count); err != nil || count <= 0 {c.SSEvent("error", "参数'count'必须是正整数")c.Writer.Flush()return}} else {count = 10 // 默认值}// 创建通道用于监听客户端断开连接clientGone := c.Writer.CloseNotify()c.Writer.Flush()for i := 1; i <= count; i++ {select {case <-clientGone:fmt.Println("客户端断开连接")returndefault:// 模拟可能的错误if rand.Intn(20) == 1 { // 5% 的概率模拟错误c.SSEvent("error", map[string]interface{}{"code":    "RANDOM_ERROR","message": "随机模拟的错误发生","step":    i,})c.Writer.Flush()continue // 继续发送，而不是返回}// 正常数据data := fmt.Sprintf("%s - 事件 #%d: 随机数: %d", message, i, rand.Intn(100))c.SSEvent("message", map[string]interface{}{"data":      data,"timestamp": time.Now().Format(time.RFC3339),"count":     i,"total":     count,})c.Writer.Flush()time.Sleep(1 * time.Second)}}// 发送结束事件c.SSEvent("end", "流式传输结束")c.Writer.Flush()})// 模拟可能抛出panic的SSE接口r.GET("/sse-risky", SSEHeaders(), func(c *gin.Context) {// 使用defer和recover捕获panicdefer func() {if r := recover(); r != nil {fmt.Printf("捕获到panic: %v\n", r)c.SSEvent("error", map[string]interface{}{"code":    "INTERNAL_ERROR","message": "服务器内部错误","details": fmt.Sprintf("%v", r),})c.Writer.Flush()}}()clientGone := c.Writer.CloseNotify()c.Writer.Flush()for i := 1; i <= 5; i++ {select {case <-clientGone:returndefault:// 模拟可能抛出panic的情况if i == 3 {panic("模拟的第3次迭代panic")}c.SSEvent("message", map[string]interface{}{"data":      fmt.Sprintf("安全数据 #%d", i),"timestamp": time.Now().Format(time.RFC3339),})c.Writer.Flush()time.Sleep(1 * time.Second)}}})// 提供测试页面r.GET("/", func(c *gin.Context) {c.HTML(http.StatusOK, "index.html", nil)})r.LoadHTMLFiles("index.html")fmt.Println("服务器启动在 :8080")r.Run(":8080")
}

改进的 HTML 测试页面

<!DOCTYPE html>
<html>
<head><title>SSE 错误处理测试</title>
</head>
<body><h1>SSE 错误处理测试</h1><div><input type="text" id="message" placeholder="输入消息" value="测试消息"><input type="number" id="count" placeholder="消息数量" value="10" min="1"><button onclick="startSSE()">开始接收数据</button><button onclick="startRiskySSE()">测试危险接口</button><button onclick="stopSSE()">停止接收</button></div><div id="output" style="margin-top: 20px; border: 1px solid #ccc; padding: 10px; height: 300px; overflow-y: scroll;"></div><script>let eventSource = null;function startSSE() {const message = document.getElementById('message').value;const count = document.getElementById('count').value;const output = document.getElementById('output');if (eventSource) {eventSource.close();}output.innerHTML += `开始连接: message=${message}, count=${count}<br>`;eventSource = new EventSource(`/sse?message=${encodeURIComponent(message)}&count=${count}`);// 监听消息事件eventSource.addEventListener('message', function(e) {const data = JSON.parse(e.data);output.innerHTML += `✅ 消息: ${data.data} (${data.count}/${data.total})<br>`;output.scrollTop = output.scrollHeight;});// 监听错误事件eventSource.addEventListener('error', function(e) {const errorData = JSON.parse(e.data);output.innerHTML += `❌ 错误: ${errorData.message} (代码: ${errorData.code})<br>`;output.scrollTop = output.scrollHeight;});// 监听结束事件eventSource.addEventListener('end', function(e) {output.innerHTML += '🏁 传输结束<br>';eventSource.close();});// 连接错误处理eventSource.onerror = function(e) {if (e.eventPhase === EventSource.CLOSED) {output.innerHTML += '连接已关闭<br>';} else {output.innerHTML += '连接错误<br>';}};}function startRiskySSE() {const output = document.getElementById('output');if (eventSource) {eventSource.close();}output.innerHTML += '开始测试危险接口...<br>';eventSource = new EventSource('/sse-risky');eventSource.addEventListener('message', function(e) {const data = JSON.parse(e.data);output.innerHTML += `✅ 安全数据: ${data.data}<br>`;});eventSource.addEventListener('error', function(e) {const errorData = JSON.parse(e.data);output.innerHTML += `💥 严重错误: ${errorData.message}<br>`;output.scrollTop = output.scrollHeight;});eventSource.onerror = function(e) {output.innerHTML += '连接错误<br>';};}function stopSSE() {if (eventSource) {eventSource.close();document.getElementById('output').innerHTML += '已手动停止连接<br>';}}</script>
</body>
</html>

更结构化的错误处理方式

// 定义错误类型
type SSEError struct {Code    string `json:"code"`Message string `json:"message"`Details string `json:"details,omitempty"`
}// 发送错误辅助函数
func sendSSEError(c *gin.Context, code, message, details string) {c.SSEvent("error", SSEError{Code:    code,Message: message,Details: details,})c.Writer.Flush()
}// 使用结构化的错误处理
r.GET("/sse-structured", SSEHeaders(), func(c *gin.Context) {// 参数验证if c.Query("required_param") == "" {sendSSEError(c, "MISSING_PARAM", "缺少必要参数", "required_param 参数不能为空")return}clientGone := c.Writer.CloseNotify()c.Writer.Flush()for i := 1; i <= 5; i++ {select {case <-clientGone:returndefault:// 业务逻辑...}}
})

常见的错误场景处理

// 处理数据库错误
func handleDatabaseSSE(c *gin.Context) {defer func() {if r := recover(); r != nil {sendSSEError(c, "DATABASE_ERROR", "数据库操作失败", fmt.Sprintf("%v", r))}}()// 模拟数据库操作if rand.Intn(10) == 1 {panic("数据库连接超时")}
}// 处理外部API错误
func handleExternalAPISSE(c *gin.Context) {// 模拟外部API调用if rand.Intn(8) == 1 {sendSSEError(c, "EXTERNAL_API_ERROR", "外部服务不可用", "第三方API响应超时")return}
}

关键错误处理要点
参数验证: 在开始流式传输前验证所有参数
Panic 恢复: 使用 defer 和 recover() 捕获 panic
连接状态检查: 定期检查客户端是否仍然连接
错误事件格式: 使用结构化的错误信息方便客户端解析
资源清理: 确保在错误发生时正确清理资源
这样的错误处理机制可以确保 SSE 服务在出现问题时能够优雅地处理，并向客户端提供有意义的错误信息。

Gin 作为客户端访问外部 SSE 接口

Gin 框架通常用于构建 HTTP 服务器。若需在 Gin 服务端内作为客户端访问外部 SSE 流式接口（例如访问大模型 API），可以使用 net/http 包，并结合 Gin 的路由处理返回数据或错误。
基本步骤：

创建 HTTP 请求：使用 http.NewRequest 创建 GET 请求（SSE 通常是 GET 请求），并设置必要的 Header（如 Accept: text/event-stream）。
发起请求并处理响应：使用 http.Client 发起请求，读取 text/event-stream 响应体。
解析 SSE 数据：SSE 数据格式通常为 data: {message}\n\n，需按此解析。
错误处理：处理网络错误、HTTP 状态码错误等。
在 Gin 路由中返回数据或错误：将获取到的 SSE 数据或错误信息通过 Gin 的上下文返回给客户端。
代码示例：

package mainimport ("bufio""fmt""io""net/http""strings""github.com/gin-gonic/gin"
)// 示例：Gin 处理函数，用于访问外部 SSE 流并返回数据
func handleSSEProxy(c *gin.Context) {// 1. 创建 HTTP 请求访问外部 SSE 端点req, err := http.NewRequest("GET", "https://example.com/api/sse-stream", nil)if err != nil {c.JSON(http.StatusInternalServerError, gin.H{"error": "创建请求失败", "details": err.Error()})return}// 2. 设置必要的请求头，表明接受事件流req.Header.Set("Accept", "text/event-stream")// 3. 发起 HTTP 请求client := &http.Client{}resp, err := client.Do(req)if err != nil {c.JSON(http.StatusInternalServerError, gin.H{"error": "请求失败", "details": err.Error()})return}defer resp.Body.Close()// 4. 检查响应状态码if resp.StatusCode != http.StatusOK {c.JSON(resp.StatusCode, gin.H{"error": "远程服务器返回错误", "status": resp.Status})return}// 5. 设置 Gin 客户端响应头为事件流（如需将流直接透传）c.Header("Content-Type", "text/event-stream")c.Header("Cache-Control", "no-cache")c.Header("Connection", "keep-alive")c.Writer.Flush() // 立即刷新头部到客户端// 6. 逐行读取 SSE 响应并发送到 Gin 客户端reader := bufio.NewReader(resp.Body)for {line, err := reader.ReadString('\n')if err != nil {if err == io.EOF {break // 流结束}// 处理读取错误，记录日志或返回错误fmt.Printf("读取流错误: %v\n", err)c.SSEvent("error", gin.H{"message": "读取数据流失败"}) // 可向客户端发送错误事件break}// 处理 SSE 数据行if strings.HasPrefix(line, "data:") {// 提取数据部分并发送给客户端data := strings.TrimPrefix(line, "data:")data = strings.TrimSpace(data)c.SSEvent("message", data) // 使用 Gin 的 SSEvent 方法发送事件c.Writer.Flush()}// 可以根据需要处理其他 SSE 字段，如 event, id 等}
}func main() {r := gin.Default()r.GET("/proxy-sse", handleSSEProxy) // 访问此路由会代理外部 SSE 流r.Run(":8080")
}

关键点与错误处理
SSE 格式解析：SSE 消息以 data: 开头，以两个换行符 \n\n 结束一个事件。代码需按此解析。
错误处理：
网络请求错误：检查 client.Do 和 reader.ReadString 的错误。
非 200 状态码：检查 resp.StatusCode。
向客户端传递错误：可使用 c.JSON 直接返回错误（会中断流），或在 SSE 流中发送特定格式的错误事件（如 c.SSEvent(“error”, …)），这取决于你的设计。
流式传输：如需将外部 SSE 流直接透传给 Gin 的客户端，需设置正确的响应头并手动刷新 (c.Writer.Flush())。
超时控制：考虑使用 context.Context 设置请求超时，避免长时间阻塞。
性能与资源：流式连接是长连接，需注意管理（如连接数控制、及时关闭响应体）
在这里插入图片描述

ctx.Render

在 Gin 框架中，ctx.Render(-1, sse.Event{…}) 这行代码用于向客户端发送一个 Server-Sent Event (SSE) 消息5。SSE 是一种允许服务器主动向客户端推送数据的技术

ctx.Render(-1, sse.Event{Event:      name,      // 事件类型，例如 "message", "update" 等Data:       message,   // 要发送的实际数据内容DataPrefix: false,     // 控制是否在每条数据行前自动添加 "data: " 前缀
})

ctx.Render: 这是 Gin 上下文的方法，用于渲染并发送响应。

-1: 通常 Render 方法的第一个参数是 HTTP 状态码（如 200）。-1 是一个特殊值，表示跳过设置 HTTP 状态码。这对于 SSE 这样的长连接流式响应很常见，因为连接会一直保持，我们可能已经在其他地方设置过状态码，或者不需要频繁设置。
sse.Event: 这是一个结构体，专门用于表示一个 SSE 事件。它通常包含以下字段：

Event: 指定事件的类型（字符串）。这是一个可选字段。客户端可以根据不同的事件类型绑定不同的监听器。如果未设置，客户端会触发默认的 message 事件。
Data: 要发送的实际数据（通常是字符串或可以被序列化为字符串的类型）。这是 SSE 消息的核心内容。
DataPrefix: 一个布尔值，控制 Gin 在发送 Data 内容时是否自动为每一行数据添加 "data: " 前缀。
■ 当 DataPrefix: false 时，Gin 不会自动添加 "data: " 前缀。这意味着你需要确保 message 参数本身已经是格式正确的 SSE 数据字符串（例如已经包含了 data: …\n\n）。
■ 当 DataPrefix: true 时（更常见且省心），Gin 会自动处理格式，为你发送的数据添加 "data: " 前缀和必要的换行符 (\n\n)。

DataPrefix: false 的注意事项
将 DataPrefix 设置为 false 意味着你需要手动确保 message 的格式完全符合 SSE 协议：
• 格式要求：SSE 数据字段应以 data: 开头，并以两个换行符 \n\n 结束。例如：

// 如果 DataPrefix: false, 你的 message 应该自己包含 "data: " 前缀和结尾的双换行
message := "data: This is a message\n\n"// 如果发送多行数据，每行都应以 "data: " 开头
multiLineMessage := "data: Line 1\ndata: Line 2\n\n"

潜在风险：如果 message 格式不正确（例如缺少 data: 前缀或结束符 \n\n），客户端可能无法正确解析该事件，导致推送失败。
使用场景：通常在你需要更精细地控制 SSE 数据格式，或者 message 已经是另一个库生成的完整 SSE 格式字符串时，才会设置 DataPrefix: false。绝大多数情况下，更简单和安全的方式是使用 DataPrefix: true（或让其默认为 true），让 Gin 帮你处理格式。

工作流程

构建事件对象：代码创建了一个 sse.Event 对象，并指定了事件类型、数据和前缀处理方式。
渲染和发送：ctx.Render(-1, …) 将这个事件对象发送给客户端。由于状态码是 -1，不会改变当前的 HTTP 状态码。
刷新缓冲区：通常在使用 SSE 时，在发送事件后需要调用 ctx.Writer.Flush() 来立即将数据刷新到客户端，而不是等待缓冲区满。这样可以确保消息实时推送。

ctx.Render(-1, sse.Event{Event:      "chat",Data:       "Hello, world!",DataPrefix: true, // 通常建议设置为 true 让 Gin 处理格式
})
ctx.Writer.Flush() // 确保消息立即发送

总结

ctx.Render(-1, sse.Event{…}) 用于在 Gin 中发送 SSE 事件。
• Event 字段允许你定义自定义事件类型。
• Data 字段包含要推送的有效负载。
• DataPrefix: false 要求你手动提供格式正确的 SSE 数据字符串（包含 data: 前缀和结束符）。这增加了灵活性但也带来了出错的风险。更常见的做法是设置 DataPrefix: true，让 Gin 自动处理格式。

SSE 数据格式详解

Server-Sent Events (SSE) 使用一种简单的文本格式来传输数据。下面是 SSE 数据格式的详细说明：

基本格式

SSE 消息由一系列字段组成，每个字段以 field: value 的形式出现，最后以两个换行符 (\n\n) 结束。

field: value\n
field: value\n
\n

核心字段

data - 数据内容（必需）
定义要发送的实际数据内容。
格式：

data: <message>\n

示例：

data: Hello World\n

多行数据：

data: First line\n
data: Second line\n

event - 事件类型（可选）
定义事件类型，客户端可以根据不同类型进行不同处理。
格式：

event: <event-name>\n

示例：

event: message\n
data: Hello World\n

id - 事件ID（可选）
为事件提供唯一标识符，用于断线重连时恢复。
格式：

id: <id>\n

示例：

id: 12345\n
data: Hello World\n

retry - 重连时间（可选）
指定连接断开后客户端重新连接的时间（毫秒）。
格式：

retry: <milliseconds>\n

示例：

retry: 3000\n
data: Hello World\n

完整示例

示例 1：基本消息

data: This is a simple message\n
\n

示例 2：带事件类型的消息

event: userMessage\n
data: {"user": "John", "message": "Hello!"}\n
\n

示例 3：多行数据

event: log\n
data: Starting process...\n
data: Loading configuration\n
data: Process started successfully\n
\n

示例 4：完整功能消息

id: 1689345210\n
event: statusUpdate\n
retry: 5000\n
data: {"status": "processing", "progress": 75}\n
\n

🔄 实际传输示例
一个 SSE 流可能包含多个消息：

data: Message 1\n
\n
event: update\n
data: Message 2\n
\n
id: 1001\n
event: complete\n
data: Process finished\n
retry: 10000\n
\n

💻 Go 代码生成示例
方法 1：手动构建格式

func sendSSEMessageManual(c *gin.Context, event, data string) {message := fmt.Sprintf("event: %s\ndata: %s\n\n", event, data)c.Writer.Write([]byte(message))c.Writer.Flush()
}

// 使用

sendSSEMessageManual(c, "message", "Hello World")

方法 2：使用 Gin 的 SSEvent 方法

func sendSSEMessageGin(c *gin.Context, event string, data interface{}) {c.SSEvent(event, data)c.Writer.Flush()
}

// 使用

sendSSEMessageGin(c, "message", map[string]string{"text": "Hello World","time": time.Now().Format(time.RFC3339),
})

方法 3：完整示例

func sendCompleteSSE(c *gin.Context) {// 设置SSE头c.Header("Content-Type", "text/event-stream")c.Header("Cache-Control", "no-cache")c.Header("Connection", "keep-alive")// 发送多个消息messages := []struct{id    stringevent stringdata  stringretry int}{{"1", "start", "Process started", 0},{"2", "progress", "25% complete", 0},{"3", "progress", "50% complete", 0},{"4", "complete", "Process finished", 5000},}for _, msg := range messages {if msg.id != "" {fmt.Fprintf(c.Writer, "id: %s\n", msg.id)}if msg.event != "" {fmt.Fprintf(c.Writer, "event: %s\n", msg.event)}if msg.retry > 0 {fmt.Fprintf(c.Writer, "retry: %d\n", msg.retry)}fmt.Fprintf(c.Writer, "data: %s\n\n", msg.data)c.Writer.Flush()time.Sleep(1 * time.Second)}
}