22 Web编程 WebSocket

[TOC]

Web编程 WebSocket

随着HTML5的普及，WebSocket技术慢慢被采用，它是HTML5提供的一种在单个TCP连接上进行双工通信的协议。WebSocket让客户端和服务器之间的数据交换变得非常简单，且允许服务器主动向客户端推送数据。

WebSocket

协议原理

在WebSocket中，浏览器和服务器只需要完成一次握手，在二者之间就可创建持久性的连接，而后进行双向数据的传输。WebSocket更节省服务器资源和带宽。
WebSocket在浏览器没有同源限制。
WebSocket通信过程中可互相发送JSON、XML、HTML或图片等任意格式的数据。由于是建立在HTTP基础上的协议，客户端发起连接一旦确立WebSocket通信连接，不论是服务器还是客户端，任意一方都可以直接向对方发送数据。
在有WebSocket之前，网站实现消息推送的方式是采用Ajax不断轮询服务器，由浏览器每隔一段时间向服务器发送HTTP请求获取最新的数据，这种模式是低效的，不断向服务器发HTTP请求，而HTTP请求包含的头部等协议的基础数据是较多的，其中真正有效的数据可能只是很小一部分，这也会浪费一部分资源。
浏览器与服务器建立WebSocket连接，先要向服务器发起一个HTTP请求，这个请求会附加一些额外的头部信息，如“Upgrade:WebSocket”，服务器解析这些附加的头信息然后产生应答返回给浏览器。WebSocket建立连接以后，双方可以通过这个连接传递数据，直到浏览器或服务器主动关闭连接为止。
HTTP是半双工协议，在同一时刻数据只能单向流动。客户端向服务器发送请求，客户端向服务器发送请求，服务器响应请求，这两个过程是单向的，且不能由服务器主动发送消息到客户端。服务器不能主动发送数据给浏览就不能实现一些需要双向通信的功能，如即时聊天。

WebSocket通信过程示意图

WebSocket 协议为了兼容现有浏览器，所以在握手阶段使用了HTTP协议。WebSocket是类似于TCP长连接的通信模式，一旦WebSocket连接建立后，后续数据都以帧序列的形式进行传输。在断开WebSocket连接前，不需要客户端和服务器重新发起连接请求，因此在高并发的场景下可以很好地节约网络带宽资源，具有明显的性能优势，且客户端发送和接收消息都是在同一持久连接上发起，实时性更好。
WebSocket协议建立在TCP协议之上，没有同源限制，客户端可以与任意服务器通信。协议的标识符是ws，加密的协议标识符是wss，服务器网址就是URL。
WS（Web Socket）是WebSocket的普通版本，WS一般默认端口是80。
WSS（Web Socket Secure）是WebSocket的加密版本，WSS默认端口是443。

HTTP握手

握手报文
Upgrade: websocket和Connection: Upgrade 表示连接将要被转换为WebSocket连接。
Sec-WebSocket-Key 用于标识这个连接。

GET ws://localhost:8080/ws HTTP/1.1
Host: localhost
Upgrade: websocket
Connection: Upgrade
Origin: http://localhost:8080
Sec-WebSocket-Key: 连接标识
Sec-WebSocket-Version: 13

握手响应报文
响应行 101 表示连接的HTTP协议即将被更改，更改后的协议是Upgrade: websocket。
版本号和子协议规定了双方能理解的数据格式，以及是否支持压缩等。

HTTP/1.1 101 Switching Protocols
Upgrade: websocket
Connection: Upgrade
Sec-WebSocket-Accept: 连接标识

WebSocket 服务器

Go实现WebSocket
需要额外引入依赖 golang.org/x/net/websocket
使用go mod 可以引入【require golang.org/x/net v0.0.0-20220127200216-cd36cc0744dd】

package main

import (
	"fmt"
	"golang.org/x/net/websocket"
	"io"
	"net/http"
	"strings"
)

func response(count uint64,req string,conn *websocket.Conn)  {
	fmt.Println("请求参数：",req)
	conn.Write([]byte(fmt.Sprintf("第%d次响应",count)))
}

