【Go進(jìn)階】手寫 Go websocket 庫（一）｜WebSocket 通信協(xié)議

這里是白澤，我將利用一個(gè)系列，為你分享如何基于 websocket 協(xié)議的 rfc 文檔，編寫一個(gè)庫的過程。并從0開始寫一遍 gorilla/websocket 這個(gè)庫，從中你可以學(xué)習(xí)到 websocket 庫中高質(zhì)量、高性能的寫法（多協(xié)程、緩沖池使用）。

前言

這里是白澤，我將利用一個(gè)系列，為你分享如何基于 websocket 協(xié)議的 rfc 文檔，編寫一個(gè)庫的過程。并從0開始寫一遍 gorilla/websocket 這個(gè)庫，從中你可以學(xué)習(xí)到 websocket 庫中高質(zhì)量、高性能的寫法（多協(xié)程、緩沖池使用）。

倉庫地址：https://github.com/gorilla/websocket，??數(shù)量：22.8k

項(xiàng)目體量不大、核心代碼5k，雖然難度較高，但在引導(dǎo)下也可以完成。

B站：白澤talk

公眾號：白澤talk

開源 Golang 學(xué)習(xí)倉庫：https://github.com/BaiZe1998/go-learning

開源短視頻應(yīng)用 DouTok：https://github.com/cloudzenith/DouTok

WebSocket 協(xié)議

這是第一課，本文將詳細(xì)介紹 WebSocket 協(xié)議的幀結(jié)構(gòu)、握手過程、數(shù)據(jù)傳輸機(jī)制及其在實(shí)際應(yīng)用中的優(yōu)勢。并以 Golang 的 websocket 開源庫作為案例進(jìn)行前瞻，探究協(xié)議是如何用 Golang 代碼實(shí)現(xiàn)的，以及如何使用這個(gè)庫，進(jìn)行 websocket 通信。

一、WebSocket 協(xié)議概述

WebSocket 是一種在單個(gè) TCP 連接上進(jìn)行全雙工通信的協(xié)議。與 HTTP 相比，WebSocket 無需像 HTTP 那樣每次通信都需要建立新的連接，從而大大減少了延遲和資源消耗。它使得客戶端和服務(wù)器之間的數(shù)據(jù)交換變得更加高效和實(shí)時(shí)。

二、WebSocket 幀結(jié)構(gòu)

作為 Go 語言 ?? 數(shù)最多的 websocket 庫，本質(zhì)上是通過 Golang，實(shí)現(xiàn)了 websocket rfc 文檔定義的消息格式、以及交互過程。

rfc：https://datatracker.ietf.org/doc/html/rfc6455#section-5.2

這個(gè)庫的核心實(shí)現(xiàn)包含在幾個(gè)文件中（client.go/serve.go/conn.go）。其中用于定義 websocket 鏈接的 conn.go 文件中，定義了一些常量，用于表示各個(gè)標(biāo)志位，乍一眼你可能看不明白。

const (
	// Frame header byte 0 bits from Section 5.2 of RFC 6455
	finalBit = 1 << 7
	rsv1Bit  = 1 << 6
	rsv2Bit  = 1 << 5
	rsv3Bit  = 1 << 4

	// Frame header byte 1 bits from Section 5.2 of RFC 6455
	maskBit = 1 << 7

	maxFrameHeaderSize         = 2 + 8 + 4 // Fixed header + length + mask
	maxControlFramePayloadSize = 125

	writeWait = time.Second

	defaultReadBufferSize  = 4096
	defaultWriteBufferSize = 4096

	continuationFrame = 0
	noFrame           = -1
)

WebSocket 協(xié)議通過幀（Frame）來傳輸數(shù)據(jù)。每個(gè)幀都包含一系列固定的字段，用于標(biāo)識幀的類型、長度、掩碼以及實(shí)際的數(shù)據(jù)內(nèi)容。

1. FIN

FIN 字段是一個(gè) 1 位的標(biāo)志位，用于指示當(dāng)前幀是否為消息中的最后一個(gè)片段。如果消息僅由一個(gè)片段組成，該位也應(yīng)被設(shè)置為 1。

