Tony Bai - Part 116

Go基于I/O多路复用的TCP协议流解析实践

七月 31, 2021
0 条评论

本文永久链接 – https://tonybai.com/2021/07/31/io-multiplexing-model-tcp-stream-protocol-parsing-practice-in-go

在《Go经典阻塞式TCP协议流解析的实践》一文中，我们基于Go经典的阻塞I/O模型实现了一个基于TCP流的自定义协议的解析。这种one-connection-per-goroutine模型的优点就是简单、好写以及好理解，降低开发者心智负担。但一旦连接数上来，goroutine的数量就会线性增加。当面对海量连接的场景，这种模型将力不从心：系统中将存在大量goroutine，goroutine调度和切换的开销过多。

那么面对海量连接场景，应该如何解决呢？业界成熟方案：使用I/O多路复用模型。了解Go net包实现的朋友想必都知晓Go在运行时底层使用的也是I/O多路复用，其实现为runtime中的netpoll。goroutine层面获得的net.Conn(无论是Accept的，还是Dial得到的)都展现出“阻塞”的特征，但这些net.Conn底层实现的fd(文件描述符)在netpoll中都是non-blocking(非阻塞)的，Go运行时负责调用epoll等多路复用机制监视这些fd是否可读或可写，并适时唤醒goroutine继续网络I/O操作，这种方式减少了系统调用，也减少了运行Goroutine的M(操作系统线程)因系统调用陷入内核态等待的频率以及因阻塞失去M而不得不去创建新线程的数量。

那么在用户层面建立自己的I/O多路复用的不足在哪里呢？复杂，不好写，不好理解。但似乎也没有其他更好的办法。除非换语言，否则就得硬着头皮上^_^。好在，Go社区已经有几个不错的Go用户层面非阻塞I/O多路复用的开发框架库可供选择，比如：evio、gnet、easygo等。我们选择gnet。但注意：选择不代表推荐，这里仅是来做这个实践而已，是否使用gnet开发上生产的程序，需要你自己评估确定。

1. 基于gnet开发TCP流协议解析程序

用框架的一个门槛就是你要去学习框架本身。好在gnet提供了几个很典型的examples，我们可以基于其中的custom_codec来快速开发我们的TCP流协议解析程序。

下面是基于gnet框架实现custom codec的一个关键循环，了解这个循环，我们就知道在什么位置调用Frame编解码以及packet编解码了，这样决定了后续demo程序的结构：

上面图中右边虚框中的frame编解码、packet编解码以及React是用户需要自己实现的，gnet框架的eventloop.loopRead方法会循环调用frame编解码和React以实现TCP流的处理以及响应的返回。有了这样一张“地图”，我们就可以明确demo程序中各个包的大致位置了。

我们的demo改自gnet的例子custom_codec，其main包结构来自于custom_codec：

// github.com/bigwhite/experiments/tree/master/tcp-stream-proto/demo4/cmd/server/main.go

type customCodecServer struct {
    *gnet.EventServer
    addr       string
    multicore  bool
    async      bool
    codec      gnet.ICodec
    workerPool *goroutine.Pool
}

func (cs *customCodecServer) OnInitComplete(srv gnet.Server) (action gnet.Action) {
    log.Printf("custom codec server is listening on %s (multi-cores: %t, loops: %d)\n",
        srv.Addr.String(), srv.Multicore, srv.NumEventLoop)
    return
}

func customCodecServe(addr string, multicore, async bool, codec gnet.ICodec) {
    var err error
    codec = frame.Frame{}
    cs := &customCodecServer{addr: addr, multicore: multicore, async: async, codec: codec, workerPool: goroutine.Default()}
    err = gnet.Serve(cs, addr, gnet.WithMulticore(multicore), gnet.WithTCPKeepAlive(time.Minute*5), gnet.WithCodec(codec))
    if err != nil {
        panic(err)
    }
}

func main() {
    var port int
    var multicore bool

    // Example command: go run server.go --port 8888 --multicore=true
    flag.IntVar(&port, "port", 8888, "server port")
    flag.BoolVar(&multicore, "multicore", true, "multicore")
    flag.Parse()
    addr := fmt.Sprintf("tcp://:%d", port)
    customCodecServe(addr, multicore, false, nil)
}

针对上面代码，有两点要注意：

customCodecServe的第三个参数我们传入了false，即我们选择同步回复应答，而不是异步回复。
我们将自定义的frame编解码器(实现了gnet.ICodec接口)实例传给了customCodecServer实例，这样后续gnet loopRead调用的就是我们自定义的frame编解码器了。

按上面流程图的顺序，gnet从conn读取的字节流将传递给我们的frame解码器，下面我们看看基于gnet的Frame解码器的实现(我们的自定义协议定义可以参考《Go经典阻塞式TCP协议流解析的实践》一文)：

// github.com/bigwhite/experiments/tree/master/tcp-stream-proto/demo4/pkg/frame/frame.go

type Frame []byte

func (cc Frame) Decode(c gnet.Conn) ([]byte, error) {
    // read length
    var frameLength uint32
    if n, header := c.ReadN(4); n == 4 {
        byteBuffer := bytes.NewBuffer(header)
        _ = binary.Read(byteBuffer, binary.BigEndian, &frameLength)

        if frameLength > 100 {
            c.ResetBuffer()
            return nil, errors.New("length value is wrong")
        }

        if n, wholeFrame := c.ReadN(int(frameLength)); n == int(frameLength) {
            c.ShiftN(int(frameLength)) // shift frame length
            return wholeFrame[4:], nil // return frame payload
        } else {
            return nil, errors.New("not enough frame payload data")
        }
    }
    return nil, errors.New("not enough frame length data")
}

