Golang SSE 服務器端推送事件
更新時間：2025年07月24日 09:56:03 作者：Rocc_
本文主要介紹了Golang SSE 服務器端推送事件,SSE基于HTTP實現(xiàn),允許服務器單向推送數(shù)據(jù),客戶端通過EventSource接收,下面就來詳細的介紹一下,感興趣的可以了解一下
寫在前面（愚蠢的我犯的錯誤）
本應該在EventStream的怎么都在響應這里出現(xiàn)
后面通過查找問題知道EventSream有特殊的回復格式為：data: [返回的內容]\n\n
示例：data: success\n\n
返回success字符串
原因
我做了一個在線點贊的實時更新的小玩意，我想著實時更新WS全雙工用不著。
SSE介紹
SSE（Server-Sent Event）是一種用于客戶端與服務器端實時通訊的技術。它允許服務器端發(fā)送事件到客戶端，客戶端可以通過 EventSource 接口來接收這些事件。通常情況下，SSE 是基于 HTTP 協(xié)議實現(xiàn)的，它不需要建立和維護長連接，但服務器可以長時間向客戶端推送數(shù)據(jù)，而客戶端只需要等待并處理收到的數(shù)據(jù)即可。
Golang實現(xiàn)方式
SSE核心代碼
//sse Server-Sent-Events 服務事件流
http.HandleFunc("/sse", func(w http.ResponseWriter, r *http.Request) {
// 設置響應頭，表明這是一個 SSE 連接
w.Header().Set("Content-Type", "text/event-stream")
w.Header().Set("Cache-Control", "no-cache")
w.Header().Set("Connection", "keep-alive")
//設置為刷新模式
flush, ok := w.(http.Flusher)
flush.Flush()
if !ok {
//判斷是否轉換成功，不成功則返回錯誤信息
responseInfo(http.StatusOK, "response cannot convert http.Flusher", w)
return
}
//這里因為我創(chuàng)建了一個Map用來存儲響應IO和Flush刷新，
//我在其他地方可以使用遍歷進行給各個通信端進行發(fā)送信息
respFlushMap[&w] = &flush
select {
case <-r.Context().Done():
delete(respFlushMap, &w)
return
}
})
發(fā)送事件請求
func main(){
//....
//點贊評論
http.HandleFunc("/favorite", favorite(client))
//...
}
// 點贊
func favorite(client *redis.Client) func(w http.ResponseWriter, r *http.Request) {
var lock = sync.RWMutex{}
return func(w http.ResponseWriter, r *http.Request) {
/* 業(yè)務處理邏輯
......
*/
//核心代碼 將點贊信息發(fā)送到各個SSE
for writer, flusher := range respFlushMap {
//一定要是這個格式“data: [數(shù)據(jù)內容]\n\n”不然前端不會體現(xiàn)在ServeEvent中而出現(xiàn)在response中
fmt.Fprintf(*writer, "data: %s@%s\n\n", member, method)
(*flusher).Flush()
}
}
}
全部代碼，包含了一些處理邏輯，可能比較混亂建議還是看看之前的
package main
import (
"context"
"encoding/json"
"fmt"
"github.com/go-redis/redis/v8"
"html/template"
"log"
"math/rand"
"net"
"net/http"
"strconv"
"sync"
"time"
)
var commentNodeHashRedisKey = "commentNodeHashRedisKey"
var commentNodeSorterSetRedisKey = "commentNodeSorterSetRedisKey"
var respFlushMap = make(map[*http.ResponseWriter]*http.Flusher)
type CommentNode struct {
Content string `json:"content"` //內容
Score float64 `json:"score"` //點贊數(shù)
IP string `json:"ip"` //IP
NickName string `json:"nickName"` //昵稱
IsFavorite bool `json:"isFavorite"` //是否點贊
Member string `json:"member"` //唯一值
}
func main() {
//靜態(tài)資源文件
staticServer := http.FileServer(http.Dir("./template"))
//處理靜態(tài)資源文件
http.Handle("/static/", http.StripPrefix("/static/", staticServer))
//創(chuàng)建客戶端
client := redis.NewClient(&redis.Options{
Addr: "192.168.192.170:6379",
Password: "",
DB: 0,
})
//判斷時候連接成功
err := client.Ping(context.Background()).Err()
if err != nil {
log.Println("連接錯誤: ", err.Error())
}
log.Println("連接成功")
//添加評論
http.HandleFunc("/addComment", addComment(client))
//點贊評論
http.HandleFunc("/favorite", favorite(client))
//sse Server-Sent-Events 服務事件流
http.HandleFunc("/sse", func(w http.ResponseWriter, r *http.Request) {
// 設置響應頭，表明這是一個 SSE 連接
w.Header().Set("Content-Type", "text/event-stream")
w.Header().Set("Cache-Control", "no-cache")
w.Header().Set("Connection", "keep-alive")
//設置為刷新模式
flush, ok := w.(http.Flusher)
flush.Flush()
if !ok {
responseInfo(http.StatusOK, "response cannot convert http.Flusher", w)
return
}
respFlushMap[&w] = &flush
ticker := time.NewTicker(time.Second)
defer ticker.Stop()
select {
case <-r.Context().Done():
delete(respFlushMap, &w)
return
}
})
http.HandleFunc("/commentList", commentList(client))
//主頁
http.HandleFunc("/", func(resp http.ResponseWriter, req *http.Request) {
//獲取模板
indexFile, err := template.ParseFiles("./template/index.html")
if err != nil {
log.Println(err.Error())
resp.Write([]byte("./template/index.html not found"))
return
}
//將內容輸出
indexFile.Execute(resp, nil)
})
//啟動服務
if err := http.ListenAndServe(":80", nil); err != nil {
log.Println("啟動服務失?。? + err.Error())
}
}
// 添加評論
func addComment(client *redis.Client) func(w http.ResponseWriter, r *http.Request) {
return func(w http.ResponseWriter, r *http.Request) {
query := r.URL.Query()
nickname := query.Get("nickName")
if nickname == "" {
nickname = "逸士"
}
//判斷內容是否為空
content := query.Get("content")
if content == "" {
responseInfo(http.StatusBadRequest, "your comment content is empty", w)
return
}
host, _, _ := net.SplitHostPort(r.RemoteAddr)
//使用時間戳
member := fmt.Sprint(time.Now().UnixMilli() ^ rand.Int63())
//序列化
comment, _ := json.Marshal(CommentNode{
Member: member,
IP: host,
NickName: nickname,
Content: content,
})
//添加到隊列中
client.HSet(r.Context(), commentNodeHashRedisKey, member, string(comment))
//更新排行
client.ZAdd(r.Context(), commentNodeSorterSetRedisKey, &redis.Z{
Score: 0,
Member: member,
})
responseInfo(http.StatusOK, "add comment success", w)
}
}
// 點贊
func favorite(client *redis.Client) func(w http.ResponseWriter, r *http.Request) {
var lock = sync.RWMutex{}
return func(w http.ResponseWriter, r *http.Request) {
lock.Lock()
defer lock.Unlock()
//查詢成員(member)是否存在
query := r.URL.Query()
member := query.Get("member")
method := r.Method
if member == "" {
responseInfo(http.StatusBadRequest, "member cannot be empty", w)
lock.Unlock()
return
}
//獲取分數(shù)
score, err := client.ZScore(r.Context(), commentNodeSorterSetRedisKey, member).Result()
//點贊減少
if method == http.MethodDelete {
score -= 2
}
score++
//更新排行
client.ZAdd(r.Context(), commentNodeSorterSetRedisKey, &redis.Z{
Score: score,
Member: member,
})
if err != nil {
//不存在返回錯誤
responseInfo(http.StatusBadRequest, "member does not exists", w)
return
}
//更新分數(shù)
var commentNode CommentNode
commentNodeStr, err := client.HGet(r.Context(), commentNodeHashRedisKey, member).Result()
if err != nil {
responseInfo(http.StatusBadRequest, "update hash scope error: "+err.Error(), w)
return
}
err = json.Unmarshal([]byte(commentNodeStr), &commentNode)
if err != nil {
responseInfo(http.StatusBadRequest, "update hash scope error: "+err.Error(), w)
return
}
commentNode.Score = score
data, _ := json.Marshal(commentNode)
if err = client.HSet(r.Context(), commentNodeHashRedisKey, member, data).Err(); err != nil {
responseInfo(http.StatusBadRequest, "update hash scope error: "+err.Error(), w)
return
}
//返回成功
responseInfo(http.StatusOK, "favorite comment success", w)
//將點贊信息發(fā)送到各個SSE
for writer, flusher := range respFlushMap {
fmt.Fprintf(*writer, "data: %s@%s\n\n", member, method)
(*flusher).Flush()
}
}
}
// 評論列表
func commentList(client *redis.Client) func(resp http.ResponseWriter, req *http.Request) {
return func(resp http.ResponseWriter, req *http.Request) {
query := req.URL.Query()
offset, err := strconv.Atoi(query.Get("offset"))
if err != nil {
offset = 100
}
//連接人地址
connectionAddr := req.RemoteAddr
log.Printf("連接人地址: %s\n", connectionAddr)
//獲取offset偏移量的排行
result, err := client.ZRevRangeWithScores(req.Context(), commentNodeSorterSetRedisKey, 0, int64(offset-1)).Result()
if err != nil || result == nil {
responseInfo(http.StatusOK, fmt.Sprint("錯誤：", err), resp)
return
}
//獲取評論詳細信息
members := make([]string, 0)
scopeMap := make(map[string]float64)
for _, item := range result {
members = append(members, item.Member.(string))
scopeMap[item.Member.(string)] = item.Score
}
rlt, err := client.HMGet(req.Context(), commentNodeHashRedisKey, members...).Result()
if err != nil {
responseInfo(http.StatusInternalServerError, err.Error(), resp)
return
}
data, _ := json.Marshal(rlt)
responseInfo(http.StatusOK, string(data), resp)
}
}
func responseInfo(code int, info string, w http.ResponseWriter) {
w.WriteHeader(code)
w.Write([]byte(info))
}
到此這篇關于Golang SSE 服務器端推送事件的文章就介紹到這了,更多相關Golang SSE推送內容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持腳本之家！
相關文章
這篇文章主要介紹了go語言變量定義用法,實例分析了go語言變量的定義及使用技巧,具有一定參考借鑒價值,需要的朋友可以參考下2015-02-02
詳解Golang利用反射reflect動態(tài)調用方法
這篇文章主要介紹了詳解Golang利用反射reflect動態(tài)調用方法，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧2018-11-11
這篇文章主要介紹了go格式“占位符”, 輸入輸出,類似python的input,本文給大家介紹的非常詳細，具有一定的參考借鑒價值，需要的朋友可以參考下2019-04-04
這篇文章主要介紹了用Go構建不可變的數(shù)據(jù)結構的方法，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧2019-09-09
本文將介紹如何在Linux操作系統(tǒng)下搭建Go語言環(huán)境,Go語言是一種開源的編程語言,具有高效、簡潔和并發(fā)性強的特點,適用于開發(fā)各種類型的應用程序,搭建Go語言環(huán)境是開始學習和開發(fā)Go語言項目的第一步,本文將詳細介紹安裝Go語言、配置環(huán)境變量以及驗證安裝是否成功的步驟2023-10-10
通過使用Go語言的encoding/json包，你可以輕松地處理JSON數(shù)據(jù)，無論是在客戶端應用、服務器端應用還是其他類型的Go程序中，這篇文章主要介紹了golang?JSON序列化和反序列化,需要的朋友可以參考下2024-04-04
重試機制是一種在程序執(zhí)行過程中出現(xiàn)錯誤后重新嘗試執(zhí)行程序的一種機制，可以減少程序運行過程中出現(xiàn)的錯誤，從而提高程序的可靠性，本文就來講講Golang中是如何實現(xiàn)重試機制的吧2023-05-05
Go語言實現(xiàn)一個Http Server框架(二) Server的抽象
上一篇文章對http庫的基本使用做了說明，這篇文章主要介紹了如何實現(xiàn)一個簡單地httpServer，文中代碼示例非常詳細，感興趣的朋友可以參考下2023-04-04
Go語言實現(xiàn)類似c++中的多態(tài)功能實例
Go本身不具有多態(tài)的特性，不能夠像Java、C++那樣編寫多態(tài)類、多態(tài)方法。但是，使用Go可以編寫具有多態(tài)功能的類綁定的方法。下面來一起看看吧2016-09-09
Context本身的含義是上下文,我們可以理解為它內部攜帶了超時信息、退出信號,以及其他一些上下文相關的值,本文給大家詳細介紹了golang中Context超時控制與原理,文中有相關的代碼示例供大家參考,需要的朋友可以參考下2024-01-01