func main() {
	http.Handle("/ws",websocket.Handler(func(conn *websocket.Conn) {
		fmt.Printf("收到来自%s的连接\n",conn.RemoteAddr().String())
		const bufLen uint16 = 1024
		var count uint64 = 0
		req := strings.Builder{}
		buf := make([]byte,1024)
		for {
			n,err := conn.Read(buf)
			if io.EOF == err {
				count++
				req.Write(buf[:n])
				response(count,req.String(),conn)
				break
			}
			if nil != err {
				fmt.Println(err)
				break
			}
			req.Write(buf[:n])
			if bufLen > uint16(n) {
				count++
				response(count,req.String(),conn)
				req = strings.Builder{}
			}
		}
	}))
	http.Handle("/wstest.html",http.FileServer(http.Dir("./web")))
	if err := http.ListenAndServe(":8080",nil); nil != err {
		fmt.Println(err)
	}
}

WebSocket 客户端

客户端连接WebSocket服务器【ws://localhost:8080/ws】
http连接使用ws://，https使用wss://
访问【http://localhost:8080/wstest.html】

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <meta http-equiv="X-UA-Compatible" content="IE=edge">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>websocket测试</title>
</head>
<body>
<div style="height: 50px;">
    <input type="text" id="host" value="ws://">
    <input type="button" id="connect" value="连接" onclick="connect()">
</div>
<div style="height: 50px;">
    <input type="text" id="sdata">
    <input type="button" id="send" value="发送" onclick="send()">
</div>
<div style="height: 50px;">
    <input type="button" id="close" value="关闭连接" onclick="closeWS()">
</div>
<div id="datalist"></div>
</body>
<script>
    var WS = 0
    var datalist = document.getElementById("datalist")
    // 连接websocket
    function connect(){
        if (WS) {
            return
        }
        var host = document.getElementById("host")
        console.log(host.value)
        if (!host.value) {
            return
        }
        WS = new WebSocket(host.value);
        listenerWS()
    }
    // 发送数据
    function send(){
        if (!WS) {
            return
        }
        var sdata = document.getElementById("sdata")
        WS.send(sdata.value);
        datalist.innerHTML += "<p>发送数据：" + sdata.value + "</p>"
        sdata.value = ""
    }
    // 关闭连接
    function closeWS(){
        if (!WS) {
            return
        }
        WS.close()
        WS = 0
    }
    // 监听websocket回调
    function listenerWS(){
        WS.onopen = function(evt) {
            datalist.innerHTML += "<p>连接打开</p>"
        };

        WS.onmessage = function(evt) {
            console.log( "收到数据: " + evt.data);
            datalist.innerHTML += "<p>收到数据：" + evt.data + "</p>"
        };

        WS.onclose = function(evt) {
            console.log("连接关闭");
            datalist.innerHTML += "<p>连接关闭</p>"
        };
    }
</script>
</html>

SSE （Server-sent Events）

SSE 是一种更简单的标准，是作为 HTML5 的扩展而开发的。
SSE 支持从服务器向客户端发送异步消息，但客户端无法向服务器发送消息。
对于客户端只需接收从服务器推送的数据，SSE 的半双工通信模型最适合。
SSE 受浏览器同源策略限制，不能跨域。

SSE 与 WebSocket的区别

SSE 只能从服务器推送到客户端，客户端不能发送数据到服务器，是半双工通信，WebSocket是双工通信。
与 WebSocket 相比，SSE 的一个优势是它是基于 HTTP 运行的，不需要其他组件，后端更容易使用 SSE 等协议来实现功能，如股票数据更新这类只需要服务端推送数据的功能SSE在大部分情况下更合适。
SSE 可以实现的所有功能都可以通过 Websockets 实现，因此 Websocket 得到更多关注，并且Websocket的浏览器原生支持要多于SSE（SSE可以通过额外的脚本 SSE polyfill 实现但是这需要更多的开发量）。
SSE 有最大连接数限制，这是浏览器的行为，浏览器限制6个，在Chrome和Firefox 中标记为“无法修复” 。

SSE 协议

协议实现
SSE 协议是客户端发起的 HTTP Get 请求，服务器在接到该请求后，返回 200 OK 状态，同时附带以下 Headers

Content-Type 规定为 text/event-stream
不允许缓存
Connection 为 Keep-Alive 一直保持 TCP 连接

