性能優化實戰：百萬級WebSockets和Go語言

MarylynYvb 7年前發布 | 64K 次閱讀 WebSocket Goroutine 性能優化 Google Go/Golang開發

大家好！我的名字叫Sergey Kamardin。我是來自 Mail.Ru 的一名工程師。這篇文章將講述我們是如何用Go語言開發一個高負荷的WebSocket服務。即使你對WebSockets熟悉但對Go語言知之甚少，我還是希望這篇文章里講到的性能優化的思路和技術對你有所啟發。

1. 介紹

作為全文的鋪墊，我想先講一下我們為什么要開發這個服務。

Mail.Ru有許多包含狀態的系統。用戶的電子郵件存儲是其中之一。有很多辦法來跟蹤這些狀態的改變。不外乎通過定期的輪詢或者系統通知來得到狀態的變化。這兩種方法都有它們的優缺點。對郵件這個產品來說，讓用戶盡快收到新的郵件是一個考量指標。郵件的輪詢會產生大概每秒5萬個HTTP請求，其中60%的請求會返回304狀態（表示郵箱沒有變化）。因此，為了減少服務器的負荷并加速郵件的接收，我們決定重寫一個publisher-subscriber服務（這個服務通常也會稱作bus，message broker或者event-channel）。這個服務負責接收狀態更新的通知，然后還處理對這些更新的訂閱。

重寫publisher-subscriber服務之前：

現在：

上面第一個圖為舊的架構。瀏覽器（Browser）會定期輪詢API服務來獲得郵件存儲服務（Storage）的更新。

第二張圖展示的是新的架構。瀏覽器（Browser）和通知API服務（notificcation API）建立一個WebSocket連接。通知API服務會發送相關的訂閱到Bus服務上。當收到新的電子郵件時，存儲服務（Storage）向Bus（1）發送一個通知，Bus又將通知發送給相應的訂閱者（2）。API服務為收到的通知找到相應的連接，然后把通知推送到用戶的瀏覽器（3）。

我們今天就來討論一下這個API服務（也可以叫做WebSocket服務）。在開始之前，我想提一下這個在線服務處理將近3百萬個連接。

2. 慣用的做法（The idiomatic way）

首先，我們看一下不做任何優化會如何用Go來實現這個服務的部分功能。在使用 net/http 實現具體功能前，讓我們先討論下我們將如何發送和接收數據。這些數據是定義在WebSocket協議之上的（例如JSON對象）。我們在下文中會成他們為packet。

我們先來實現 Channel 結構。它包含相應的邏輯來通過WebScoket連接發送和接收packet。

2.1. Channel結構

// Packet represents application level data.
type Packet struct {
    ...
}

// Channel wraps user connection.
type Channel struct {
    conn net.Conn    // WebSocket connection.
    send chan Packet // Outgoing packets queue.
}

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

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

    return c
}

這里我要強調的是讀和寫這兩個goroutines。每個goroutine都需要各自的內存棧。棧的初始大小由操作系統和Go的版本決定，通常在2KB到8KB之間。我們之前提到有3百萬個在線連接，如果每個goroutine棧需要4KB的話，所有連接就需要24GB的內存。這還沒算上給 Channel 結構，發送packet用的 ch.send 和其它一些內部字段分配的內存空間。

2.2. I/O goroutines

接下來看一下“reader”的實現：

func (c *Channel) reader() {
    // We make a buffered read to reduce read syscalls.
    buf := bufio.NewReader(c.conn)

    for {
        pkt, _ := readPacket(buf)
        c.handle(pkt)
    }
}

這里我們使用了 bufio.Reader 。每次都會在 buf 大小允許的范圍內盡量讀取多的字節，從而減少 read() 系統調用的次數。在無限循環中，我們期望會接收到新的數據。請記住之前這句話：期望接收到新的數據。我們之后會討論到這一點。

我們把packet的解析和處理邏輯都忽略掉了，因為它們和我們要討論的優化不相關。不過 buf 值得我們的關注：它的缺省大小是4KB。這意味著所有連接將消耗掉額外的 12 GB 內存。“writer”也是類似的情況：

