通过LiveGo学习WebSocket

作者：小小林熬夜学编程 | 2024-02-28 03:54:03

踩

livego

一直对网络直播、在线互动的前端应用特比感兴趣，偶然间在Gitee上看到了livego的开源代码，是基于golang的直播服务实现。代码量不大，于是抽丝剥茧的分析了一下其设计思路，觉得还蛮有意思的。

所以，用户至少具备两个基本属性，用户的id和关联数据推送的WebSocket连接。除了这两个基本信息外，可根据实际业务需要增加其他信息，如用户名、头像等。上线用户的临时组管理无用户认证、用户信息管理等复杂业务逻辑，也没有数据信息的持久化处理。

对于消息的定义有如下字段：消息类型-MType、消息内容-Data和源用户的头像信息-Img。消息类型的串行化处理采用了json.Marshall。

1. 直播中的业务处理
直播是一端数据上传，多端数据分发的过程。数据的上传和转发都要借助长连接，所以在livego中采用了WebSocket。livego是一个简易的开放式聊天服务，在线群组管理做的比较简单：
1) 为请求连接的用户创建唯一标识，若连接成功则在群组中保留；否则将其移除；
2) 消息转发过程中，根据数据推送的状态更新群群组状态，如数据推送失败则断开连接，并将用户剔除。

对于消息的定义有如下字段：消息类型-MType、消息内容-Data和源用户的头像信息-Img。消息类型的串行化处理采用了json.Marshall。

2. WebSocket会话

WebSocket会话直播服务的核心基础构件。在livego中自带了一套websocket的封装，协议字段的定义、ws帧解析参考规范文档，draft-ietf-hybi-thewebsocketprotocol-10。这个文档是可以网上找到的，对照文档解读源码，思路会比较清晰易懂。

从WebSocket客户端请求说起。WebSocket请求处理的第一补是解析ws的url，其格式与http的URL格式类似，不同的是把scheme换成了ws或wss，该scheme的处理策略是创建ws连接。处理如下：

1）根据URL及origin够构造DiagConfig，该配置参数会用于后续的TCP连接的建立和参数配置1


	// Dial opens a new client connection to a WebSocket.
	func Dial(url_, protocol, origin string) (ws *Conn, err error) {
	    config, err := NewConfig(url_, origin)
	    if err != nil {
	        return nil, err
	    }
	    if protocol != "" {
	        config.Protocol = []string{protocol}
	    }
	    return DialConfig(config)
	}
		Ø 借助golang的net库，创建TCP连接
	func dialWithDialer(dialer *net.Dialer, config *Config) (conn net.Conn, err error) {
	    switch config.Location.Scheme {
	    case "ws":
	        conn, err = dialer.Dial("tcp", parseAuthority(config.Location))
	    case "wss":
	        conn, err = tls.DialWithDialer(dialer, "tcp", parseAuthority(config.Location), config.TlsConfig)
	    default:
	        err = ErrBadScheme
	    }
	    return
	}

2）借助golang的net库，创建TCP连接


	func dialWithDialer(dialer *net.Dialer, config *Config) (conn net.Conn, err error) {
	    switch config.Location.Scheme {
	    case "ws":
	        conn, err = dialer.Dial("tcp", parseAuthority(config.Location))
	    case "wss":
	        conn, err = tls.DialWithDialer(dialer, "tcp", parseAuthority(config.Location), config.TlsConfig)
	    default:
	        err = ErrBadScheme
	    }
	    return
}

3）TCP连接创建成功后，构造WS Client


	// DialConfig opens a new client connection to a WebSocket with a config.
	func DialConfig(config *Config) (ws *Conn, err error) {
	    var client net.Conn
	    if config.Location == nil {
	        return nil, &DialError{config, ErrBadWebSocketLocation}
	    }
	    if config.Origin == nil {
	        return nil, &DialError{config, ErrBadWebSocketOrigin}
	    }
	    dialer := config.Dialer
	    if dialer == nil {
	        dialer = &net.Dialer{}
	    }
		// client是TCP连接对象
	    client, err = dialWithDialer(dialer, config)
	    if err != nil {
	        goto Error
	    }
		// 基于TCP连接创建ws连接
	    ws, err = NewClient(config, client)
	    if err != nil {
	        client.Close()
	        goto Error
	    }
	    return
	Error:
	    return nil, &DialError{config, err}
}

