大家好！我是 Echa，是互聯網的一名工程師。先奉上驚喜給讀者你，打開看看《 》

本文主要介紹如何使用 Go 開發高負載的 WebSocket 服務。

如果你熟悉 WebSockets，但對 Go 瞭解不多，仍希望你對這篇文章的想法和性能優化方面感興趣。

1. 簡介

為了定義本文的討論範圍，有必要說明我們為什麼需要這個服務。

Mail.Ru 有很多有狀態系統。用戶的電子郵件存儲就是其中之一。我們有幾種方法可以跟蹤該系統的狀態變化以及系統事件，主要是通過定期系統輪詢或者狀態變化時的系統通知來實現。

兩種方式各有利弊。但是對於郵件而言，用戶收到新郵件的速度越快越好。

郵件輪詢大約每秒 50,000 個 HTTP 查詢，其中 60％ 返回 304 狀態，這意味著郵箱中沒有任何更改。

因此，為了減少服務器的負載並加快向用戶發送郵件的速度，我們決定通過用發佈 - 訂閱服務（也稱為消息總線，消息代理或事件管道）的模式來造一個輪子。一端接收有關狀態更改的通知，另一端訂閱此類通知。

之前的架構：

現在的架構：

第一個方案是之前的架構。瀏覽器定期輪詢 API 並查詢存儲（郵箱服務）是否有更改。

第二種方案是現在的架構。瀏覽器與通知 API 建立了 WebSocket 連接，通知 API 是總線服務的消費者。一旦接收到新郵件後，Storage 會將有關它的通知發送到總線（1），總線將其發送給訂閱者（2）。 API 通過連接發送這個收到的通知，將其發送到用戶的瀏覽器（3）。

所以現在我們將討論這個 API 或者這個 WebSocket 服務。展望一下未來，我們的服務將來可能會有 300 萬個在線連接。

2. 常用的方式

我們來看看如何在沒有任何優化的情況下使用 Go 實現服務器的某些部分。

在我們繼續使用 net/http 之前，來談談如何發送和接收數據。這個數據位於 WebSocket 協議上（例如 JSON 對象），我們在下文中將其稱為包。

我們先來實現 Channel 結構體，該結構體將包含在 WebSocket 連接上發送和接收數據包的邏輯。

2.1 Channel 結構體

<code>// WebSocket Channel 的實現
// Packet 結構體表示應用程序級數據
type Packet struct {
    ...
}

// Channel 裝飾用戶連接
type Channel struct {
    conn net.Conn    // WebSocket 連接
    send chan Packet // 傳出的 packets 隊列
}

func NewChannel(conn net.Conn) *Channel {
    c := &Channel{
        conn: conn,
        send: make(chan Packet, N),
    }
 

    go c.reader()
    go c.writer()

    return c
}
複製代碼/<code>

我想讓你注意的是 reader 和 writer goroutines。每個 goroutine 都需要內存棧，初始大小可能為 2 到 8 KB，具體取決於操作系統和 Go 版本。

關於上面提到的 300 萬個線上連接，為此我們需要消耗 24 GB 的內存（假設單個 goroutine 消耗 4 KB 棧內存）用於所有的連接。並且這還沒包括為 Channel 結構體分配的內存，ch.send傳出的數據包占用的內存以及其他內部字段的內存。

2.2 I/O goroutines

讓我們來看看 reader 的實現：

<code>// Channel’s reading goroutine.
func (c *Channel) reader() {
    // 創建一個緩衝 read 來減少 read 的系統調用
    buf := bufio.NewReader(c.conn)

    for {
        pkt, _ := readPacket(buf)
        c.handle(pkt)
    }
}
複製代碼/<code>

這裡我們使用了 bufio.Reader 來減少 read() 系統調用的次數，並儘可能多地讀取 buf 中緩衝區大小所允許的數量。在這個無限循環中，我們等待新數據的到來。請先記住這句話：等待新數據的到來。我們稍後會回顧。

我們先不考慮傳入的數據包的解析和處理，因為它對我們討論的優化並不重要。但是，buf 值得我們關注：默認情況下，它是 4 KB，這意味著連接還需要 12 GB 的內存。writer 也有類似的情況：

<code>// Channel’s writing goroutine.
func (c *Channel) writer() {
    // 創建一個緩衝 write 來減少 write 的系統調用
    buf := bufio.NewWriter(c.conn)

    for pkt := range c.send {
        _ := writePacket(buf, pkt)
        buf.Flush()
    }
}
複製代碼/<code>

我們通過 Channel 的 c.send 遍歷將數據包傳出 並將它們寫入緩衝區。細心的讀者可能猜到了，這是我們 300 萬個連接的另外 12 GB 的內存消耗。

2.3 HTTP