func (c *Channel) writer() {
    // We make buffered write to reduce write syscalls.
    buf := bufio.NewWriter(c.conn)

    for pkt := range c.send {
        _ := writePacket(buf, pkt)
        buf.Flush()
    }
}

我們在待發送packet的 c.send channel上循環將packet寫到緩存（buffer）里。細心的讀者肯定已經發現，這又是額外的4KB內存。3百萬個連接會占用 12GB 的內存。

2.3. HTTP

我們已經有了一個簡單的 Channel 實現。現在我們需要一個WebSocket連接。因為還在 通常做法（Idiomatic Way) 的標題下，那么就先來看看通常是如何實現的。

注：如果你不知道WebSocket是怎么工作的，那么這里值得一提的是客戶端是通過一個叫升級（Upgrade）請求的特殊HTTP機制來建立WebSocket的。在成功處理升級請求以后，服務端和客戶端使用TCP連接來交換二進制的WebSocket幀（frames）。這里有關于幀結構的描述。

import (
    "net/http"
    "some/websocket"
)

http.HandleFunc("/v1/ws", func(w http.ResponseWriter, r *http.Request) {
    conn, _ := websocket.Upgrade(r, w)
    ch := NewChannel(conn)
    //...
})

請注意這里的 http.ResponseWriter 結構包含 bufio.Reader 和 bufio.Writer （各自分別包含4KB的緩存）。它們用于 \*http.Request 初始化和返回結果。

不管是哪個WebSocket，在成功回應一個升級請求之后，服務端在調用 responseWriter.Hijack() 之后會接收到一個I/O緩存和對應的TCP連接。

注：有時候我們可以通過 net/http.putBufio{Reader,Writer} 調用把緩存釋放回 net/http 里的 sync.Pool 。

這樣，這3百萬個連接又需要額外的 24 GB 內存。

所以，為了這個什么都不干的程序，我們已經占用了 72 GB 的內存！

3. 優化

我們來回顧一下前面介紹的用戶連接的工作流程。在建立WebSocket之后，客戶端會發送請求訂閱相關事件（我們這里忽略類似 ping/pong 的請求）。接下來，在整個連接的生命周期里，客戶端可能就不會發送任何其它數據了。

連接的生命周期可能會持續幾秒鐘到幾天。

所以在大部分時間里， Channel.reader() 和 Channel.writer() 都在等待接收和發送數據。與它們一起等待的是各自分配的4 KB的I/O緩存。

現在，我們發現有些地方是可以做進一步優化的，對吧？

3.1. Netpoll

你還記得 Channel.reader() 的實現使用了 bufio.Reader.Read() 嗎？ bufio.Reader.Read() 又會調用 conn.Read() 。這個調用會被阻塞以等待接收連接上的新數據。如果連接上有新的數據，Go的運行環境（runtime）就會喚醒相應的goroutine讓它去讀取下一個packet。之后，goroutine會被再次阻塞來等待新的數據。我們來研究下Go的運行環境是怎么知道goroutine需要被喚醒的。

如果我們看一下 conn.Read() 的實現，就會看到它調用了 net.netFD.Read() ：

// 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
    }
    //...
}

Go使用了sockets的非阻塞模式。EAGAIN表示socket里沒有數據了但不會阻塞在空的socket上，OS會把控制權返回給用戶進程。

這里它首先對連接文件描述符進行 read() 系統調用。如果 read() 返回的是 EAGAIN 錯誤，運行環境就是調用 pollDesc.waitRead() ：

// 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)
   //...
}

如果繼續深挖，我們可以看到netpoll的實現在Linux里用的是epoll而在BSD里用的是kqueue。我們的這些連接為什么不采用類似的方式呢？只有在socket上有可讀數據時，才分配緩存空間并啟用讀數據的goroutine。

在github.com/golang/go上，有一個關于開放（exporting）netpoll函數的問題。

3.2. 干掉goroutines

假設我們用Go語言實現了netpoll。我們現在可以避免創建 Channel.reader() 的goroutine，取而代之的是從訂閱連接里收到新數據的事件。

ch := NewChannel(conn)