4） rwc是tcp连接对象，并由此构造缓冲读写的对象br，bw。并开始执行WS会话中的关键步骤-握手。握手完成后随即构造WS连接对象。hybiClientHandshake实现是纯HTTP的协议交互，按照HTTP的协议规范构造协议内容，如头部、请求体等，发送给对端，根据对端返回的状态码及必要字段的合法性判断握手过程是否成功。


	// NewClient creates a new WebSocket client connection over rwc.
	func NewClient(config *Config, rwc io.ReadWriteCloser) (ws *Conn, err error) {
	    br := bufio.NewReader(rwc)
	    bw := bufio.NewWriter(rwc)
	    err = hybiClientHandshake(config, br, bw)
	    if err != nil {
	        return
	    }
	    buf := bufio.NewReadWriter(br, bw)
	    ws = newHybiClientConn(config, buf, rwc)
	    return
	}
		Ø WS连接创建的过程即为Conn对象的构造和初始化过程，各自字段的含义及备注说明如下：
	// newHybiConn creates a new WebSocket connection speaking hybi draft protocol.
	func newHybiConn(config *Config, buf *bufio.ReadWriter, rwc io.ReadWriteCloser, request *http.Request) *Conn {
	    if buf == nil {
	        br := bufio.NewReader(rwc)
	        bw := bufio.NewWriter(rwc)
	        buf = bufio.NewReadWriter(br, bw)
	    }
	    ws := &Conn{config: config, request: request, buf: buf, rwc: rwc,
	        frameReaderFactory: hybiFrameReaderFactory{buf.Reader},
	        frameWriterFactory: hybiFrameWriterFactory{
	            buf.Writer, request == nil},
	        PayloadType:        TextFrame,
	        defaultCloseStatus: closeStatusNormal}
	    ws.frameHandler = &hybiFrameHandler{conn: ws}
	    return ws
	}
	
	
	// Conn represents a WebSocket connection.
	//
	// Multiple goroutines may invoke methods on a Conn simultaneously.
	type Conn struct {
	    config  *Config           // 连接建立的基础参数配置
	    request *http.Request     // 服务端WS处理用，保存对端WS连接的握手请求对象
	    buf *bufio.ReadWriter     // TCP连接的缓存读写
	    rwc io.ReadWriteCloser    // 关闭连接的接口对象
	
	    rio sync.Mutex
	    frameReaderFactory        // 工厂类-构造帧读对象
	    frameReader               // 读对象记录：每次的读都要构造新的读对象，用于处理连接中的数据。该对象记录用于读取上一次读操作后剩余的数据
	
	    wio sync.Mutex
	    frameWriterFactory        // 工厂类-构造写对象
	
	    frameHandler              // 读操作中的中继处理，如连续读中的状态码记录和转换
	    PayloadType        byte   // 与conn对象绑定，写数据时默认的数据类型
	    defaultCloseStatus int
	    // MaxPayloadBytes limits the size of frame payload received over Conn
	    // by Codec's Receive method. If zero, DefaultMaxPayloadBytes is used.
	    MaxPayloadBytes int       // 读数据时最大的有效长度，超长直接报错
	}

5）连接建立之后，就是WS帧的读写操作。WS帧的结构为：

|------------- Frame Header -----------------|------------------- PlayLoad ----------------|

FrameHeader是对帧状态、长度等的基本信息描述，对PayLoad的读取起决定作用。