Content-Type: text/event-stream
Cache-Control: no-cache
Connection: keep-alive

SSE通信过程示意图

数据格式
文本流基础格式，以行为单位的，以冒号分割 Field 和 Value，每行结尾为 \n，每行会Trim掉前后空字符，因此 \r\n 也可以。

field: value

事件
事件之间用额外的\n 隔断，每个事件既可以为单行，也可为多行。

两个由单行组成的事件

data:  message\n\n
data:  message2\n

由多行组成的一个事件，更加易读

data: {\n
data: "foo": "bar",\n
data: "baz", 555\n
data: }\n

事件唯一标标识ID
浏览器会一直跟踪最近的事件ID，如果发生了重连，浏览器会把最近接收到的事件ID放入 HTTP Header “Last-Event-ID” 中，作为一种简单的同步机制。

id: msg1\n
data: message\n

命名事件
除了 ID 唯一标示一个事件之外，也可以通过命名的方式，区分一组类型的事件。默认情况下，事件会被命名为 “message”。
三个事件，第一个事件命名为 “foo”，第二个事件没有命名，第三个事件命名为”bar”。可以看出，在一个事件内部，”event” 可以放在前面，也可以放在末尾。

event: foo\n
data: a foo event\n\n
data: an unnamed event\n\n
data: a bar event\n
event: bar\n

重连时间
一般情况下，连接中断的时候，客户端会在 3 秒内进行重连，这个时间也可以由服务器来指定

retry: 10000

Go SSE实现

Go标准库没有封装 SSE 实现，在 https://github.com/antage/eventsource 第三方封装了SSE实现。

eventsource包下consumer.go和eventsource.go封装了SSE实现。
consumer.go

package eventsource

import (
	"compress/gzip"
	"io"
	"net"
	"net/http"
	"strings"
	"time"
)

type consumer struct {
	conn   io.WriteCloser
	es     *eventSource
	in     chan []byte
	staled bool
}

type gzipConn struct {
	net.Conn
	*gzip.Writer
}

func (gc gzipConn) Write(b []byte) (int, error) {
	n, err := gc.Writer.Write(b)
	if err != nil {
		return n, err
	}

	return n, gc.Writer.Flush()
}

func (gc gzipConn) Close() error {
	err := gc.Writer.Close()
	if err != nil {
		return err
	}

	return gc.Conn.Close()
}

func newConsumer(resp http.ResponseWriter, req *http.Request, es *eventSource) (*consumer, error) {
	conn, _, err := resp.(http.Hijacker).Hijack()
	if err != nil {
		return nil, err
	}

	consumer := &consumer{
		conn:   conn,
		es:     es,
		in:     make(chan []byte, 10),
		staled: false,
	}

	_, err = conn.Write([]byte("HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\n"))
	if err != nil {
		conn.Close()
		return nil, err
	}

	_, err = conn.Write([]byte("Vary: Accept-Encoding\r\n"))
	if err != nil {
		conn.Close()
		return nil, err
	}

	if es.gzip && (req == nil || strings.Contains(req.Header.Get("Accept-Encoding"), "gzip")) {
		_, err = conn.Write([]byte("Content-Encoding: gzip\r\n"))
		if err != nil {
			conn.Close()
			return nil, err
		}

		consumer.conn = gzipConn{conn, gzip.NewWriter(conn)}
	}

	if es.customHeadersFunc != nil {
		for _, header := range es.customHeadersFunc(req) {
			_, err = conn.Write(header)
			if err != nil {
				conn.Close()
				return nil, err
			}
			_, err = conn.Write([]byte("\r\n"))
			if err != nil {
				conn.Close()
				return nil, err
			}
		}
	}

	_, err = conn.Write([]byte("\r\n"))
	if err != nil {
		conn.Close()
		return nil, err
	}

	go func() {
		idleTimer := time.NewTimer(es.idleTimeout)
		defer idleTimer.Stop()
		for {
			select {
			case message, open := <-consumer.in:
				if !open {
					consumer.conn.Close()
					return
				}
				conn.SetWriteDeadline(time.Now().Add(consumer.es.timeout))
				_, err := consumer.conn.Write(message)
				if err != nil {
					netErr, ok := err.(net.Error)
					if !ok || !netErr.Timeout() || consumer.es.closeOnTimeout {
						consumer.staled = true
						consumer.conn.Close()
						consumer.es.staled <- consumer
						return
					}
				}
				idleTimer.Reset(es.idleTimeout)
			case <-idleTimer.C:
				consumer.conn.Close()
				consumer.es.staled <- consumer
				return
			}
		}
	}()

	return consumer, nil
}

