当前位置：首页 > 文章列表 > Golang > Go教程 > Golang实现SSE服务器实时事件流

Golang实现SSE服务器实时事件流

2025-07-29 13:14:51 0浏览收藏

推广推荐

支持 PC / 移动端，安全直达

大家好，今天本人给大家带来文章《Golang构建SSE服务器实现实时事件流》，文中内容主要涉及到，如果你对Golang方面的知识点感兴趣，那就请各位朋友继续看下去吧~希望能真正帮到你们，谢谢！

构建SSE服务器推送的核心在于利用HTTP协议的流式传输特性，使服务器能单向实时推送数据给客户端。1.设置响应头Content-Type为text/event-stream以启用SSE；2.使用http.Flusher强制刷新缓冲区确保数据实时推送；3.通过r.Context().Done()检测客户端断连并释放资源；4.采用带缓冲的channel作为连接池限制并发连接数；5.序列化结构体为JSON实现复杂数据推送；6.启用HTTPS和JWT保障通信安全性。上述要点与文章顺序一致，完整实现了SSE服务器的关键功能及优化措施。

如何用Golang构建SSE服务器推送实现实时事件流传输机制

构建SSE服务器推送的核心在于利用HTTP协议的流式传输特性，让服务器可以单向、实时地向客户端推送数据更新。Golang凭借其高效的并发处理能力，非常适合构建这类服务器。

package main

import (
    "fmt"
    "log"
    "net/http"
    "time"
)

func sseHandler(w http.ResponseWriter, r *http.Request) {
    // 设置响应头，关键是Content-Type
    w.Header().Set("Content-Type", "text/event-stream")
    w.Header().Set("Cache-Control", "no-cache")
    w.Header().Set("Connection", "keep-alive")

    // 创建一个channel，用于发送数据
    messageChan := make(chan string)
    // 启动一个goroutine，模拟数据更新
    go func() {
        ticker := time.NewTicker(5 * time.Second) // 每5秒发送一次数据
        defer ticker.Stop()
        for range ticker.C {
            messageChan <- fmt.Sprintf("data: Server time is %s\n\n", time.Now().Format(time.RFC3339))
        }
    }()

    // 循环读取channel，并将数据写入response writer
    flusher, ok := w.(http.Flusher)
    if !ok {
        http.Error(w, "Streaming unsupported!", http.StatusInternalServerError)
        return
    }

    for msg := range messageChan {
        fmt.Fprintf(w, msg)
        flusher.Flush() // 强制刷新缓冲区，确保数据实时推送
    }
}

func main() {
    http.HandleFunc("/events", sseHandler)
    fmt.Println("SSE server started at :8080")
    log.Fatal(http.ListenAndServe(":8080", nil))
}

这段代码创建了一个简单的SSE服务器，它每5秒向客户端发送一次服务器时间。关键点在于Content-Type设置为text/event-stream，以及使用http.Flusher确保数据实时推送。当然，实际应用中，数据来源可能更加复杂，比如来自数据库更新、消息队列等等。

如何处理客户端断连？

客户端断连是SSE服务器需要考虑的重要问题。如果客户端断开连接，服务器继续推送数据会造成资源浪费，甚至导致程序崩溃。Golang可以通过r.Context().Done()来检测客户端是否断开连接。

func sseHandler(w http.ResponseWriter, r *http.Request) {
    // ... (之前的代码)

    for {
        select {
        case msg := <-messageChan:
            fmt.Fprintf(w, msg)
            flusher.Flush()
        case <-r.Context().Done():
            log.Println("Client disconnected")
            close(messageChan) // 关闭channel，结束goroutine
            return
        }
    }
}

修改后的代码使用select语句同时监听messageChan和r.Context().Done()。当r.Context().Done()被触发时，表示客户端已断开连接，此时关闭messageChan并退出函数，同时结束发送数据的goroutine，释放资源。

如何处理大量并发连接？

SSE服务器需要处理大量并发连接，这涉及到Golang的并发模型。虽然Golang的goroutine非常轻量级，但如果连接数过多，仍然可能导致性能问题。一种常见的优化方法是使用连接池，限制同时处理的连接数。

// 创建一个带缓冲的channel，作为连接池
var connectionPool = make(chan struct{}, 100) // 限制最大并发连接数为100

func sseHandler(w http.ResponseWriter, r *http.Request) {
    // 从连接池获取一个连接
    connectionPool <- struct{}{}
    defer func() {
        // 释放连接到连接池
        <-connectionPool
    }()

    // ... (之前的代码)
}

这段代码使用一个带缓冲的channel connectionPool作为连接池。每个新的连接都需要先从连接池获取一个空槽位，处理完成后再释放回连接池。这样可以限制同时处理的连接数，防止服务器过载。需要注意的是，连接池的大小需要根据服务器的硬件资源和实际负载情况进行调整。

如何实现更复杂的数据推送？

除了简单的文本数据，SSE还可以推送更复杂的数据结构，比如JSON。只需要将数据序列化成JSON字符串，然后作为data字段发送即可。

import "encoding/json"

type EventData struct {
    Timestamp string `json:"timestamp"`
    Message   string `json:"message"`
}

func sseHandler(w http.ResponseWriter, r *http.Request) {
    // ... (之前的代码)

    go func() {
        ticker := time.NewTicker(5 * time.Second)
        defer ticker.Stop()
        for range ticker.C {
            data := EventData{
                Timestamp: time.Now().Format(time.RFC3339),
                Message:   "Hello from server!",
            }
            jsonData, _ := json.Marshal(data)
            messageChan <- fmt.Sprintf("data: %s\n\n", jsonData)
        }
    }()

    // ... (之前的代码)
}

这段代码定义了一个EventData结构体，包含时间戳和消息内容。然后使用json.Marshal将结构体序列化成JSON字符串，并作为data字段发送。客户端需要解析JSON字符串才能获取数据。另外，event字段可以自定义事件类型，客户端可以根据事件类型进行不同的处理。例如：event: my_event\ndata: ...\n\n。