先看读操作


	func (ws *Conn) Read(msg []byte) (n int, err error) {
	    ws.rio.Lock()
	    defer ws.rio.Unlock()
	again:
	    if ws.frameReader == nil {
		   // 构造读对象。实现上是从连接中读数据，构造ws帧的描述信息，即Frame Header，包含状态位，帧长度等；
	        frame, err := ws.frameReaderFactory.NewFrameReader()
	        if err != nil {
	            return 0, err
	        }
	       // 可能是中间读，所以通过handler校正下PayloadType
	        ws.frameReader, err = ws.frameHandler.HandleFrame(frame)
	        if err != nil {
	            return 0, err
	        }
	        if ws.frameReader == nil {
	            goto again
	        }
	    }
	    // 根据FrameHeader描述，读取连接中的PayLoad数据。如果当前的OperateCode是FinishFrame则表示
	         //   整个帧读取结束，返回io.EOF
	    n, err = ws.frameReader.Read(msg)
	    if err == io.EOF {
	        if trailer := ws.frameReader.TrailerReader(); trailer != nil {
	            io.Copy(ioutil.Discard, trailer)
	        }
	        ws.frameReader = nil
	        goto again //再未遇到CloseFrame之前，要多次、连续读
	    }
	    return n, err
	}

6）写操作与读类似，读是分头解析和数据读取，那么写自然也是分两部分。值得注意是，写操作是一次性将所有数据单帧写入到连接，不做拆分，每一帧都结束帧，不存在中间连续帧的情况。


	// Write implements the io.Writer interface:
	// it writes data as a frame to the WebSocket connection.
	func (ws *Conn) Write(msg []byte) (n int, err error) {
	    ws.wio.Lock()
	    defer ws.wio.Unlock()
		// 构造FrameHeader对象，完成Header中描述字段的初始化赋值
	    w, err := ws.frameWriterFactory.NewFrameWriter(ws.PayloadType)
	    if err != nil {
	        return 0, err
	    }
		// 写Header和PayLoad数据
	    n, err = w.Write(msg)
	    w.Close()
	    return n, err
	}

至此，WS客户端的WS建立及数据读写过程就说完了。其实WS Client的建立和读写操作对服务同样适用，剩下的是服务端如何处理WS的握手请求并建立WS连接的。首先从http的请求分发开始

1）服务启动时会注册如下的http请求处理，表示如果是chat请求则走WS的握手过程


	func main() {
	    //fmt.Printf("LiveGoServer is ready...\n")
	    go func() {
	        http.Handle("/chat", websocket.Handler(pwint)) //这里是将函数对象强转为Handler类型
    }()

2） Handler是实现了http.Handler接口的类型，自然在接口的实现中开始处理对端的WS握手请求


		// Handler是函数类型别名
        type Handler func(*Conn)
		// ServeHTTP implements the http.Handler interface for a WebSocket
		func (h Handler) ServeHTTP(w http.ResponseWriter, req *http.Request) {
		    s := Server{Handler: h, Handshake: checkOrigin}
		    s.serveWebSocket(w, req)
        }

3）serveWebSocket是完成握手过程，创建WS连接


		func (s Server) serveWebSocket(w http.ResponseWriter, req *http.Request) {
		    rwc, buf, err := w.(http.Hijacker).Hijack()  // 接管http的tcp连接对象
		    if err != nil {
		        panic("Hijack failed: " + err.Error())
		    }
		    // The server should abort the WebSocket connection if it finds
		    // the client did not send a handshake that matches with protocol
		    // specification.
		    defer rwc.Close()  // 直播业务结束后，断开tcp连接
			// 创建WS连接
		    conn, err := newServerConn(rwc, buf, req, &s.Config, s.Handshake)
		    if err != nil {
		        return
		    }
		    if conn == nil {
		        panic("unexpected nil conn")
		    }
			// 这里类似接口回调，转发到上方的pwint接口中。pwint为阻塞接口，持续的连接读写。
		    s.Handler(conn)
}

4）最后看一下pwint，可以发现是最上层的业务处理：更新临时组、从当前的连接中读后转发给其他连接，完成直播中的多端分发处理。

好了，没写多少文字，大部分都是在分析代码，相信应该知道websocket是怎么回事了，这就是与new WebSocket而后Send、Receive调用相比有意思的地方的，就是可以看到背后的细节。

本文内容由网友自发贡献，转载请注明出处：【wpsshop博客】

通过LiveGo学习WebSocket

1. 直播中的业务处理

2. WebSocket会话