eventsource.go

package eventsource

import (
	"bytes"
	"container/list"
	"fmt"
	"log"
	"net/http"
	"strings"
	"sync"
	"time"
)

type eventMessage struct {
	id    string
	event string
	data  string
}

type retryMessage struct {
	retry time.Duration
}

type eventSource struct {
	customHeadersFunc func(*http.Request) [][]byte

	sink           chan message
	staled         chan *consumer
	add            chan *consumer
	close          chan bool
	idleTimeout    time.Duration
	retry          time.Duration
	timeout        time.Duration
	closeOnTimeout bool
	gzip           bool

	consumersLock sync.RWMutex
	consumers     *list.List
}

type Settings struct {
	// SetTimeout sets the write timeout for individual messages. The
	// default is 2 seconds.
	Timeout time.Duration

	// CloseOnTimeout sets whether a write timeout should close the
	// connection or just drop the message.
	//
	// If the connection gets closed on a timeout, it's the client's
	// responsibility to re-establish a connection. If the connection
	// doesn't get closed, messages might get sent to a potentially dead
	// client.
	//
	// The default is true.
	CloseOnTimeout bool

	// Sets the timeout for an idle connection. The default is 30 minutes.
	IdleTimeout time.Duration

	// Gzip sets whether to use gzip Content-Encoding for clients which
	// support it.
	//
	// The default is false.
	Gzip bool
}

func DefaultSettings() *Settings {
	return &Settings{
		Timeout:        2 * time.Second,
		CloseOnTimeout: true,
		IdleTimeout:    30 * time.Minute,
		Gzip:           false,
	}
}

// EventSource interface provides methods for sending messages and closing all connections.
type EventSource interface {
	// it should implement ServerHTTP method
	http.Handler

	// send message to all consumers
	SendEventMessage(data, event, id string)

	// send retry message to all consumers
	SendRetryMessage(duration time.Duration)

	// consumers count
	ConsumersCount() int

	// close and clear all consumers
	Close()
}

type message interface {
	// The message to be sent to clients
	prepareMessage() []byte
}

func (m *eventMessage) prepareMessage() []byte {
	var data bytes.Buffer
	if len(m.id) > 0 {
		data.WriteString(fmt.Sprintf("id: %s\n", strings.Replace(m.id, "\n", "", -1)))
	}
	if len(m.event) > 0 {
		data.WriteString(fmt.Sprintf("event: %s\n", strings.Replace(m.event, "\n", "", -1)))
	}
	if len(m.data) > 0 {
		lines := strings.Split(m.data, "\n")
		for _, line := range lines {
			data.WriteString(fmt.Sprintf("data: %s\n", line))
		}
	}
	data.WriteString("\n")
	return data.Bytes()
}

func controlProcess(es *eventSource) {
	for {
		select {
		case em := <-es.sink:
			message := em.prepareMessage()
			func() {
				es.consumersLock.RLock()
				defer es.consumersLock.RUnlock()

				for e := es.consumers.Front(); e != nil; e = e.Next() {
					c := e.Value.(*consumer)

					// Only send this message if the consumer isn't staled
					if !c.staled {
						select {
						case c.in <- message:
						default:
						}
					}
				}
			}()
		case <-es.close:
			close(es.sink)
			close(es.add)
			close(es.staled)
			close(es.close)

			func() {
				es.consumersLock.RLock()
				defer es.consumersLock.RUnlock()

				for e := es.consumers.Front(); e != nil; e = e.Next() {
					c := e.Value.(*consumer)
					close(c.in)
				}
			}()

			es.consumersLock.Lock()
			defer es.consumersLock.Unlock()

			es.consumers.Init()
			return
		case c := <-es.add:
			func() {
				es.consumersLock.Lock()
				defer es.consumersLock.Unlock()

				es.consumers.PushBack(c)
			}()
		case c := <-es.staled:
			toRemoveEls := make([]*list.Element, 0, 1)
			func() {
				es.consumersLock.RLock()
				defer es.consumersLock.RUnlock()

				for e := es.consumers.Front(); e != nil; e = e.Next() {
					if e.Value.(*consumer) == c {
						toRemoveEls = append(toRemoveEls, e)
					}
				}
			}()
			func() {
				es.consumersLock.Lock()
				defer es.consumersLock.Unlock()

				for _, e := range toRemoveEls {
					es.consumers.Remove(e)
				}
			}()
			close(c.in)
		}
	}
}

