go实现SSE

实现SSE

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
func main() {
    r := gin.Default()
    r.GET("/sse", 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", "*")

        // 发送初始消息
        fmt.Fprintf(c.Writer, "data: 连接已建立\n\n")
        c.Writer.Flush()

        // 模拟每隔2秒发送一次事件
        for i := 0; i < 5; i++ {
            time.Sleep(2 * time.Second)
            message := fmt.Sprintf("data: 事件 %d\n\n", i+1)
            fmt.Fprintf(c.Writer, message)
            c.Writer.Flush()
        }

        // 结束连接
        fmt.Fprintf(c.Writer, "data: 连接结束\n\n")
        c.Writer.Flush()
    })
    r.Run()
}

封装工具类

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
package main

import (
	"net/http"
)

// SSE SSE结构体
type SSE struct {
	// 用于向客户端写入数据
	writer  http.ResponseWriter
	// 用于刷新写入缓冲区
	flusher http.Flusher
	// 用于接收待发送的数据
	sendCh  chan []byte
	// 用于关闭sse
	closeCh chan bool
}

// NewSSE 创建sse实例
func NewSSE(w http.ResponseWriter) *SSE {
	setHttpHeader(w)
	return &SSE{
		writer:  w,
		flusher: w.(http.Flusher),
		sendCh:  make(chan []byte),
		closeCh: make(chan bool),
	}
}

// 设置HTTP响应头，以适配SSE
func setHttpHeader(w http.ResponseWriter) {
	w.Header().Set("Content-Type", "text/event-stream")
	w.Header().Set("Cache-Control", "no-cache")
	w.Header().Set("Connection", "keep-alive")
	w.Header().Set("Access-Control-Allow-Origin", "*")
}

// Run 运行sse
func (s *SSE) Run() {
	for {
		select {
		case msg := <-s.sendCh:
			s.send(msg)
		case <-s.closeCh:
			return
		}
	}
}

// 发送数据
func (s *SSE) send(msg []byte) {
	if _, err := s.writer.Write(msg); err != nil {
		return
	}
	s.flusher.Flush()
}

// Send 向sendCh中写入待发送数据
func (s *SSE) Send(data []byte) {
	s.sendCh <- data
}

// Close 关闭sse
func (s *SSE) Close() {
	s.closeCh <- true
}

使用

服务端

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
func main() {
	r := gin.Default()
	r.GET("/sse", func(context *gin.Context) {
		fmt.Println("this is start....................")
		sse := NewSSE(context.Writer)
		go func() {
			for i := 0; i < 5; i++ {
				time.Sleep(2 * time.Second)
				data, _ := json.Marshal(map[string]string{"timestamp": time.Now().Format(time.RFC3339)})
				sse.Send([]byte(fmt.Sprintf("data: %s\n\n", data)))
			}
			time.Sleep(2 * time.Second)
			sse.Close()
		}()
		sse.Run()
		fmt.Println("this is end....................")
	})
	r.Run()
}

客户端

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>SSE Example</title>
</head>
<body>

<script>
    const eventSource = new EventSource('http://localhost:8080/sse')

    eventSource.onmessage = function (event) {
        console.log('Message received: ', event.data)
    }

    eventSource.onerror = function (event) {
        console.error('An error occurred:', event)
        // 关闭sse连接
        eventSource.close()
    }
</script>
</body>
</html>