// Make conn to be observed by netpoll instance.
poller.Start(conn, netpoll.EventRead, func() {
    // We spawn goroutine here to prevent poller wait loop
    // to become locked during receiving packet from ch.
    go ch.Receive()
})

// Receive reads a packet from conn and handles it somehow.
func (ch *Channel) Receive() {
    buf := bufio.NewReader(ch.conn)
    pkt := readPacket(buf)
    c.handle(pkt)
}

Channel.writer() 相對容易一點，因為我們只需在發送packet的時候創建goroutine并分配緩存。

func (ch *Channel) Send(p Packet) {
    if c.noWriterYet() {
        go ch.writer()
    }
    ch.send <- p
}

注意，這里我們沒有處理 write() 系統調用時返回的 EAGAIN 。我們依賴Go運行環境去處理它。這種情況很少發生。如果需要的話我們還是可以像之前那樣來處理。

從 ch.send 讀取待發送的packets之后， ch.writer() 會完成它的操作，最后釋放goroutine的棧和用于發送的緩存。

很不錯！通過避免這兩個連續運行的goroutine所占用的I/O緩存和棧內存，我們已經節省了 48 GB 。

3.3. 控制資源

大量的連接不僅僅會造成大量的內存消耗。在開發服務端的時候，我們還不停地遇到競爭條件（race conditions）和死鎖（deadlocks）。隨之而來的是所謂的自我分布式阻斷攻擊（self-DDOS）。在這種情況下，客戶端會悍然地嘗試重新連接服務端而把情況搞得更加糟糕。

舉個例子，如果因為某種原因我們突然無法處理 ping/pong 消息，這些空閑連接就會不斷地被關閉（它們會以為這些連接已經無效因此不會收到數據）。然后客戶端每N秒就會以為失去了連接并嘗試重新建立連接，而不是繼續等待服務端發來的消息。

在這種情況下，比較好的辦法是讓負載過重的服務端停止接受新的連接，這樣負載均衡器（例如nginx）就可以把請求轉到其它的服務端上去。

撇開服務端的負載不說，如果所有的客戶端突然（很可能是因為某個bug）向服務端發送一個packet，我們之前節省的 48 GB 內存又將會被消耗掉。因為這時我們又會和開始一樣給每個連接創建goroutine并分配緩存。

Goroutine池

可以用一個goroutine池來限制同時處理packets的數目。下面的代碼是一個簡單的實現：

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 <- task:
    case p.sem <- struct{}{}:
        go p.worker(task)
    }
}

func (p *Pool) worker(task func()) {
    defer func() { <-p.sem }
    for {
        task()
        task = <-p.work
    }
}

我們使用netpoll的代碼就變成下面這樣：

pool := gopool.New(128)

poller.Start(conn, netpoll.EventRead, func() {
    // We will block poller wait loop when
    // all pool workers are busy.
    pool.Schedule(func() {
        ch.Receive()
    })
})

現在我們不僅要等可讀的數據出現在socket上才能讀packet，還必須等到從池里獲取到空閑的goroutine。

同樣的，我們修改下 Send() 的代碼：

pool := gopool.New(128)

func (ch *Channel) Send(p Packet) {
    if c.noWriterYet() {
        pool.Schedule(ch.writer)
    }
    ch.send <- p
}

這里我們沒有調用 go ch.writer() ，而是想重復利用池里goroutine來發送數據。所以，如果一個池有 N 個goroutines的話，我們可以保證有 N 個請求被同時處理。而 N + 1 個請求不會分配 N + 1 個緩存。goroutine池允許我們限制對新連接的 Accept() 和 Upgrade() ，這樣就避免了大部分DDoS的情況。

3.4. 零拷貝升級（Zero-copy upgrade）

之前已經提到，客戶端通過HTTP升級（Upgrade）請求切換到WebSocket協議。下面顯示的是一個升級請求：

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

我們接收HTTP請求和它的頭部只是為了切換到WebSocket協議，而 http.Request 里保存了所有頭部的數據。從這里可以得到啟發，如果是為了優化，我們可以放棄使用標準的 net/http 服務并在處理HTTP請求的時候避免無用的內存分配和拷貝。