2. RSV1, RSV2, RSV3

RSV1、RSV2 和 RSV3 是三個(gè) 1 位的保留位，它們必須為 0，除非在 WebSocket 握手階段已經(jīng)協(xié)商了具有特定含義的擴(kuò)展。如果使用了擴(kuò)展，并且這些位被賦予了特定的意義，那么它們將用于指示該幀是否遵循了這些擴(kuò)展的特定規(guī)則。

3. Opcode

Opcode 字段是一個(gè) 4 位的操作碼，用于定義有效載荷數(shù)據(jù)的含義。不同的操作碼代表不同類型的幀：

%x0：表示連續(xù)幀（Continuation Frame），用于將消息分割成多個(gè)片段。
%x1：表示文本幀（Text Frame），包含 UTF-8 編碼的文本數(shù)據(jù)。
%x2：表示二進(jìn)制幀（Binary Frame），包含二進(jìn)制數(shù)據(jù)。
%x8：表示連接關(guān)閉幀（Connection Close Frame），用于關(guān)閉連接。
%x9：表示 Ping 幀，用于連接檢測。
%xA：表示 Pong 幀，作為對 Ping 幀的響應(yīng)。

4. Mask

Mask 字段是一個(gè) 1 位的標(biāo)志位，用于指示是否對有效載荷數(shù)據(jù)進(jìn)行了掩碼處理。在客戶端發(fā)送到服務(wù)器的幀中，該位必須設(shè)置為 1，并附帶一個(gè)掩碼鍵（Masking-key）。服務(wù)器發(fā)送到客戶端的幀則不應(yīng)被掩碼，因此該位應(yīng)為 0。

5. Payload length

Payload length 字段用于表示有效載荷數(shù)據(jù)的總長度。它可以是 7 位、7+16 位或 7+64 位，具體取決于數(shù)據(jù)的長度：

如果長度小于或等于 125 字節(jié)，則使用 7 位表示。
如果長度在 126 到 65,535 字節(jié)之間，則前 7 位設(shè)置為 126，并使用隨后的 16 位來表示長度。
如果長度超過 65,535 字節(jié)，則前 7 位設(shè)置為 127，并使用隨后的 64 位來表示長度。

6. Masking-key

如果 Mask 位為 1，則 Masking-key 字段存在，并包含 32 位的掩碼。該掩碼用于對有效載荷數(shù)據(jù)進(jìn)行掩碼處理，以確保數(shù)據(jù)的安全性。

7. Payload data

Payload data 字段包含實(shí)際要傳輸?shù)臄?shù)據(jù)，它由擴(kuò)展數(shù)據(jù)（Extension data）和應(yīng)用數(shù)據(jù)（Application data）組成。擴(kuò)展數(shù)據(jù)是可選的，其長度和格式取決于在 WebSocket 握手階段協(xié)商的擴(kuò)展。應(yīng)用數(shù)據(jù)則包含了實(shí)際要傳輸?shù)臄?shù)據(jù)，如文本消息、二進(jìn)制數(shù)據(jù)等。

?? 思考一下：

提問：websocket/conn.go:31 中，maxFrameHeaderSize = 2 + 8 + 4 的含義是？

回答：2字節(jié)的固定長度 + 8字節(jié)的最大負(fù)載長度(可變) + 4字節(jié)掩碼（可變）。

三、WebSocket 握手過程

WebSocket 的握手過程是基于 HTTP 的。當(dāng)客戶端希望與服務(wù)器建立 WebSocket 連接時(shí)，它會首先發(fā)送一個(gè) HTTP 請求到服務(wù)器。這個(gè)請求包含了幾個(gè)關(guān)鍵的頭部字段，用于指示客戶端希望升級到 WebSocket 協(xié)議。