// New creates new EventSource instance.
func New(settings *Settings, customHeadersFunc func(*http.Request) [][]byte) EventSource {
	if settings == nil {
		settings = DefaultSettings()
	}

	es := new(eventSource)
	es.customHeadersFunc = customHeadersFunc
	es.sink = make(chan message, 1)
	es.close = make(chan bool)
	es.staled = make(chan *consumer, 1)
	es.add = make(chan *consumer)
	es.consumers = list.New()
	es.timeout = settings.Timeout
	es.idleTimeout = settings.IdleTimeout
	es.closeOnTimeout = settings.CloseOnTimeout
	es.gzip = settings.Gzip
	go controlProcess(es)
	return es
}

func (es *eventSource) Close() {
	es.close <- true
}

// ServeHTTP implements http.Handler interface.
func (es *eventSource) ServeHTTP(resp http.ResponseWriter, req *http.Request) {
	cons, err := newConsumer(resp, req, es)
	if err != nil {
		log.Print("Can't create connection to a consumer: ", err)
		return
	}
	es.add <- cons
}

func (es *eventSource) sendMessage(m message) {
	es.sink <- m
}

func (es *eventSource) SendEventMessage(data, event, id string) {
	em := &eventMessage{id, event, data}
	es.sendMessage(em)
}

func (m *retryMessage) prepareMessage() []byte {
	return []byte(fmt.Sprintf("retry: %d\n\n", m.retry/time.Millisecond))
}

func (es *eventSource) SendRetryMessage(t time.Duration) {
	es.sendMessage(&retryMessage{t})
}

func (es *eventSource) ConsumersCount() int {
	es.consumersLock.RLock()
	defer es.consumersLock.RUnlock()

	return es.consumers.Len()
}

Go SSE 服务端

package main

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

func main() {
	es := eventsource.New(nil, nil)
	defer es.Close()
	http.Handle("/", http.FileServer(http.Dir("./public")))
	http.Handle("/events", es)
	// http://localhost:8080/sse.html
	http.Handle("/sse.html",http.FileServer(http.Dir("./web/")))
	count := 0
	go func() {
		for {
			count ++
			es.SendEventMessage("更新的数据", "", fmt.Sprintf("%d",count))
			log.Printf("接收数据的客户端数量：%d", es.ConsumersCount())
			es.SendEventMessage("更新指定事件的数据", "myevent", fmt.Sprintf("%d",count))
			time.Sleep(3 * time.Second)
		}
	}()
	//http://localhost:8080/
	err := http.ListenAndServe(":8080", nil)
	if err != nil {
		log.Fatal(err)
	}
}

sse.html

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>ees测试</title>
</head>
<body>
    <h1>SSE test</h1>
    <div>
        <ul id="log">
        </ul>
    </div>
</body>
<script type="text/javascript">
    window.addEventListener("DOMContentLoaded", function () {
        var evsrc = new EventSource("/events");
        evsrc.onmessage = function (ev) {
            console.log(ev)
            var log = document.getElementById("log")
            log.innerHTML += "<p>" + ev.data + "("+ ev.lastEventId +")"+ "</p>"
        }
        evsrc.onerror = function (ev) {
            console.log("readyState = " + ev.currentTarget.readyState);
        }
        evsrc.addEventListener("myevent", function(ev) {
            console.log(ev)
            var log = document.getElementById("log")
            log.innerHTML += "<p>事件myevent：" + ev.data + "("+ ev.lastEventId +")"+ "</p>"
        }, false);
    })
</script>
</html>

Previous21 Web编程模板引擎 Next23 内存分配

Last updated 3 years ago