网站首页 返回列表像“草根”一样，紧贴着地面，低调的存在，冬去春来，枯荣无恙。

性能优化实战：百万级WebSockets和Go语言

2020-06-10 03:14:27 admin 524

[翻译原文链接](https://link.jianshu.com?t=http://xiecode.cn/post/cn_06_a_million_websockets_and_go/)
转帖／转载请注明出处

[原文链接@medium.com](https://link.jianshu.com?t=https://medium.freecodecamp.org/million-
websockets-and-go-cc58418460bb) 发表于2017/08/03

大家好！我的名字叫Sergey Kamardin。我是来自 **Mail.Ru**
的一名工程师。这篇文章将讲述我们是如何用Go语言开发一个高负荷的WebSocket服务。即使你对WebSockets熟悉但对Go语言知之甚少，我还是希望这篇文章里讲到的性能优化的思路和技术对你有所启发。

## 1\. 介绍

作为全文的铺垫，我想先讲一下我们为什么要开发这个服务。

**Mail.Ru**
有许多包含状态的系统。用户的电子邮件存储是其中之一。有很多办法来跟踪这些状态的改变。不外乎通过定期的轮询或者系统通知来得到状态的变化。这两种方法都有它们的优缺点。对邮件这个产品来说，让用户尽快收到新的邮件是一个考量指标。邮件的轮询会产生大概每秒5万个HTTP请求，其中60%的请求会返回304状态（表示邮箱没有变化）。因此，为了减少服务器的负荷并加速邮件的接收，我们决定重写一个publisher-
subscriber服务（这个服务通常也会称作bus，message broker或者event-
channel）。这个服务负责接收状态更新的通知，然后还处理对这些更新的订阅。

重写publisher-subscriber服务之前：

0_pull.png

现在：

1_push.png

上面第一个图为旧的架构。浏览器（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）。[这里](https://link.jianshu.com?t=https://tools.ietf.org/html/rfc6455#section-5.2)有关于帧结构的描述。

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](https://link.jianshu.com?t=http://github.com/golang/go)上，有一个关于开放（exporting）netpoll函数的<a
> href="https://github.com/golang/go/issues/15735#issuecomment-266574151"
> target="_blank" rel="noopener">问题</a>。

### 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://github.com/mailru/easygo](https://link.jianshu.com?t=https://github.com/mailru/easygo)
  * [https://github.com/gobwas/ws](https://link.jianshu.com?t=https://github.com/gobwas/ws)
  * [https://github.com/gobwas/ws-examples](https://link.jianshu.com?t=https://github.com/gobwas/ws-examples)
  * [https://github.com/gobwas/httphead](https://link.jianshu.com?t=https://github.com/gobwas/httphead)

转载文章，原文链接：性能优化实战：百万级WebSockets和Go语言

关键字词go WebSockets

分享到：

上一篇：数据说话：Go语言的Switch和Map性能实测

下一篇：用Golang处理每分钟百万级请求

如需留言，请登录，没有账号？请注册

0 条评论 0 人参与

网站首页 返回列表像“草根”一样，紧贴着地面，低调的存在，冬去春来，枯荣无恙。

性能优化实战：百万级WebSockets和Go语言

admin 1年前

zhou 2年前

zhou 2年前

zhou 2年前

zhou 2年前

网站首页 返回列表 像“草根”一样，紧贴着地面，低调的存在，冬去春来，枯荣无恙。

性能优化实战：百万级WebSockets和Go语言

admin 官方 1年前

zhou 2年前

zhou 2年前

zhou 2年前

zhou 2年前

网站首页返回列表像“草根”一样，紧贴着地面，低调的存在，冬去春来，枯荣无恙。

admin 1年前