gin实现event stream

event stream是属于http的一种通信方式，可以实现服务器主动推送。原理于客户端请求服务器之后一直保持链接，服务端持续返回结果给客户端。相比较于websocket有如下区别：

1. 基于http的通信方式，在各类框架的加持下不需要开发人员自己维护链接状态，而websocket一般需要开发自己维护客户端链接(一般是一个map)
2. 也是因为基于http，客户端请求之后便处于接收状态(发送信道关闭？)，所以只能接收服务端推送，而不能客户端推送，比较适合用作通知等场景。

以gin框架为例实现：

func TestEventStream(c *gin.Context) {
	// 声明数据格式为event stream
	c.Writer.Header().Set("Content-Type", "text/event-stream")
	c.Writer.Header().Set("Cache-Control", "no-cache")
	c.Writer.Header().Set("Connection", "keep-alive")
	// 禁用nginx缓存,防止nginx会缓存数据导致数据流是一段一段的
	c.Writer.Header().Set("X-Accel-Buffering", "no")

	w := c.Writer
	flusher, _ := w.(http.Flusher)
	flusher.Flush()
	// 数据chan
	msgChan := make(chan string)
	// 错误chan
	errChan := make(chan error, 1)

	// 开启另一个协程处理业务，通过msgChan和errChan传递信息和错误
	go handle(msgChan, errChan)

	// 读取消息
	for {
		msg, ok := <-msgChan
		if !ok {
			break
		}
		fmt.Fprintf(w, "event: message\n")
		fmt.Fprintf(w, "data: %s\n\n", msg)
		flusher.Flush()
	}

	// 检查错误
	for {
		err, ok := <-errChan
		if !ok {
			return
		}
		fmt.Println(err)
		fmt.Fprintf(w, "event: error\n")
		fmt.Fprintf(w, "data: %s\n\n", err.Error())
		flusher.Flush()
	}
}

//逻辑处理,读取文件中每一行的内容返回给eventstream
func handle(msgChan chan string, errChan chan error) {
	defer func() {
		if r := recover(); r != nil {
			errChan <- errors.New("system panic")
		}
		close(msgChan)
		close(errChan)
	}()
	file, err := os.Open("temp.txt")
	if err != nil {
		errChan <- err
		return
	}
	scanner := bufio.NewScanner(file)
	for scanner.Scan() {
		msgChan <- scanner.Text()
	}
}
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