已經實現了一個簡單的 Channel，現在我們需要使用 WebSocket 連接。由於仍然處於常用的方式的標題下，所以我們以常用的方式繼續。

注意：如果你不知道 WebSocket 的運行原理，需要記住客戶端會通過名為 Upgrade 的特殊 HTTP 機制轉換到 WebSocket 協議。在成功處理 Upgrade 請求後，服務端和客戶端將使用 TCP 連接來傳輸二進制的 WebSocket 幀。這裡是連接的內部結構的說明。

<code>// 常用的轉換為 WebSocket 的方法
import (
    "net/http"
    "some/websocket"
)

http.HandleFunc("/v1/ws", func(w http.ResponseWriter, r *http.Request) {
    conn, _ := websocket.Upgrade(r, w)
    ch := NewChannel(conn)
    //...
})
複製代碼/<code>

需要注意的是，http.ResponseWriter 為 bufio.Reader 和 bufio.Writer（均為 4 KB 的緩衝區）分配了內存，用於對 *http.Request 的初始化和進一步的響應寫入。

無論使用哪種 WebSocket 庫，在 Upgrade 成功後，服務端在調用 responseWriter.Hijack() 之後都會收到 I/O 緩衝區和 TCP 連接。

提示：在某些情況下，go:linkname 可被用於通過調用 net/http.putBufio {Reader, Writer} 將緩衝區返回給 net/http 內的 sync.Pool。

因此，我們還需要 24 GB 的內存用於 300 萬個連接。

那麼，現在為了一個什麼功能都沒有的應用程序，一共需要消耗 72 GB 的內存！

3. 優化

我們回顧一下在簡介部分中談到的內容，並記住用戶連接的方式。在切換到 WebSocket 後，客戶端會通過連接發送包含相關事件的數據包。然後（不考慮 ping/pong 等消息），客戶端可能在整個連接的生命週期中不會發送任何其他內容。

連接的生命週期可能持續幾秒到幾天。

因此，大部分時間 Channel.reader() 和 Channel.writer() 都在等待接收或發送數據。與它們一起等待的還有每個大小為 4 KB 的 I/O 緩衝區。

現在我們對哪些地方可以做優化應該比較清晰了。

3.1 Netpoll

Channel.reader() 通過給 bufio.Reader.Read() 內的 conn.Read() 加鎖來等待新數據的到來（譯者注：上文中的伏筆），一旦連接中有數據，Go runtime（譯者注：runtime 包含 Go 運行時的系統交互的操作，這裡保留原文）“喚醒” goroutine 並允許它讀取下一個數據包。在此之後，goroutine 再次被鎖定，同時等待新的數據。讓我們看看 Go runtime 來理解 goroutine 為什麼必須“被喚醒”。

如果我們查看 conn.Read() 的實現，將會在其中看到 net.netFD.Read() 調用：

<code>// Go 內部的非阻塞讀.
// net/fd_unix.go

func (fd *netFD) Read(p []byte) (n int, err error) {
    //...
    for {
        n, err = syscall.Read(fd.sysfd, p)
        if err != nil {
            n = 0
            if err == syscall.EAGAIN {
                if err = fd.pd.waitRead(); err == nil {
                    continue
                } 

            }
        }
        //...
        break
    }
    //...
}
複製代碼/<code>

Go 在非阻塞模式下使用套接字。 EAGAIN 表示套接字中沒有數據，並且讀取空套接字時不會被鎖定，操作系統將返回控制權給我們。(譯者注：EAGAIN 表示目前沒有可用數據，請稍後再試)

我們從連接文件描述符中看到一個 read() 系統調用函數。如果 read 返回 EAGAIN 錯誤，則 runtime 調用 pollDesc.waitRead()：

<code>// Go 內部關於 netpoll 的使用
// net/fd_poll_runtime.go

func (pd *pollDesc) waitRead() error {
   return pd.wait('r')
}

func (pd *pollDesc) wait(mode int) error {
   res := runtime_pollWait(pd.runtimeCtx, mode)
   //...
}
複製代碼/<code>

如果深入挖掘，我們將看到 netpoll 在 Linux 中是使用 epoll 實現的，而在 BSD 中是使用 kqueue 實現的。為什麼不對連接使用相同的方法？我們可以分配一個 read 緩衝區並僅在真正需要時啟動 read goroutine：當套接字中有可讀的數據時。

在 github.com/golang/go 上，有一個導出 netpoll 函數的 issue。

3.2 去除 goroutines 的內存消耗

假設我們有 Go 的 netpoll 實現。現在我們可以避免在內部緩衝區啟動 Channel.reader() goroutine，而是在連接中訂閱可讀數據的事件：

<code>// 使用 netpoll
ch := NewChannel(conn)

// 通過 netpoll 實例觀察 conn
poller.Start(conn, netpoll.EventRead, func() {
    // 我們在這裡產生 goroutine 以防止在輪詢從 ch 接收數據包時被鎖。
    go Receive(ch)
})