上面Frame的Decode实现既负责frame解码，同时也会对frame的当前数据完整性进行校验，如果一个完整的frame尚未就绪，Decode会返回错误，之后gnet还会在连接(conn)可读时再次调用该Decode函数。这里实现的关键就是gnet.Conn.ReadN这个方法，这个方法本质上是一个Peek操作(gnet称之为lazyRead)，即只预览数据, 不挪动数据流中的“读指针”的位置。frame未完全就绪时，gnet在底层会使用RingBuffer存放已经到位的frame的部分数据。如果frame所有数据都就绪了，那么Decode会调用gnet.Conn.ShiftN方法来挪动底层RingBuffer的“读指针”的位置，表明这段数据已经被上层读取了。

如果预读取到的frame长度过长（这里代码中的100是一个魔数，仅做demo演示之用，你可以根据实际情况使用frame可能的最大值），则会清空当前缓存并返回错误。(但gnet并没有因此而断开与客户端的连接，这块儿gnet的机制是否合理还有待商榷。)

如果解码顺利，根据我们自定义的协议spec，我们会将frame的payload返回，即从frame的第五个字节开始返回。

从上图看到，frame Decode返回的payload将作为输入数据传给eventHandler.React方法，这个方法也是我们自己实现的：

// github.com/bigwhite/experiments/tree/master/tcp-stream-proto/demo4/cmd/server/main.go

func (cs *customCodecServer) React(framePayload []byte, c gnet.Conn) (out []byte, action gnet.Action) {
    var p packet.Packet
    var ackFramePayload []byte
    p, err := packet.Decode(framePayload)
    if err != nil {
        fmt.Println("react: packet decode error:", err)
        action = gnet.Close // close the connection
        return
    }

    switch p.(type) {
    case *packet.Submit:
        submit := p.(*packet.Submit)
        fmt.Printf("recv submit: id = %s, payload=%s\n", submit.ID, string(submit.Payload))
        submitAck := &packet.SubmitAck{
            ID:     submit.ID,
            Result: 0,
        }
        ackFramePayload, err = packet.Encode(submitAck)
        if err != nil {
            fmt.Println("handleConn: packet encode error:", err)
            action = gnet.Close // close the connection
            return
        }
        out = []byte(ackFramePayload)
        return
    default:
        return nil, gnet.Close // close the connection
    }
}

在React中，我们利用packet包对传入的frame payload进行Decode并处理得到的Packet，处理后将packet响应进行编码(encode)，编码后得到的字节序列（ackFramePayload)将作为React的第一个返回值out返回。

frame会对React返回的ackFramePayload进行Encode，编码后的字节序列将被gnet写入outbound的tcp流中去：

// github.com/bigwhite/experiments/tree/master/tcp-stream-proto/demo4/pkg/frame/frame.go

func (cc Frame) Encode(c gnet.Conn, framePayload []byte) ([]byte, error) {
    result := make([]byte, 0)

    buffer := bytes.NewBuffer(result)

    // encode frame length(4+ framePayload length)
    length := uint32(4 + len([]byte(framePayload)))
    if err := binary.Write(buffer, binary.BigEndian, length); err != nil {
        s := fmt.Sprintf("Pack length error , %v", err)
        return nil, errors.New(s)
    }

    // encode frame payload
    n, err := buffer.Write(framePayload)
    if err != nil {
        s := fmt.Sprintf("Pack frame payload error , %v", err)
        return nil, errors.New(s)
    }

    if n != len(framePayload) {
        s := fmt.Sprintf("Pack frame payload length error , %v", err)
        return nil, errors.New(s)
    }

    return buffer.Bytes(), nil
}

这样一个loopRead循环就完成了。我们可以使用《Go经典阻塞式TCP协议流解析的实践》一文中的client对该程序进行测试：

// demo2的client
$./client
2021/07/25 16:35:34 dial ok
send submit id = 00000001, payload=full-bluestreak-207e
the result of submit ack[00000001] is 0
send submit id = 00000002, payload=cosmic-spider-ham-2985
the result of submit ack[00000002] is 0
send submit id = 00000003, payload=true-forge-3552
the result of submit ack[00000003] is 0