func main() {
	flag.Parse()
	hub := newHub()
	go hub.run()
	http.HandleFunc("/", serveHome) // HTTP 請求
	http.HandleFunc("/ws", func(w http.ResponseWriter, r *http.Request) {
		serveWs(hub, w, r) // HTTP 協(xié)議升級成 ws 協(xié)議
	})
	err := http.ListenAndServe(*addr, nil)
	if err != nil {
		log.Fatal("ListenAndServe: ", err)
	}
}

// serveWs handles websocket requests from the peer.
func serveWs(hub *Hub, w http.ResponseWriter, r *http.Request) {
	conn, err := upgrader.Upgrade(w, r, nil) // 劫持 conn，并升級為 ws 協(xié)議的函數(shù)，這部分是 websocket 庫實(shí)現(xiàn)的
	if err != nil {
		log.Println(err)
		return
	}
	client := &Client{hub: hub, conn: conn, send: make(chan []byte, 256)}
	client.hub.register <- client

	// Allow collection of memory referenced by the caller by doing all work in
	// new goroutines.
	go client.writePump()
	go client.readPump()
}

服務(wù)器在接收到這個(gè)請求后，會驗(yàn)證請求的有效性，并返回一個(gè)包含 Upgrade: websocket 和 Connection: Upgrade 頭部的 HTTP 響應(yīng)。這個(gè)響應(yīng)表示服務(wù)器同意升級到 WebSocket 協(xié)議，并且連接已經(jīng)成功建立。

四、WebSocket 數(shù)據(jù)傳輸機(jī)制

一旦 WebSocket 連接建立成功，客戶端和服務(wù)器就可以通過幀來傳輸數(shù)據(jù)了。每個(gè)幀都包含上述的幀結(jié)構(gòu)，用于標(biāo)識幀的類型、長度、掩碼以及實(shí)際的數(shù)據(jù)內(nèi)容。

在數(shù)據(jù)傳輸過程中，客戶端和服務(wù)器可以自由地發(fā)送和接收數(shù)據(jù)幀，實(shí)現(xiàn)全雙工通信。這種通信方式使得實(shí)時(shí)應(yīng)用能夠高效地處理數(shù)據(jù)交換，減少延遲和資源消耗。

五、一個(gè)聊天室的通信 demo

服務(wù)端通過維護(hù)一個(gè) hub 管理所有客戶端活躍的鏈接。

// 比如服務(wù)端通過維護(hù)一個(gè) hub 管理所有客戶端活躍的鏈接。
type Hub struct {
	// Registered clients.
	clients map[*Client]bool

	// Inbound messages from the clients.
	broadcast chan []byte

	// Register requests from the clients.
	register chan *Client

	// Unregister requests from clients.
	unregister chan *Client
}

服務(wù)端啟動協(xié)程監(jiān)聽客戶端實(shí)例的注冊，以及接收來自客戶端的消息，并且廣播給所有的客戶端。

func (h *Hub) run() {
    for {
       select {
       case client := <-h.register: // 升級成 ws 協(xié)議之后，注冊 client 到 hub
          h.clients[client] = true
       case client := <-h.unregister: // 連接斷開則注銷 client 實(shí)例，釋放資源
          if _, ok := h.clients[client]; ok {
             delete(h.clients, client)
             close(client.send)
          }
       case message := <-h.broadcast: // 一個(gè)阻塞的 channal，接收來自 client 的需要廣播的消息
          for client := range h.clients { // 發(fā)送給所有的 client 實(shí)例
             select {
             case client.send <- message:
             default:
                close(client.send)
                delete(h.clients, client)
             }
          }
       }
    }
}

客戶端的結(jié)構(gòu)。

// Client is a middleman between the websocket connection and the hub.
type Client struct {
    hub *Hub

    // The websocket connection.
    conn *websocket.Conn

    // Buffered channel of outbound messages.
    send chan []byte
}

從 http 協(xié)議升級成 ws 協(xié)議之后，將 client 注冊到 hub 中。

http.HandleFunc("/ws", func(w http.ResponseWriter, r *http.Request) {
    serveWs(hub, w, r) // HTTP 協(xié)議升級成 ws 協(xié)議
})