// Receive 從 conn 讀取數據包並以某種方式處理它。
func (ch *Channel) Receive() {
    buf := bufio.NewReader(ch.conn)
    pkt := readPacket(buf)
    c.handle(pkt)
}
複製代碼/<code>

Channel.writer() 更簡單，因為我們只能在發送數據包時運行 goroutine 並分配緩衝區：

<code>// 當我們需要時啟動 writer goroutine
func (ch *Channel) Send(p Packet) {
    if c.noWriterYet() {
        go ch.writer()
    }
    ch.send }
複製代碼/<code>

需要注意的是，當操作系統在 write() 調用上返回 EAGAIN 時，我們不處理這種情況。我們依靠 Go runtime 來處理這種情況，因為這種情況在服務器上很少見。然而，如果有必要，它可以以與 reader() 相同的方式處理。

當從 ch.send（一個或幾個）讀取傳出數據包後，writer 將完成其操作並釋放 goroutine 的內存和發送緩衝區的內存。

完美！我們通過去除兩個運行的 goroutine 中的內存消耗和 I/O 緩衝區的內存消耗節省了 48 GB。

3.3 資源控制

大量連接不僅僅涉及到內存消耗高的問題。在開發服務時，我們遇到了反覆出現的競態條件和 self-DDoS 造成的死鎖。

例如，如果由於某種原因我們突然無法處理 ping/pong 消息，但是空閒連接的處理程序繼續關閉這樣的連接（假設連接被破壞，沒有提供數據），客戶端每隔 N 秒失去連接並嘗試再次連接而不是等待事件。

被鎖或超載的服務器停止服務，如果它之前的負載均衡器（例如，nginx）將請求傳遞給下一個服務器實例，這將是不錯的。

此外，無論服務器負載如何，如果所有客戶端突然（可能是由於錯誤原因）向我們發送數據包，之前的 48 GB 內存的消耗將不可避免，因為需要為每個連接分配 goroutine 和緩衝區。

Goroutine 池

上面的情況，我們可以使用 goroutine 池限制同時處理的數據包數量。下面是這種池的簡單實現：

<code>// goroutine 池的簡單實現
package gopool

func New(size int) *Pool {
    return &Pool{
        work: make(chan func()),
        sem:  make(chan struct{}, size),
    }
}

func (p *Pool) Schedule(task func()) error {
    select {
    case p.work     case p.sem         go p.worker(task)
    }
}