// demo4的server
$./server
2021/07/25 16:35:31 custom codec server is listening on :8888 (multi-cores: true, loops:  recv submit: id = 00000001, payload=full-bluestreak-207e
recv submit: id = 00000002, payload=cosmic-spider-ham-2985
recv submit: id = 00000003, payload=true-forge-3552

2. 压测对比

gnet针对内存分配、缓存重用等做了很多优化，我们来将其与阻塞I/O模型程序在性能上做一下简单比较(由于资源有限，我们这里的压测也和上一文中一样，采用100个client连接尽力(best effort)发送，而不是海量连接)。

下面是demo1(阻塞I/O模型未优化）、demo3(阻塞I/O模型优化后)以及demo4(io多路复用模型）的性能对比：

粗略来看，采用gnet I/O多路复用模型的程序(demo4)在性能上平均比阻塞I/O模型优化后的程序(demo3)高出15%~20%。

不仅如此，通过dstat采集的系统监控数据也表明跑demo4时，cpu系统时间(sys)占用也比demo3少了5个点左右：

跑demo3时的dstat -tcdngym输出:

----system---- ----total-cpu-usage---- -dsk/total- -net/total- ---paging-- ---system-- ------memory-usage-----
     time     |usr sys idl wai hiq siq| read  writ| recv  send|  in   out | int   csw | used  buff  cach  free
23-07 17:03:17|  2   1  97   0   0   0|3458B   19k|   0     0 |   0     0 | 535  2475 |1921M  225M 5354M 8386M
23-07 17:03:18| 40  45   5   0   0  11|   0     0 |  66B   54B|   0     0 |  11k   15k|1922M  225M 5354M 8384M
23-07 17:03:19| 39  46   6   0   0   9|   0     0 |  66B 1158B|   0     0 |  12k   18k|1922M  225M 5354M 8384M
23-07 17:03:20| 35  48   7   0   0  11|   0     0 |  66B  462B|   0     0 |  12k   22k|1922M  225M 5354M 8385M
23-07 17:03:21| 39  44   7   0   0  10|   0    12k|  66B  462B|   0     0 |  11k   16k|1922M  225M 5354M 8385M
23-07 17:03:22| 38  45   6   0   0  10|   0     0 |  66B  102B|   0     0 |  11k   16k|1923M  225M 5354M 8384M
23-07 17:03:23| 38  45   7   0   0  10|   0     0 |  66B  470B|   0     0 |  12k   20k|1923M  225M 5354M 8384M
23-07 17:03:24| 39  46   6   0   0   9|   0     0 |  66B  462B|   0     0 |  11k   19k|1923M  225M 5354M 8384M

跑demo4时的dstat -tcdngym输出：

----system---- ----total-cpu-usage---- -dsk/total- -net/total- ---paging-- ---system-- ------memory-usage-----
     time     |usr sys idl wai hiq siq| read  writ| recv  send|  in   out | int   csw | used  buff  cach  free
24-07 20:28:38| 43  42   7   0   0   8|   0    20k|1050B   14k|   0     0 |  11k   18k|1954M  234M 5959M 7738M
24-07 20:28:39| 44  41   9   0   0   7|   0    16k| 396B 7626B|   0     0 |  11k   17k|1954M  234M 5959M 7739M
24-07 20:28:40| 43  42   6   0   0   8|   0     0 | 132B 7044B|   0     0 |  11k   16k|1954M  234M 5959M 7738M
24-07 20:28:41| 42  42   8   0   0   8|   0     0 | 630B   12k|   0     0 |  12k   20k|1955M  234M 5959M 7738M
24-07 20:28:42| 45  41   7   0   0   7|   0     0 | 726B 9980B|   0     0 |  11k   16k|1955M  234M 5959M 7738M

2. 异步回应答

在上面的例子中，我们采用的是gnet同步回应答的方式，gnet还支持异步回应答的方式，即将React中得到的ackFramePayload提交给gnet创建的一个goroutine Worker池，由worker池中的某个空闲goroutine在后续将ackFramePayload编码为一个完整的ackFrame后返回给client端。

要支持异步回应答，我们需要对demo4做几处修改（见demo5），主要修改点都在cmd/server/main.go中。

第一处：main函数调用customCodecServe时，将第三个参数async设置为true：

// github.com/bigwhite/experiments/tree/master/tcp-stream-proto/demo5/cmd/server/main.go

func main() {
    ... ...
    customCodecServe(addr, multicore, true, nil)
}

第二处：在customCodecServer的React方法中，我们得到编码后的ackFramePayload后，不要立即将其赋值给out并返回，而是判断是否要异步返回应答。如果异步返回应答，则将ackFramePayload提交给workerpool，workerPool后续会分配goroutine，并通过gnet.Conn的AsyncWrite将应答写回client。如果非异步，在将ackFramePayload赋值给out并返回。

// github.com/bigwhite/experiments/tree/master/tcp-stream-proto/demo5/cmd/server/main.go

func (cs *customCodecServer) React(framePayload []byte, c gnet.Conn) (out []byte, action gnet.Action) {
    ... ...
    switch p.(type) {
    case *packet.Submit:
        submit := p.(*packet.Submit)
        fmt.Printf("recv submit: id = %s, payload=%s\n", submit.ID, string(submit.Payload))
        submitAck := &packet.SubmitAck{
            ID:     submit.ID,
            Result: 0,
        }
        ackFramePayload, err = packet.Encode(submitAck)
        if err != nil {
            fmt.Println("handleConn: packet encode error:", err)
            action = gnet.Close // close the connection
            return
        }
    default:
        return nil, gnet.Close // close the connection
    }

    if cs.async {
        data := append([]byte{}, ackFramePayload...)
        _ = cs.workerPool.Submit(func() {
            fmt.Println("handleConn: async write ackFramePayload")
            c.AsyncWrite(data)
        })
        return
    }
    out = ackFramePayload
    return
}

除此之外，其他包的代码不变。我们依然还做个压测，看看异步回应答的demo5性能究竟如何！

从上图来看，在这个场景下通过异步回应答的方式，性能反而下降很多，甚至还不如阻塞式I/O模型的程序。对此没有做深究，但猜测可能是应答过多且同时集中回复时workerpool创建了很多goroutine，不仅没有起到池化的作用，还带来的goroutine创建和调度的开销。

3. 小结

在本文中，我们将阻塞式I/O模型换成了I/O多路复用模型，并基于gnet框架重新实现了自定义TCP流协议的解析程序。在同步回应答的策略下，基于gnet开发TCP流协议解析程序相比于阻塞I/O模型程序的性能有一定提升。

本文涉及的所有代码可以从这里下载：https://github.com/bigwhite/experiments/tree/master/tcp-stream-proto

“Gopher部落”知识星球正式转正（从试运营星球变成了正式星球）！“gopher部落”旨在打造一个精品Go学习和进阶社群！高品质首发Go技术文章，“三天”首发阅读权，每年两期Go语言发展现状分析，每天提前1小时阅读到新鲜的Gopher日报，网课、技术专栏、图书内容前瞻，六小时内必答保证等满足你关于Go语言生态的所有需求！部落目前虽小，但持续力很强。在2021年上半年，部落将策划两个专题系列分享，并且是部落独享哦：

Go技术书籍的书摘和读书体会系列
Go与eBPF系列

欢迎大家加入！

Go技术专栏“改善Go语⾔编程质量的50个有效实践”正在慕课网火热热销中！本专栏主要满足广大gopher关于Go语言进阶的需求，围绕如何写出地道且高质量Go代码给出50条有效实践建议，上线后收到一致好评！欢迎大家订
阅！

img{512x368}

我的网课“Kubernetes实战：高可用集群搭建、配置、运维与应用”在慕课网热卖中，欢迎小伙伴们订阅学习！

img{512x368}

我爱发短信：企业级短信平台定制开发专家 https://tonybai.com/。smspush : 可部署在企业内部的定制化短信平台，三网覆盖，不惧大并发接入，可定制扩展；短信内容你来定，不再受约束, 接口丰富，支持长短信，签名可选。2020年4月8日，中国三大电信运营商联合发布《5G消息白皮书》，51短信平台也会全新升级到“51商用消息平台”，全面支持5G RCS消息。

著名云主机服务厂商DigitalOcean发布最新的主机计划，入门级Droplet配置升级为：1 core CPU、1G内存、25G高速SSD，价格5$/月。有使用DigitalOcean需求的朋友，可以打开这个链接地址：https://m.do.co/c/bff6eed92687 开启你的DO主机之路。

Gopher Daily(Gopher每日新闻)归档仓库 – https://github.com/bigwhite/gopherdaily

我的联系方式：

微博：https://weibo.com/bigwhite20xx
微信公众号：iamtonybai
博客：tonybai.com
github: https://github.com/bigwhite
“Gopher部落”知识星球：https://public.zsxq.com/groups/51284458844544

微信赞赏：
img{512x368}

商务合作方式：撰稿、出书、培训、在线课程、合伙创业、咨询、广告合作。

Go经典阻塞式TCP协议流解析的实践

七月 28, 2021
0 条评论

本文永久链接 – https://tonybai.com/2021/07/28/classic-blocking-network-tcp-stream-protocol-parsing-practice-in-go

1. Go经典阻塞I/O的TCP网络编程模型

Go语言诞生十多年来取得了飞速发展，并得到了全世界开发者的广泛接纳和应用，其应用领域广泛，包括：Web服务、数据库、网络编程、系统编程、DevOps、安全检测与管控、数据科学以及人工智能等。下面是2020年Go官方开发者调查的部分结果：

图：2020年Go官方开发者调查之Go语言的应用领域(对比2019)

我们看到“Web编程”和“网络编程”分别位列第一名和第四名，这个应用领域数据分布与Go语言最初的面向大规模分布式网络服务的设计目标十分契合。网络通信这块是服务端程序必不可少也是至关重要的一部分。Go标准库的net包是在Go中进行网络编程的基础。即便您没有直接使用到net包中有关TCP Socket方面的函数/方法或接口，但net/http包想必大家总是用过的，http包实现的是HTTP这个应用层协议，其在传输层使用的依旧是TCP Socket。

Go是自带运行时的跨平台编程语言，由于Go运行时调度的需要，Go基于I/O多路复用机制(linux上使用epoll，macOS和freebsd上使用kqueue)设计和实现了一套适合自己的TCP Socket网络编程模型。并且，Go秉承了自己一贯的追求简单的设计哲学，Go向语言使用者暴露了简单的TCP Socket API接口，而将Go TCP socket网络编程的“复杂性”留给了自己并隐藏在Go运行时的实现中。这样，大多数情况下，Go开发者无需关心Socket是否是阻塞的，也无需亲自将Socket文件描述符的回调函数注册到类似epoll这样的系统调用中，而只需在每个连接对应的goroutine中以最简单最易用的“阻塞I/O模型”的方式进行Socket操作即可(像下图所示)，这种设计大大降低了网络应用开发人员的心智负担。

这是经典的Go tcp网络编程模型。由于TCP是全双工模型，每一端(peer)都可以单独在已经建立的连接上进行读写，因此在Go中，我们常常针对一个已建立的TCP连接建立两个goroutine，一个负责从连接上读取数据(如需响应(ack)，也可以由该read goroutine直接回复)，一个负责将新生成的业务数据写入连接。

以read goroutine为例，其典型的程序结构如下：

func handleConn(c net.Conn) {
    defer c.Close()
    for {
        // read from the connection c
        ... ...
        // write ack to the connection c
        ... ...
    }
}

func main() {
    l, err := net.Listen("tcp", ":8888")
    if err != nil {
        fmt.Println("listen error:", err)
        return
    }

    for {
        c, err := l.Accept()
        if err != nil {
            fmt.Println("accept error:", err)
            break
        }
        // start a new goroutine to handle
        // the new connection.
        go handleConn(c) // start a read goroutine
    }
}

从上面代码，我们看到，针对每一个向server建立成功的连接，程序都会启动一个reader goroutine负责从连接读取数据，并在处理后，返回(向连接写入)响应(ack)。这样的程序结构已经直白到无法再直白了，即便你是网络编程小白，看懂这样的程序想必也不会费多少脑细胞。

我们知道，TCP传输控制协议是一种面向连接的、可靠的、基于字节流的传输层通信协议，因此TCP socket编程多为流数据(streaming)处理。这种数据的特点是按序逐个字节传输，在传输层没有明显的数据边界(只有应用层能识别出协议数据的边界，这个依赖应用层协议的定义)。TCP发送端发送了1000个字节，TCP接收端就会接收到1000个字节。发送端可能通过一次发送操作就发送了这1000个字节，但接收端可能通过10次读取操作才读完这1000个字节，也就是说发送端的发送动作与接收端的接收动作并没有严格的一一对应关系。这与UDP协议基于数据报(diagram)形式的数据传输形式有本质差别(更多关于tcp与udp差别的内容可以详见《TCP/IP详解卷1：协议》一书)。

本文我们就来了解一下基于经典Go阻塞式网络I/O模型对基于TCP流的自定义协议进行解析的基本模式。

2. 自定义协议简述

为了便于后续内容展开，我们现在这里说明一下我们即将解析的自定义流协议。基于TCP的自定义应用层流协议有两种常见的定义模式：

二进制模式

采用长度字段分隔，常见的包括：mqtt(物联网最常用的应用层协议之一)、cmpp(中国移动互联网短信网关接口协议)等。

文本模式

采用特定分隔符分割和识别，常见的包括http等。

这里我们使用二进制模式来定义我们即将解析的应用层协议，下面是协议的定义：

这是一个请求应答协议，请求包和应答包的第一个字段都是包总长度，这也是在应用层用于“分割包”的最重要字段。第二个字段则是用于标识包类型，这里我们定义四种类型：

onst (
    CommandConn   = iota + 0x01 // 0x01，连接请求包
    CommandSubmit               // 0x02，消息发送请求包
)

const (
    CommandConnAck   = iota + 0x80 // 0x81，连接请求的响应包
    CommandSubmitAck               //0x82，消息发送请求的响应包
)

ID是每个连接上请求的消息流水，多用于请求发送方后续匹配响应包之用。请求包与响应包唯一的不同之处在于最后一个字段，请求包定义了有效载荷(payload)，而响应包则定义了请求包的响应状态字段(result)。

明确了应用层协议包的定义后，我们就来看看如何解析这样的一个流协议吧。

3. 建立Frame和Packet抽象

在真正开始编写代码前，我们先来针对上述应用层协议建立两个抽象概念：Frame和Packet。

首先，我们设定无论是从client到server，还是server到client，数据流都是由一个接一个Frame组成的，上述的协议就封装在这一个个的Frame中。我们可以通过特定的方法将Frame与Frame分割开来：

每个Frame由一个totalLength和frame payload构成，如下图左侧Frame结构所示：

这样，我们通过Frame header: totalLength即可将Frame之间隔离开来。我们将Frame payload定义为一个packet，每个Packet的结构如上图右侧所示。每个packet包含commandID、ID和payload(packet payload)字段。

这样我们就将上述的协议转换为由Frame和Packet两个抽象组成的TCP流了。

4. 阻塞式TCP流协议解析的基本程序结构

建立完抽象后，我们就要开始解析这个协议了！下图是该阻塞式TCP流协议解析的server流程图：

我们看到tcp流数据先后经由frame decode和packet decode后得到应用层所需的packet数据，应用层回复的响应则先后经过packet的encode与frame的encode后写入tcp响应流中。

下面我们就先来看看frame编解码的代码。我们首先定义frame编码器的接口类型：

// github.com/bigwhite/experiments/tree/master/tcp-stream-proto/demo1/pkg/frame/frame.go

type FramePayload []byte

type StreamFrameCodec interface {
    Encode(io.Writer, FramePayload) error   // data -> frame，并写入io.Writer
    Decode(io.Reader) (FramePayload, error) // 从io.Reader中提取frame payload，并返回给上层
}

我们将流数据的输入定义为io.Reader，将流数据输出定义为io.Writer。和上图中的设计意义，Decode方法返回framePayload，而Encode会将输入的framePayload编码为frame并写入outbound的tcp流。

一旦确定好接口方法集，我们就来给出一个StreamFrameCodec接口的实现：

// github.com/bigwhite/experiments/tree/master/tcp-stream-proto/demo1/pkg/frame/frame.go

type myFrameCodec struct{}

func NewMyFrameCodec() StreamFrameCodec {
    return &myFrameCodec{}
}

func (p *myFrameCodec) Encode(w io.Writer, framePayload FramePayload) error {
    var f = framePayload
    var totalLen int32 = int32(len(framePayload)) + 4

    err := binary.Write(w, binary.BigEndian, &totalLen)
    if err != nil {
        return err
    }

    // make sure all data will be written to outbound stream
    for {
        n, err := w.Write([]byte(f)) // write the frame payload to outbound stream
        if err != nil {
            return err
        }
        if n >= len(f) {
            break
        }
        if n < len(f) {
            f = f[n:]
        }
    }
    return nil
}

func (p *myFrameCodec) Decode(r io.Reader) (FramePayload, error) {
    var totalLen int32
    err := binary.Read(r, binary.BigEndian, &totalLen)
    if err != nil {
        return nil, err
    }

    buf := make([]byte, totalLen-4)
    _, err = io.ReadFull(r, buf)
    if err != nil {
        return nil, err
    }
    return FramePayload(buf), nil
}

在上面在这段实现中，有三点要注意：

网络字节序使用大端字节序(BigEndian)，因此无论是Encode还是Decode，我们都是用binary.BigEndian；
binary.Read或Write会根据参数的宽度读取或写入对应的字节个数的字节，这里totalLen使用int32，那么Read或Write只会操作流中的4个字节；
这里没有设置deadline，因此io.ReadFull一般会读满你所需的字节数，除非遇到EOF或ErrUnexpectedEOF。

接下来，我们再看看Packet的编解码。和Frame不同，Packet有多种类型(这里仅定义了Conn, submit，connack, submit ack)。因此我们首先抽象一下这些类型需要遵循的共同接口：

// github.com/bigwhite/experiments/tree/master/tcp-stream-proto/demo1/pkg/packet/packet.go

type Packet interface {
    Decode([]byte) error     // []byte -> struct
    Encode() ([]byte, error) //  struct -> []byte
}

其中Decode是将一段字节流数据解码为一个Packet类型，可能是conn，可能是submit等(根据解码出来的commandID判断)。而Encode则是将一个Packet类型编码为一段字节流数据。下面是submit和submitack类型的Packet接口实现：

// github.com/bigwhite/experiments/tree/master/tcp-stream-proto/demo1/pkg/packet/packet.go

type Submit struct {
    ID      string
    Payload []byte
}

func (s *Submit) Decode(pktBody []byte) error {
    s.ID = string(pktBody[:8])
    s.Payload = pktBody[8:]
    return nil
}

func (s *Submit) Encode() ([]byte, error) {
    return bytes.Join([][]byte{[]byte(s.ID[:8]), s.Payload}, nil), nil
}

type SubmitAck struct {
    ID     string
    Result uint8
}

func (s *SubmitAck) Decode(pktBody []byte) error {
    s.ID = string(pktBody[0:8])
    s.Result = uint8(pktBody[8])
    return nil
}

func (s *SubmitAck) Encode() ([]byte, error) {
    return bytes.Join([][]byte{[]byte(s.ID[:8]), []byte{s.Result}}, nil), nil
}

不过上述各种类型的编解码被调用的前提是明确数据流是什么类型的，因此我们需要在包级提供一个对外的函数Decode，该函数负责从字节流中解析出对应的类型(根据commandID)，并调用对应类型的Decode方法：

// github.com/bigwhite/experiments/tree/master/tcp-stream-proto/demo1/pkg/packet/packet.go
func Decode(packet []byte) (Packet, error) {
    commandID := packet[0]
    pktBody := packet[1:]

    switch commandID {
    case CommandConn:
        return nil, nil
    case CommandConnAck:
        return nil, nil
    case CommandSubmit:
        s := Submit{}
        err := s.Decode(pktBody)
        if err != nil {
            return nil, err
        }
        return &s, nil
    case CommandSubmitAck:
        s := SubmitAck{}
        err := s.Decode(pktBody)
        if err != nil {
            return nil, err
        }
        return &s, nil
    default:
        return nil, fmt.Errorf("unknown commandID [%d]", commandID)
    }
}

同样，我们也需要包级的Encode函数，根据传入的packet类型调用对应的Encode方法实现对象的编码：

// github.com/bigwhite/experiments/tree/master/tcp-stream-proto/demo1/pkg/packet/packet.go
func Encode(p Packet) ([]byte, error) {
    var commandID uint8
    var pktBody []byte
    var err error

    switch t := p.(type) {
    case *Submit:
        commandID = CommandSubmit
        pktBody, err = p.Encode()
        if err != nil {
            return nil, err
        }
    case *SubmitAck:
        commandID = CommandSubmitAck
        pktBody, err = p.Encode()
        if err != nil {
            return nil, err
        }
    default:
        return nil, fmt.Errorf("unknown type [%s]", t)
    }
    return bytes.Join([][]byte{[]byte{commandID}, pktBody}, nil), nil
}

好了，万事俱备只欠东风！下面我们就来编写程序结构，将tcp conn与Frame、Packet连接起来：

// github.com/bigwhite/experiments/tree/master/tcp-stream-proto/demo1/cmd/server/main.go

package main

import (
    "fmt"
    "net"

    "github.com/bigwhite/tcp-stream-proto/demo1/pkg/frame"
    "github.com/bigwhite/tcp-stream-proto/demo1/pkg/packet"
)

func handlePacket(framePayload []byte) (ackFramePayload []byte, err error) {
    var p packet.Packet
    p, err = packet.Decode(framePayload)
    if err != nil {
        fmt.Println("handleConn: packet decode error:", err)
        return
    }

    switch p.(type) {
    case *packet.Submit:
        submit := p.(*packet.Submit)
        fmt.Printf("recv submit: id = %s, payload=%s\n", submit.ID, string(submit.Payload))
        submitAck := &packet.SubmitAck{
            ID:     submit.ID,
            Result: 0,
        }
        ackFramePayload, err = packet.Encode(submitAck)
        if err != nil {
            fmt.Println("handleConn: packet encode error:", err)
            return nil, err
        }
        return ackFramePayload, nil
    default:
        return nil, fmt.Errorf("unknown packet type")
    }
}

func handleConn(c net.Conn) {
    defer c.Close()
    frameCodec := frame.NewMyFrameCodec()

    for {
        // read from the connection

        // decode the frame to get the payload
        // the payload is undecoded packet
        framePayload, err := frameCodec.Decode(c)
        if err != nil {
            fmt.Println("handleConn: frame decode error:", err)
            return
        }

        // do something with the packet
        ackFramePayload, err := handlePacket(framePayload)
        if err != nil {
            fmt.Println("handleConn: handle packet error:", err)
            return
        }

        // write ack frame to the connection
        err = frameCodec.Encode(c, ackFramePayload)
        if err != nil {
            fmt.Println("handleConn: frame encode error:", err)
            return
        }
    }
}

func main() {
    l, err := net.Listen("tcp", ":8888")
    if err != nil {
        fmt.Println("listen error:", err)
        return
    }

    for {
        c, err := l.Accept()
        if err != nil {
            fmt.Println("accept error:", err)
            break
        }
        // start a new goroutine to handle
        // the new connection.
        go handleConn(c)
    }
}

在上面这个程序中，main函数是标准的“one connection per goroutine”的结构，重点逻辑都在handleConn中。在handleConn中，我们看到十分清晰的代码结构：

read conn
    ->frame decode
        -> handle packet
            -> packet decode
            -> packet(ack) encode
    ->frame(ack) encode
write conn

到这里，一个经典阻塞式TCP流解析的demo就完成了(你可以将demo中提供的client和server run起来验证一下)。

5. 可能的优化点

在上面的demo1中，我们直接将net.Conn实例传给frame.Decode作为io.Reader参数的实参，这样我们每次调用Read方法都是直接从Conn中读取数据。不过Go runtime使用net poller将net.Conn.Read转换为io多路复用的等待，避免了每次从net.Conn直接读取都转换为一次系统调用。但即便如此，也可能会多一次goroutine的上下文切换(在数据尚未ready的情况下)。虽然goroutine的上下文切换代价相较于线程切换要小许多，但毕竟这种切换并不是免费的，我们要减少这种切换。我们可以通过缓存读的方式来减少net.Conn.Read真实调用的频率。我们可以像下面这样改造demo1的例子：

// github.com/bigwhite/experiments/tree/master/tcp-stream-proto/demo2/cmd/server/main.go

func handleConn(c net.Conn) {
    defer c.Close()
    frameCodec := frame.NewMyFrameCodec()
    rbuf := bufio.NewReader(c) // 为io增加缓存

    for {
        // read from the connection

        // decode the frame to get the payload
        // the payload is undecoded packet
        framePayload, err := frameCodec.Decode(rbuf) // 使用bufio，减少直接read conn.Conn的次数
        if err != nil {
            fmt.Println("handleConn: frame decode error:", err)
            return
        }
        ... ...
    }
    ... ...
}

bufio内部每次从net.Conn尝试读取其内部缓存(buf)大小的数据，而不是用户传入的希望读取的数据大小。这些数据缓存在内存中，这样后续Read就可以直接从内存中得到数据，而不是每次都从net.Conn读取，从而降低goroutine上下文切换的频率。

除此之外，我们在frame包中的frame Decode实现如下：

// github.com/bigwhite/experiments/tree/master/tcp-stream-proto/demo2/pkg/frame/frame.go

func (p *myFrameCodec) Decode(r io.Reader) (FramePayload, error) {
    var totalLen int32
    err := binary.Read(r, binary.BigEndian, &totalLen)
    if err != nil {
        return nil, err
    }

    buf := make([]byte, totalLen-4)
    _, err = io.ReadFull(r, buf)
    if err != nil {
        return nil, err
    }
    return FramePayload(buf), nil
}

我们看到每次调用这个方法都会分配一个buf，并且buf是不定长的，这些在程序关键路径上的堆内存对象分配会给GC带来压力，我们要尽量避免或减小其频度，一个可行的办法是尽量重用对象，在Go中一提到重用内存对象，我们就想到了sync.Pool，但这里还有一个问题，那就是“不定长”，这给sync.Pool的使用增加了难度。

mcache是字节技术团队开源的多级sync.Pool包，它可以根据你所要分配的对象大小选择不同的sync.Pool池，有些类似tcmalloc的多级(class)内存对象管理，与Go runtime的mcache也是类似的，mcache一共分为46个等级，每个等级一个sync.Pool：

// github.com/bytedance/gopkg/tree/master/lang/mcache/mcache.go
const maxSize = 46

// index contains []byte which cap is 1<<index
var caches [maxSize]sync.Pool

我们可以从mcache中分配内存来换掉每次都申请一个[]byte的动作以达到内存对象重用，降低GC压力的目的：

// github.com/bigwhite/experiments/tree/master/tcp-stream-proto/demo3/pkg/frame/frame.go

func (p *myFrameCodec) Decode(r io.Reader) (FramePayload, error) {
    var totalLen int32
    err := binary.Read(r, binary.BigEndian, &totalLen)
    if err != nil {
        return nil, err
    }

    buf := mcache.Malloc(int(totalLen - 4))  // 这里我们重用mcache中的内存对象
    _, err = io.ReadFull(r, buf)
    if err != nil {
        return nil, err
    }
    return FramePayload(buf), nil
}

有了mcache.Malloc，我们就需要在特定位置调用mcache.Free归还内存对象，而packet中的Decode就是最好的位置：

// github.com/bigwhite/experiments/tree/master/tcp-stream-proto/demo3/pkg/packet/packet.go

func Decode(packet []byte) (Packet, error) {
    defer mcache.Free(packet) // 在decode结束后，释放对象回mcache
    commandID := packet[0]
    pktBody := packet[1:]
    ... ...
}

上面是两个在不动用pprof这样的工具的前提下就能识别出的较为明显的可优化的点，可优化的点可能还有很多，这里不一一列举了。

6. 简单的压力测试

既然给出了优化的点，我们就来粗略压测一下优化前和优化后的程序。我们为两个版本程序添加上基于标准库expvar的计数器(以优化前的demo1为例)：

// github.com/bigwhite/experiments/tree/master/tcp-stream-proto/demo1-with-metrics/cmd/server/main.go

func handleConn(c net.Conn) {
    defer c.Close()
    frameCodec := frame.NewMyFrameCodec()

    for {
        // read from the connection
        ... ...
        // write ack frame to the connection
        err = frameCodec.Encode(c, ackFramePayload)
        if err != nil {
            fmt.Println("handleConn: frame encode error:", err)
            return
        }
        monitor.SubmitInTotal.Add(1) // 每处理完一条消息，计数器+1
    }
}

在monitor包中，我们每秒计算一下处理性能：

// github.com/bigwhite/experiments/tree/master/tcp-stream-proto/demo1-with-metrics/pkg/monitor/monitor.go
func init() {
    // register statistics index
    SubmitInTotal = expvar.NewInt("submitInTotal")
    submitInRate = expvar.NewInt("submitInRate")

    go func() {
        var lastSubmitInTotal int64

        ticker := time.NewTicker(time.Second)
        defer ticker.Stop()
        for {
            select {
            case <-ticker.C:
                newSubmitInTotal := SubmitInTotal.Value()
                submitInRate.Set(newSubmitInTotal - lastSubmitInTotal) // 两秒处理的消息量之差作为处理速度
                lastSubmitInTotal = newSubmitInTotal
            }
        }
    }()
}

有了基于expvar的计数器，我们就可以通过带有导出csv功能的expvarmon工具获取程序每秒的处理性能了（压测客户端可以使用demo1-with-metrics的client)。下面的性能对比图是在一个4核8g的云主机上获得的（条件有限，压测client与server放在一台机器上了，必然相互干扰）：

我们看到，优化后的程序从趋势上看略微好于优化前的(虽然不是很稳定)。

如果你觉得采集瞬时值太够专业^_^，也可以在被测程序上添加基于go-metrics的metric，这个作业就留给大家了:)

7. 小结

在本文中，我们简单说明了Go经典阻塞I/O的TCP网络编程模型，这种模型最大的好处就是简单，降低开发人员在处理网络I/O时的心智负担，将更多关注集中在业务层面。文中基于这种模型，给出了一个自定义流协议的解析实现框架，并说明了一些可优化的点。在非超大连接数量的场景下，这类模型会有不错性能和开发效率。一旦连接数量猛增，相应的处理这些连接的goroutine数量就会线性增加，Goroutine调度的开销就会显著增加，这个时候我们就要考虑是否使用其他模型应对了，这个我们在后续篇章再说。

本文涉及的所有代码可以从这里下载：https://github.com/bigwhite/experiments/tree/master/tcp-stream-proto

Go技术书籍的书摘和读书体会系列
Go与eBPF系列

欢迎大家加入！

img{512x368}

我的网课“Kubernetes实战：高可用集群搭建、配置、运维与应用”在慕课网热卖中，欢迎小伙伴们订阅学习！

img{512x368}

Gopher Daily(Gopher每日新闻)归档仓库 – https://github.com/bigwhite/gopherdaily

我的联系方式：

微博：https://weibo.com/bigwhite20xx
微信公众号：iamtonybai
博客：tonybai.com
github: https://github.com/bigwhite
“Gopher部落”知识星球：https://public.zsxq.com/groups/51284458844544

微信赞赏：
img{512x368}

商务合作方式：撰稿、出书、培训、在线课程、合伙创业、咨询、广告合作。

Go基于I/O多路复用的TCP协议流解析实践

1. 基于gnet开发TCP流协议解析程序

2. 压测对比

2. 异步回应答

3. 小结

Go经典阻塞式TCP协议流解析的实践

1. Go经典阻塞I/O的TCP网络编程模型

2. 自定义协议简述

3. 建立Frame和Packet抽象

4. 阻塞式TCP流协议解析的基本程序结构

5. 可能的优化点

6. 简单的压力测试

7. 小结

欢迎使用邮件订阅我的博客

文章

评论

分类

归档

链接

开源项目

翻译项目

Go基于I/O多路复用的TCP协议流解析实践

1. 基于gnet开发TCP流协议解析程序

2. 压测对比

2. 异步回应答

3. 小结

Go经典阻塞式TCP协议流解析的实践

1. Go经典阻塞I/O的TCP网络编程模型

2. 自定义协议简述

3. 建立Frame和Packet抽象

4. 阻塞式TCP流协议解析的基本程序结构

5. 可能的优化点

6. 简单的压力测试

7. 小结

欢迎使用邮件订阅我的博客

文章

评论

分类

标签

归档

链接

开源项目

翻译项目