// serveWs handles websocket requests from the peer.
func serveWs(hub *Hub, w http.ResponseWriter, r *http.Request) {
	conn, err := upgrader.Upgrade(w, r, nil) // 升級為 ws 協(xié)議，劫持 conn，后續(xù)進(jìn)行全雙工通信
	if err != nil {
		log.Println(err)
		return
	}
	client := &Client{hub: hub, conn: conn, send: make(chan []byte, 256)}
	client.hub.register <- client

	// Allow collection of memory referenced by the caller by doing all work in
	// new goroutines.
	go client.writePump() // 開啟 client 向 web 端回寫消息的協(xié)程
	go client.readPump() // 開啟 client 監(jiān)聽來自 web 端的消息，并發(fā)送給 hub，進(jìn)行廣播
}

注冊在 hub 中的客戶端實(shí)例，不斷監(jiān)聽讀取來自 web 端的消息，并轉(zhuǎn)發(fā)給 hub。

func (c *Client) readPump() {
    defer func() {
       c.hub.unregister <- c
       c.conn.Close()
    }()
    c.conn.SetReadLimit(maxMessageSize)
    c.conn.SetReadDeadline(time.Now().Add(pongWait))
    c.conn.SetPongHandler(func(string) error { c.conn.SetReadDeadline(time.Now().Add(pongWait)); return nil })
    for {
       _, message, err := c.conn.ReadMessage() // 讀取來自 web 端的消息
       if err != nil {
          if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway, websocket.CloseAbnormalClosure) {
             log.Printf("error: %v", err)
          }
          break
       }
       message = bytes.TrimSpace(bytes.Replace(message, newline, space, -1)) // 格式化 
       c.hub.broadcast <- message
    }
}

注冊在 hub 中的客戶端實(shí)例，同時(shí)監(jiān)聽發(fā)送給自己的消息，并且寫入 conn 句柄，將消息再次轉(zhuǎn)發(fā)回 web 端的實(shí)例。

func (c *Client) writePump() {
    ticker := time.NewTicker(pingPeriod)
    defer func() {
       ticker.Stop()
       c.conn.Close()
    }()
    for {
       select {
       case message, ok := <-c.send: // 是否有需要回寫給 web 端的消息
          c.conn.SetWriteDeadline(time.Now().Add(writeWait))
          if !ok {
             // The hub closed the channel.
             c.conn.WriteMessage(websocket.CloseMessage, []byte{})
             return
          }

          w, err := c.conn.NextWriter(websocket.TextMessage)
          if err != nil {
             return
          }
          w.Write(message) // 發(fā)送文本消息

          // Add queued chat messages to the current websocket message.
          n := len(c.send)
          for i := 0; i < n; i++ {
             w.Write(newline)
             w.Write(<-c.send)
          }

          if err := w.Close(); err != nil {
             return
          }
       case <-ticker.C: // 周期發(fā)送 ping 消息
          c.conn.SetWriteDeadline(time.Now().Add(writeWait))
          if err := c.conn.WriteMessage(websocket.PingMessage, nil); err != nil {
             return
          }
       }
    }
}

六、WebSocket 協(xié)議在實(shí)際應(yīng)用中的優(yōu)勢

實(shí)時(shí)性：WebSocket 協(xié)議允許客戶端和服務(wù)器之間進(jìn)行實(shí)時(shí)的雙向通信，使得應(yīng)用能夠更快地響應(yīng)用戶操作。
高效性：與 HTTP 相比，WebSocket 無需每次通信都建立新的連接，從而大大減少了延遲和資源消耗。
靈活性：WebSocket 協(xié)議支持文本和二進(jìn)制數(shù)據(jù)的傳輸，使得應(yīng)用能夠處理多種類型的數(shù)據(jù)。
安全性：WebSocket 協(xié)議提供了對數(shù)據(jù)的掩碼處理，以確保數(shù)據(jù)在傳輸過程中的安全性。

小結(jié)

未完待續(xù)，期待你的關(guān)注。

posted on 2024-12-22 00:34 白澤talk 閱讀(704) 評論(0) 收藏舉報(bào)

刷新頁面返回頂部