func (p *Pool) worker(task func()) {
    defer func() {     for {
        task()
        task =     }
}
複製代碼/<code>

現在我們的 netpoll 代碼如下：

<code>// 處理 goroutine 池中的輪詢事件。
pool := gopool.New(128)

poller.Start(conn, netpoll.EventRead, func() {
    // 我們在所有 worker 被佔用時阻塞 poller
    pool.Schedule(func() {
        Receive(ch)
    })
})
複製代碼/<code>

現在我們不僅在套接字中有可讀數據時讀取，而且還可以佔用池中的空閒的 goroutine。

同樣，我們修改 Send()：

<code>// 複用 writing goroutine
pool := gopool.New(128)

func (ch *Channel) Send(p Packet) {
    if c.noWriterYet() {
        pool.Schedule(ch.writer)
    }
    ch.send }
複製代碼/<code>

取代 go ch.writer() ，我們想寫一個複用的 goroutines。因此，對於擁有 N 個 goroutines 的池，我們可以保證同時處理 N 個請求並且在 N + 1的時候， 我們不會分配 N + 1 個緩衝區。 goroutine 池還允許我們限制新連接的 Accept() 和 Upgrade() ，並避免大多數的 DDoS 攻擊。

3.4 upgrade 零拷貝

如前所述，客戶端使用 HTTP Upgrade 切換到 WebSocket 協議。這就是 WebSocket 協議的樣子：

<code>## HTTP Upgrade 示例

GET /ws HTTP/1.1
Host: mail.ru
Connection: Upgrade
Sec-Websocket-Key: A3xNe7sEB9HixkmBhVrYaA==
Sec-Websocket-Version: 13
Upgrade: websocket

HTTP/1.1 101 Switching Protocols
Connection: Upgrade
Sec-Websocket-Accept: ksu0wXWG+YmkVx+KQR2agP0cQn4=
Upgrade: websocket
複製代碼/<code>

也就是說，在我們的例子中，需要 HTTP 請求及其 Header 用於切換到 WebSocket 協議。這些知識以及 http.Request 中存儲的內容表明，為了優化，我們需要在處理 HTTP 請求時放棄不必要的內存分配和內存複製，並棄用 net/http 庫。

例如，http.Request 有一個與 Header 具有相同名稱的字段，這個字段用於將數據從連接中複製出來填充請求頭。想象一下，該字段需要消耗多少額外內存，例如碰到比較大的 Cookie 頭。

WebSocket 的實現

不幸的是，在我們優化的時候所有存在的庫都是使用標準的 net/http 庫進行升級。而且，（兩個）庫都不能使用上述的讀寫優化方案。為了採用這些優化方案，我們需要用一個比較低級的 API 來處理 WebSocket。要重用緩衝區，我們需要把協議函數變成這樣：

<code>func ReadFrame(io.Reader) (Frame, error)
func WriteFrame(io.Writer, Frame) error
複製代碼/<code>

如果有一個這種 API 的庫，我們可以按下面的方式從連接中讀取數據包（數據包的寫入也一樣）：

<code>// 預期的 WebSocket 實現API
// getReadBuf, putReadBuf 用來複用 *bufio.Reader (with sync.Pool for example).
func getReadBuf(io.Reader) *bufio.Reader
func putReadBuf(*bufio.Reader)

// 當 conn 中的數據可讀取時，readPacket 被調用
func readPacket(conn io.Reader) error {
    buf := getReadBuf()
    defer putReadBuf(buf)

    buf.Reset(conn)
    frame, _ := ReadFrame(buf)
    parsePacket(frame.Payload) 

    //...
}
複製代碼/<code>

簡單來說，我們需要自己的 WebSocket 庫。

github.com/gobwas/ws

在意識形態上，編寫 ws 庫是為了不將其協議操作邏輯強加給用戶。所有讀寫方法都實現了標準的 io.Reader 和 io.Writer 接口，這樣就可以使用或不使用緩衝或任何其他 I/O 。

除了來自標準庫 net/http 的升級請求之外，ws 還支持零拷貝升級，升級請求的處理以及切換到 WebSocket 無需分配內存或複製內存。ws.Upgrade() 接受 io.ReadWriter（net.Conn 實現了此接口）。換句話說，我們可以使用標準的 net.Listen() 將接收到的連接從 ln.Accept() 轉移給 ws.Upgrade() 。該庫使得可以複製任何請求數據以供應用程序使用（例如，Cookie 用來驗證會話）。

下面是升級請求的基準測試結果：標準庫 net/http 的服務與用零拷貝升級的 net.Listen()：

<code>BenchmarkUpgradeHTTP    5156 ns/op    8576 B/op    9 allocs/op
BenchmarkUpgradeTCP     973 ns/op     0 B/op       0 allocs/op
複製代碼/<code>

切換到 ws 和零拷貝升級為我們節省了另外的 24 GB 內存 - 在 net/http 處理請求時為 I/O 緩衝區分配的空間。

3.5 摘要

我們總結一下這些優化。

• 內部有緩衝區的 read goroutine 是代價比較大的。解決方案：netpoll（epoll，kqueue）; 重用緩衝區。
• 內部有緩衝區的 write goroutine 是代價比較大的。解決方案：需要的時候才啟動 goroutine; 重用緩衝區。
• 如果有大量的連接，netpoll 將無法正常工作。解決方案：使用 goroutines 池並限制池的 worker 數。
• net/http 不是處理升級到 WebSocket 的最快方法。解決方案：在裸 TCP 連接上使用內存零拷貝升級。

服務的代碼看起來如下所示：

<code>// WebSocket 服務器示例，包含 netpoll，goroutine 池和內存零拷貝的升級。
import (
    "net"
    "github.com/gobwas/ws"
)

ln, _ := net.Listen("tcp", ":8080")

for {
    // 嘗試在空閒池的 worker 內的接收傳入的連接。如果超過 1ms 沒有空閒 worker，則稍後再試。這有助於防止 self-ddos 或耗盡服務器資源的情況。
    err := pool.ScheduleTimeout(time.Millisecond, func() {
        conn := ln.Accept()
        _ = ws.Upgrade(conn) 


        // 使用 Channel 結構體包裝 WebSocket 連接
        // 將幫助我們處理應用包
        ch := NewChannel(conn)

        // 等待連接傳入字節
        poller.Start(conn, netpoll.EventRead, func() {
            // 不要超過資源限制
            pool.Schedule(func() {
                // 讀取並處理傳入的包
                ch.Recevie()
            })
        })
    })
    if err != nil {
        time.Sleep(time.Millisecond)
    }
}
複製代碼/<code>

總結

過早優化是編程中所有邪惡（或至少大部分）的根源。 -- Donald Knuth

當然，上述優化是和需求相關的，但並非所有情況下都是如此。例如，如果空閒資源（內存，CPU）和線上連接數之間的比率比較高，則優化可能沒有意義。但是，通過了解優化的位置和內容，我們會受益匪淺。

感謝你的關注！

閱讀更多 Echa攻城獅 的文章

關鍵字: 瀏覽器 操作系統 百萬