Go 实现百万WebSocket连接的方法示例
大家好!我是 sergey kamardin,是 mail.ru 的一名工程师。
本文主要介绍如何使用 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 结构体
// 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 }
我想让你注意的是 reader
和 writer
goroutines。每个 goroutine 都需要内存栈,初始大小可能为 2 到 8 kb,具体 取决于操作系统 和 go 版本。
关于上面提到的 300 万个线上连接,为此我们需要消耗 24 gb 的内存(假设单个 goroutine 消耗 4 kb 栈内存)用于所有的连接。并且这还没包括为 channel
结构体分配的内存, ch.send
传出的数据包占用的内存以及其他内部字段的内存。
2.2 i/o goroutines
让我们来看看 reader
的实现:
// channel's reading goroutine. func (c *channel) reader() { // 创建一个缓冲 read 来减少 read 的系统调用 buf := bufio.newreader(c.conn) for { pkt, _ := readpacket(buf) c.handle(pkt) } }
这里我们使用了 bufio.reader
来减少 read()
系统调用的次数,并尽可能多地读取 buf
中缓冲区大小所允许的数量。在这个无限循环中,我们等待新数据的到来。请先记住这句话: 等待新数据的到来 。我们稍后会回顾。
我们先不考虑传入的数据包的解析和处理,因为它对我们讨论的优化并不重要。但是, buf
值得我们关注:默认情况下,它是 4 kb,这意味着连接还需要 12 gb 的内存。 writer
也有类似的情况:
// 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() } }
我们通过 channel 的 c.send
遍历将数据包传出 并将它们写入缓冲区。细心的读者可能猜到了,这是我们 300 万个连接的另外 12 gb 的内存消耗。
2.3 http
已经实现了一个简单的 channel
,现在我们需要使用 websocket 连接。由于仍然处于常用的方式的标题下,所以我们以常用的方式继续。
注意:如果你不知道 websocket 的运行原理,需要记住客户端会通过名为 upgrade 的特殊 http 机制转换到 websocket 协议。在成功处理 upgrade 请求后,服务端和客户端将使用 tcp 连接来传输二进制的 websocket 帧。 这里 是连接的内部结构的说明。
// 常用的转换为 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) //... })
需要注意的是, 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()
调用 :
// 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 } //... }
go 在非阻塞模式下使用套接字。 eagain 表示套接字中没有数据,并且读取空套接字时不会被锁定,操作系统将返回控制权给我们。(译者注:eagain 表示目前没有可用数据,请稍后再试)
我们从连接文件描述符中看到一个 read()
系统调用函数。如果 read 返回 eagain 错误 ,则 runtime 调用 polldesc.waitread() :
// 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) //... }
如果 深入挖掘 ,我们将看到 netpoll 在 linux 中是使用 epoll 实现的,而在 bsd 中是使用 kqueue 实现的。为什么不对连接使用相同的方法?我们可以分配一个 read 缓冲区并仅在真正需要时启动 read goroutine:当套接字中有可读的数据时。
在 github.com/golang/go 上,有一个导出 netpoll 函数的 issue 。
3.2 去除 goroutines 的内存消耗
假设我们有 go 的 netpoll 实现 。现在我们可以避免在内部缓冲区启动 channel.reader()
goroutine,而是在连接中订阅可读数据的事件:
// 使用 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) }
channel.writer()
更简单,因为我们只能在发送数据包时运行 goroutine 并分配缓冲区:
// 当我们需要时启动 writer goroutine func (ch *channel) send(p packet) { if c.nowriteryet() { go ch.writer() } ch.send <- p }
需要注意的是,当操作系统在 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 池限制同时处理的数据包数量。下面是这种池的简单实现:
// 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 <- 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 代码如下:
// 处理 goroutine 池中的轮询事件。 pool := gopool.new(128) poller.start(conn, netpoll.eventread, func() { // 我们在所有 worker 被占用时阻塞 poller pool.schedule(func() { receive(ch) }) })
现在我们不仅在套接字中有可读数据时读取,而且还在第一次机会获取池中的空闲 goroutine。??
同样,我们修改 send()
:
// 复用 writing goroutine pool := gopool.new(128) func (ch *channel) send(p packet) { if c.nowriteryet() { pool.schedule(ch.writer) } ch.send <- p }
取代 go ch.writer()
,我们想写一个复用的 goroutines。因此,对于拥有 n
个 goroutines 的池,我们可以保证同时处理 n
个请求并且在 n + 1
的时候, 我们不会分配 n + 1
个缓冲区。 goroutine 池还允许我们限制新连接的 accept()
和 upgrade()
,并避免大多数的 ddos 攻击。
3.4 upgrade 零拷贝
如前所述,客户端使用 http upgrade 切换到 websocket 协议。这就是 websocket 协议的样子:
## 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
也就是说,在我们的例子中,需要 http 请求及其 header 用于切换到 websocket 协议。这些知识以及 http.request
中存储的内容 表明,为了优化,我们需要在处理 http 请求时放弃不必要的内存分配和内存复制,并弃用 net/http
库。
例如, http.request
有一个与 header 具有相同名称的字段 ,这个字段用于将数据从连接中复制出来填充请求头。想象一下,该字段需要消耗多少额外内存,例如碰到比较大的 cookie 头。
websocket 的实现
不幸的是,在我们优化的时候所有存在的库都是使用标准的 net/http
库进行升级。而且,(两个)库都不能使用上述的读写优化方案。为了采用这些优化方案,我们需要用一个比较低级的 api 来处理 websocket。要重用缓冲区,我们需要把协议函数变成这样:
func readframe(io.reader) (frame, error) func writeframe(io.writer, frame) error
如果有一个这种 api 的库,我们可以按下面的方式从连接中读取数据包(数据包的写入也一样):
// 预期的 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) //... }
简单来说,我们需要自己的 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()
:
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 摘要
我们总结一下这些优化。
- 内部有缓冲区的 read goroutine 是代价比较大的。解决方案:netpoll(epoll,kqueue); 重用缓冲区。
- 内部有缓冲区的 write goroutine 是代价比较大的。解决方案:需要的时候才启动 goroutine; 重用缓冲区。
- 如果有大量的连接,netpoll 将无法正常工作。解决方案:使用 goroutines 池并限制池的 worker 数。
- net/http 不是处理升级到 websocket 的最快方法。解决方案:在裸 tcp 连接上使用内存零拷贝升级。
服务的代码看起来如下所示:
// 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) } }
总结
过早优化是编程中所有邪恶(或至少大部分)的根源。 -- donald knuth
当然,上述优化是和需求相关的,但并非所有情况下都是如此。例如,如果空闲资源(内存,cpu)和线上连接数之间的比率比较高,则优化可能没有意义。但是,通过了解优化的位置和内容,我们会受益匪浅。
感谢你的关注!
引用
russian version of this article
以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持。
下一篇: golang解析html网页的方法