gin如何具体利用Server-Send-Events(SSE)实时推送技术实现消息推送

gin-left:0px;">业务场景

gin-left:0px;">解决方案

gin-left:40px;">1. 轮询

gin-left:40px;">2. WebSocket

gin-left:40px;">3. SSE(Server-Send-Events)

gin-left:0px;">代码实现

gin-left:0px;">总结

业务场景

在抖音、美团等APP中，我们经常会遇到APP内部的消息推送，如关注的人的动态消息推送、点赞评论互动消息推送以及算法推荐消息推送。这些场景都是服务端主动向客户端进行推送实时消息，可以很大程度上提高用户体验。

解决方案

1. 轮询

短轮询：客户端会定时向服务端发送请求，询问在上次询问到现在时刻是否有新的数据，如有则返回给客户端。

长轮询：是对短轮询的优化。每次询问，若有新数据则立刻返回给客户端；若没有新数据，则服务端会等待一段时间，若有新数据则返回给客户端，否则返回空数据。

总之，轮询实现起来最简单，但是无法保证实时性，延迟大且对服务端资源消耗高。

2. WebSocket

websocket是一种双向通信协议，同时支持服务端和客户端之间的实时交互。WebSocket 是基于 TCP 的长连接，和HTTP 协议相比，它能实现轻量级的、低延迟的数据传输，非常适合实时通信场景，主要用于交互性强的双向通信。

3. SSE(Server-Send-Events)

SSE（Server-Sent Events）是一种基于 HTTP 协议的推送技术。服务端可以使用 SSE 来向客户端推送数据，但客户端不能通过SSE向服务端发送数据。相较于 WebSocket，SSE 更简单、更轻量级，但只能实现单向通信。

针对这三种解决方案来看，SSE最贴合当前的业务场景。在消息推送中，只需要服务端对客户端进行单向通信，使用SSE在保证实时性的同时，比WebSocket更加轻量级。

代码实现

1. 我们通过两个接口来实现sse的一个简单demo：分别用来实现SSE的连接和消息推送的触发。

r.GET("/notification/socket-connection", SocketConnection) // 建立sse连接
r.GET("/notification/export-excel", ExportExcel)           // 触发通知，发送消息

2. 建立sse连接:每一个客户端和服务端都需要有一个单独的通道去维持双方的连接，这样才能保证既可以推送广播类的消息，也可以推送定制化的消息。

所以，我们可以通过一个map来存储所有客户端和服务端之间的通道连接。

var channelsMap sync.Map

需要注意的是：这里会对map进行并发操作，所以这里可以使用sync.Map来保证在访问map时的并发安全。

为了保证连接的不丢失，当客户端建立SSE连接，我们就需要创建一个独属于该客户端(如userId作为唯一标识)的通道，然后通过遍历当前通道去保证主线程的阻塞，若通道中有信息则将信息推送到客户端，并继续等待消息。

同时开一个协程去监听客户端连接是否关闭，若关闭则关闭相应通道，结束请求。

func AddChannel(userEmail string, traceId string) {if !ifChannelsMapInit {channelsMap = sync.Map{}ifChannelsMapInit = true}newChannel := make(chan string)channelsMap.Store(userEmail+traceId, newChannel)log.Print("Build SSE connection for user = " + userEmail + ", trace id = " + traceId)
}

func BuildNotificationChannel(userEmail string, traceId string, c *gin.Context) {AddChannel(userEmail, traceId)c.Writer.Header().Set("Content-Type", "text/event-stream")c.Writer.Header().Set("Cache-Control", "no-cache")c.Writer.Header().Set("Connection", "keep-alive")// 获取http写入器并断言为flusher，让其将缓冲器的数据立即写入w := c.Writerflusher, _ := w.(http.Flusher)// 监听客户端通道是否被关闭closeNotify := c.Request.Context().Done()go func() {<-closeNotifychannelsMap.Delete(userEmail + traceId)log.Print("SSE close for user = " + userEmail + ", trace id = " + traceId)return}()curChan, _ := channelsMap.Load(userEmail + traceId)for msg := range curChan.(chan string) {fmt.Fprintf(w, "data:%s\n\n", msg)flusher.Flush()}
}

3. 触发消息推送

当发布系统消息、或者某个用户对该用户的文章进行了点赞，则需要遍历map找到对应的通道，向通道中推送消息。在sse请求中，正在遍历当前的channel，接收到通道的消息后，实时推送到客户端。

func SendNotification(userEmail string, messageBody string, actionType string) {log.Print("Send notification to user = " + userEmail)var msg = NotificationLog{MessageBody: messageBody,UserEmail:   userEmail,Type:        actionType,Status:      "UNREAD",CreatTime:   time.Now(),}msgBytes, _ := json.Marshal(msg)channelsMap.Range(func(key, value any) bool {k := key.(string)if strings.Contains(k, userEmail) {channel := value.(chan string)channel <- string(msgBytes)}return true})
}

4. 跨域

sse是不支持跨域的，所以我们可以加一个跨域中间件。

func CORSMiddleware() gin.HandlerFunc {return func(c *gin.Context) {method := c.Request.Method               //请求方法origin := c.Request.Header.Get("Origin") //请求头部var headerKeys []string                  // 声明请求头keysfor k, _ := range c.Request.Header {headerKeys = append(headerKeys, k)}headerStr := strings.Join(headerKeys, ", ")if headerStr != "" {headerStr = fmt.Sprintf("access-control-allow-origin, access-control-allow-headers, %s", headerStr)} else {headerStr = "access-control-allow-origin, access-control-allow-headers"}if origin != "" {c.Writer.Header().Set("Access-Control-Allow-Origin", "*")c.Header("Access-Control-Allow-Origin", "*")                                       // 这是允许访问所有域c.Header("Access-Control-Allow-Methods", "POST, GET, OPTIONS, PUT, DELETE,UPDATE") //服务器支持的所有跨域请求的方法,为了避免浏览次请求的多次'预检'请求//  header的类型c.Header("Access-Control-Allow-Headers", "Authorization, Content-Length, X-CSRF-Token, Token,session,X_Requested_With,Accept, Origin, Host, Connection, Accept-Encoding, Accept-Language,DNT, X-CustomHeader, Keep-Alive, User-Agent, X-Requested-With, If-Modified-Since, Cache-Control, Content-Type, Pragma")//              允许跨域设置                                                                                                      可以返回其他子段c.Header("Access-Control-Expose-Headers", "Content-Length, Access-Control-Allow-Origin, Access-Control-Allow-Headers,Cache-Control,Content-Language,Content-Type,Expires,Last-Modified,Pragma,FooBar") // 跨域关键设置 让浏览器可以解析c.Header("Access-Control-Max-Age", "172800")                                                                                                                                                           // 缓存请求信息 单位为秒c.Header("Access-Control-Allow-Credentials", "false")                                                                                                                                                  //  跨域请求是否需要带cookie信息 默认设置为truec.Set("content-type", "application/json")                                                                                                                                                              // 设置返回格式是json}//放行所有OPTIONS方法if method == "OPTIONS" {c.JSON(http.StatusOK, "Options Request!")}// 处理请求c.Next() //  处理请求}
}