golang高性能异步IO网络编程插件库gaio的使用

gaio是一个高性能的异步IO网络编程库，专为Go语言设计，特别适合需要处理大量并发连接的场景。

简介

在典型的Go网络程序中，我们通常使用conn := lis.Accept()接受连接，然后启动一个goroutine使用go func(net.Conn)处理传入数据，接着分配缓冲区buf := make([]byte, 4096)并使用conn.Read(buf)等待数据。

对于管理超过10,000个连接且频繁发送短消息(如<512字节)的服务器，上下文切换的成本会显著高于接收消息的成本。gaio通过边缘触发的I/O多路复用技术，消除了每个连接一个goroutine的需求，节省了每个goroutine通常使用的2KB(读)+2KB(写)堆栈空间。

工作原理

gaio实现了proactor模式，有效解决了内存限制和性能目标。它使用dup函数复制net.Conn的文件描述符：

NAME
       dup, dup2, dup3 - duplicate a file descriptor

DESCRIPTION
       The dup() system call allocates a new file descriptor that refers to the same open file description as the descriptor oldfd.

特性

• 高性能：在高频交易环境中测试，单个HVM服务器可处理30K-40K RPS
• 可扩展性：设计用于超过C10K并发连接
• 灵活缓冲：使用Read(ctx, conn, buffer)时，nil缓冲区可使用内部交换缓冲区
• 非侵入式集成：兼容net.Listener和net.Conn
• 高效上下文切换：最小化小消息的上下文切换成本
• 轻量级：约1,000行代码，易于调试
• 跨平台支持：兼容Linux和BSD

使用示例

基本回显服务器示例

package main

import (
        "log"
        "net"

        "github.com/xtaci/gaio"
)

// 这个goroutine将等待所有IO事件，并以异步方式回显收到的所有内容
func echoServer(w *gaio.Watcher) {
        for {
                // 循环等待任何IO事件
                results, err := w.WaitIO()
                if err != nil {
                        log.Println(err)
                        return
                }

                for _, res := range results {
                        switch res.Operation {
                        case gaio.OpRead: // 读取完成事件
                                if res.Error == nil {
                                        // 回显所有内容，在写入完成前不会再次开始读取
                                        // 提交异步写入请求
                                        w.Write(nil, res.Conn, res.Buffer[:res.Size])
                                }
                        case gaio.OpWrite: // 写入完成事件
                                if res.Error == nil {
                                        // 写入已完成，再次开始在此连接上读取
                                        w.Read(nil, res.Conn, res.Buffer[:cap(res.Buffer)])
                                }
                        }
                }
        }
}

func main() {
        w, err := gaio.NewWatcher()
        if err != nil {
              log.Fatal(err)
        }
        defer w.Close()
	
        go echoServer(w)

        ln, err := net.Listen("tcp", "localhost:0")
        if err != nil {
                log.Fatal(err)
        }
        log.Println("echo server listening on", ln.Addr())

        for {
                conn, err := ln.Accept()
                if err != nil {
                        log.Println(err)
                        return
                }
                log.Println("new client", conn.RemoteAddr())

                // 提交第一个异步读取IO请求
                err = w.Read(nil, conn, make([]byte, 128))
                if err != nil {
                        log.Println(err)
                        return
                }
        }
}

推送服务器示例

package main

import (
        "fmt"
        "log"
        "net"
        "time"

        "github.com/xtaci/gaio"
)

func main() {
        ln, err := net.Listen("tcp", "localhost:0")
        if err != nil {
                log.Fatal(err)
        }

        log.Println("pushing server listening on", ln.Addr(), ", use telnet to receive push")

        // 创建watcher
        w, err := gaio.NewWatcher()
        if err != nil {
                log.Fatal(err)
        }

        // 通道
        ticker := time.NewTicker(time.Second)
        chConn := make(chan net.Conn)
        chIO := make(chan gaio.OpResult)

        // watcher.WaitIO goroutine
        go func() {
                for {
                        results, err := w.WaitIO()
                        if err != nil {
                                log.Println(err)
                                return
                        }

                        for _, res := range results {
                                chIO <- res
                        }
                }
        }()

        // 主逻辑循环
        go func() {
                var conns []net.Conn
                for {
                        select {
                        case res := <-chIO: // 从watcher接收IO事件
                                if res.Error != nil {
                                        continue
                                }
                                conns = append(conns, res.Conn)
                        case t := <-ticker.C: // 接收定时器事件
                                push := []byte(fmt.Sprintf("%s\n", t))
                                // 所有连接将收到相同的'push'内容
                                for _, conn := range conns {
                                        w.Write(nil, conn, push)
                                }
                                conns = nil
                        case conn := <-chConn: // 接收新连接事件
                                conns = append(conns, conn)
                        }
                }
        }()

        // 此循环持续接受连接并发送到主循环
        for {
                conn, err := ln.Accept()
                if err != nil {
                        log.Println(err)
                        return
                }
                chConn <- conn
        }
}

性能基准

使用注意事项

1. 在MacOS上运行基准测试前，需要增加最大打开文件限制：

sysctl -w kern.ipc.somaxconn=4096
sysctl -w kern.maxfiles=100000
sysctl -w kern.maxfilesperproc=100000
sysctl -w net.inet.ip.portrange.first=1024
sysctl -w net.inet.ip.portrange.last=65535

ulimit -S -n 65536

1. 如果遇到编译错误，请确保已安装gcc/clang：

# github.com/xtaci/gaio [github.com/xtaci/gaio.test]
./aio_linux.go:155:7: undefined: setAffinity
./watcher.go:588:4: undefined: setAffinity
FAIL	github.com/xtaci/gaio [build failed]
FAIL

gaio是一个稳定可靠的异步IO库，特别适合需要高性能网络编程的场景。