舉個例子， http.Request 包含了一個叫做Header的字段。標準 net/http 服務會將請求里的所有頭部數據全部無條件地拷貝到Header字段里。你可以想象這個字段會保存許多冗余的數據，例如一個包含很長cookie的頭部。

我們如何來優化呢？

WebSocket實現

不幸的是，在我們優化服務端的時候所有能找到的庫只支持對標準 net/http 服務做升級。而且沒有一個庫允許我們實現上面提到的讀和寫的優化。為了使這些優化成為可能，我們必須有一套底層的API來操作WebSocket。為了重用緩存，我們需要類似下面這樣的協議函數：

func ReadFrame(io.Reader) (Frame, error)
func WriteFrame(io.Writer, Frame) error

如果我們有一個包含這樣API的庫，我們就按照下面的方式從連接上讀取packets：

// getReadBuf, putReadBuf are intended to
// reuse *bufio.Reader (with sync.Pool for example).
func getReadBuf(io.Reader) *bufio.Reader
func putReadBuf(*bufio.Reader)

// readPacket must be called when data could be read from conn.
func readPacket(conn io.Reader) error {
    buf := getReadBuf()
    defer putReadBuf(buf)

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

簡而言之，我們需要自己寫一個庫。

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 來驗證一個session）。

下面是處理升級請求的性能測試：標準 net/http 庫的實現和使用零拷貝升級的 net.Listen() ：

BenchmarkUpgradeHTTP    5156 ns/op    8576 B/op    9 allocs/op
BenchmarkUpgradeTCP     973 ns/op     0 B/op       0 allocs/op

使用 ws 以及零拷貝升級為我們節省了24 GB的空間。這些空間原本被用做 net/http 里處理請求的I/O緩存。

3.5. 回顧

讓我們來回顧一下之前提到過的優化：

一個包含緩存的讀goroutine會占用很多內存。 方案： netpoll（epoll, kqueue）；重用緩存。
一個包含緩存的寫goroutine會占用很多內存。 方案： 在需要的時候創建goroutine；重用緩存。
存在大量連接請求的時候，netpoll不能很好的限制連接數。 方案： 重用goroutines并且限制它們的數目。
net/http 對升級到WebSocket請求的處理不是最高效的。 方案： 在TCP連接上實現零拷貝升級。

下面是服務端的大致實現代碼：

import (
    "net"
    "github.com/gobwas/ws"
)

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

for {
    // Try to accept incoming connection inside free pool worker.
    // If there no free workers for 1ms, do not accept anything and try later.
    // This will help us to prevent many self-ddos or out of resource limit cases.
    err := pool.ScheduleTimeout(time.Millisecond, func() {
        conn := ln.Accept()
        _ = ws.Upgrade(conn)

        // Wrap WebSocket connection with our Channel struct.
        // This will help us to handle/send our app's packets.
        ch := NewChannel(conn)

        // Wait for incoming bytes from connection.
        poller.Start(conn, netpoll.EventRead, func() {
            // Do not cross the resource limits.
            pool.Schedule(func() {
                // Read and handle incoming packet(s).
                ch.Recevie()
            })
        })
    })
    if err != nil {
        time.Sleep(time.Millisecond)
    }
}

4. 結論

在程序設計時，過早優化是萬惡之源。Donald Knuth

上面的優化是有意義的，但不是所有情況都適用。舉個例子，如果空閑資源（內存，CPU）與在線連接數之間的比例很高的話，優化就沒有太多意義。當然，知道什么地方可以優化以及如何優化總是有幫助的。

謝謝你的關注！

5. 引用

來自：https://segmentfault.com/a/1190000011162605

本文由用戶 MarylynYvb 自行上傳分享，僅供網友學習交流。所有權歸原作者，若您的權利被侵害，請聯系管理員。

轉載本站原創文章，請注明出處，并保留原始鏈接、圖片水印。

本站是一個以用戶分享為主的開源技術平臺，歡迎各類分享！

本文地址：http://www.baiduhome.net/lib/view/open1505370872096.html

WebSocket Goroutine 性能優化 Google Go/Golang開發