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

用Go实现 SSE 实时推送消息(消息通知)——思悟项目技术4

article 2025/2/11 15:02:28

简介

工作原理

例子

使用场景

简介

SSE（Server - Sent Events）是一种允许服务器向客户端实时推送更新的 Web 技术。是一种基于 HTTP 协议的单向通信机制，服务器可以在客户端建立连接后，持续不断地向客户端发送事件流。客户端只需发起一次请求，服务器就能随时向客户端推送新的数据，无需客户端反复请求。

工作原理

客户端发起请求：客户端通过创建一个 EventSource 对象（在浏览器环境中）向服务器发起一个 HTTP 请求，请求的响应头中 Content - Type 为 text/event - stream。
服务器建立连接：服务器接收到请求后，与客户端建立长连接，保持连接处于打开状态。
服务器推送数据：服务器可以在有新数据时，将数据以特定的格式（事件流格式）发送给客户端。事件流由多个事件组成，每个事件包含事件类型、数据等信息。
客户端接收数据：客户端的 EventSource 对象会持续监听服务器发送的事件流，当接收到新事件时，会触发相应的事件处理函数。

例子

后端（Go + Gin）:

POST /send 端点是用于接收前端发送的消息并广播到所有 SSE 客户端，它的作用是触发消息的推送。

而 for msg := range client 这部分代码是具体执行消息推送的逻辑，它确保 SSE 连接的客户端能够实时收到消息。

package main

import (
    "fmt"
    "net/http"
    "sync"
    "time"

    "github.com/gin-gonic/gin"
)

type Client chan string

var (
    clients   = make(map[Client]bool) // 记录所有 SSE 连接
    clientsMu sync.Mutex              // 保护 clients 并发访问
)

func main() {
    r := gin.Default()

    // SSE 端点：用于客户端接收推送
    r.GET("/stream", func(c *gin.Context) {
       client := make(Client, 10) // 创建一个带缓冲的通道
       clientsMu.Lock()
       clients[client] = true // 添加新客户端
       clientsMu.Unlock()

       // 设置 SSE 头部
       c.Header("Content-Type", "text/event-stream")
       c.Header("Cache-Control", "no-cache")
       c.Header("Connection", "keep-alive")

       // 监听消息通道
       // 监听 SSE（Server-Sent Events） 连接的消息通道，并将新消息推送到客户端。它实现了 服务器端的推送机制。
       for msg := range client {
          fmt.Fprintf(c.Writer, "data: %s\n\n", msg)
          c.Writer.Flush()
       }
    })

    // 发送消息端点：接收前端消息，并推送给所有客户端
    r.POST("/send", func(c *gin.Context) {
       var json struct {
          Message string `json:"message"`
       }
       if err := c.ShouldBindJSON(&json); err != nil {
          c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
          return
       }

       // 发送消息到所有 SSE 客户端
       broadcast(json.Message)
       c.JSON(http.StatusOK, gin.H{"status": "sent"})
    })

    r.GET("/", func(c *gin.Context) {
       c.Header("Content-Type", "text/html")
       c.String(http.StatusOK, htmlContent)
    })

    r.Run(":8080")
}

// 发送消息给所有已连接的 SSE 客户端
func broadcast(message string) {
    clientsMu.Lock()
    defer clientsMu.Unlock()
    for client := range clients {
       select {
       case client <- fmt.Sprintf("%s - %s", message, time.Now().Format("15:04:05")):
       default:
          close(client)           // 关闭不可用的客户端
          delete(clients, client) // 从列表中移除
       }
    }
}

const htmlContent = `<!DOCTYPE html>
<html lang="zh">
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>SSE 实时消息</title>
</head>
<body>
    <h2>Server-Sent Events (SSE) 消息推送</h2>
    <input type="text" id="messageInput" placeholder="输入消息">
    <button onclick="sendMessage()">发送消息</button>
    <h3>收到的消息：</h3>
    <div id="messages"></div>

    <script>
        // 监听 SSE 事件
        const eventSource = new EventSource("/stream");
        eventSource.onmessage = function(event) {
            const div = document.createElement("div");
            div.textContent = "收到：" + event.data;
            document.getElementById("messages").appendChild(div);
        };

        // 发送消息到后端
        function sendMessage() {
            const message = document.getElementById("messageInput").value;
            if (!message) return;

            fetch("/send", {
                method: "POST",
                headers: { "Content-Type": "application/json" },
                body: JSON.stringify({ message: message })
            }).then(response => response.json())
              .then(data => console.log("消息已发送:", data));
        }
    </script>
</body>
</html>`

结果：