7.6 网络编程高级

高性能网络服务器设计

并发模型

package main

import (
    "io"
    "log"
    "net"
    "sync"
)

// 模型 1: 每连接一个 goroutine
func serveOneConnPerGoroutine(ln net.Listener) {
    for {
        conn, err := ln.Accept()
        if err != nil {
            log.Println(err)
            continue
        }
        go handleConn(conn)
    }
}

func handleConn(conn net.Conn) {
    defer conn.Close()
    io.Copy(conn, conn)  // echo
}

// 模型 2: Goroutine 池
type Server struct {
    pool chan struct{}  // 信号量控制并发数
    wg   sync.WaitGroup
}

func NewServer(maxConn int) *Server {
    return &Server{
        pool: make(chan struct{}, maxConn),
    }
}

func (s *Server) Serve(ln net.Listener) {
    for {
        conn, err := ln.Accept()
        if err != nil {
            log.Println(err)
            continue
        }

        s.pool <- struct{}{}  // 获取信号量
        s.wg.Add(1)

        go func() {
            defer s.wg.Done()
            defer func() { <-s.pool }()  // 释放信号量
            defer conn.Close()

            handleConn(conn)
        }()
    }
}

func (s *Server) Stop() {
    s.wg.Wait()
}

// 模型 3: Worker 池
type WorkerPool struct {
    workers   int
    jobQueue  chan net.Conn
    wg        sync.WaitGroup
}

func NewWorkerPool(workers int) *WorkerPool {
    return &WorkerPool{
        workers:  workers,
        jobQueue: make(chan net.Conn, 1024),
    }
}

func (wp *WorkerPool) Start() {
    for i := 0; i < wp.workers; i++ {
        wp.wg.Add(1)
        go func(id int) {
            defer wp.wg.Done()
            for conn := range wp.jobQueue {
                handleConn(conn)
                conn.Close()
            }
        }(i)
    }
}

func (wp *WorkerPool) AddJob(conn net.Conn) {
    wp.jobQueue <- conn
}

func (wp *WorkerPool) Stop() {
    close(wp.jobQueue)
    wp.wg.Wait()
}

---

epoll 与 kqueue

Linux epoll 示例 (使用 syscall)

package main

import (
    "syscall"
    "unsafe"
)

// epoll 封装
type Epoll struct {
    fd          int
    connections map[int]net.Conn
}

func NewEpoll() (*Epoll, error) {
    fd, err := syscall.EpollCreate1(0)
    if err != nil {
        return nil, err
    }
    return &Epoll{
        fd:          fd,
        connections: make(map[int]net.Conn),
    }, nil
}

func (e *Epoll) Add(conn net.Conn) error {
    fd := socketFD(conn)
    event := syscall.EpollEvent{
        Events: syscall.EPOLLIN | syscall.EPOLLOUT | syscall.EPOLLET,
        Fd:     int32(fd),
    }
    return syscall.EpollCtl(e.fd, syscall.EPOLL_CTL_ADD, fd, &event)
}

func (e *Epoll) Remove(conn net.Conn) error {
    fd := socketFD(conn)
    return syscall.EpollCtl(e.fd, syscall.EPOLL_CTL_DEL, fd, nil)
}

func (e *Epoll) Wait(events []syscall.EpollEvent, msec int) (int, error) {
    return syscall.EpollWait(e.fd, events, msec)
}

func (e *Epoll) Close() error {
    return syscall.Close(e.fd)
}

// 获取 socket fd (需要 unsafe 转换)
func socketFD(conn net.Conn) int {
    // 实际使用需要反射或类型断言获取底层 fd
    return 0
}

// 使用示例
func serveWithEpoll() {
    ln, _ := net.Listen("tcp", ":8080")
    epoll, _ := NewEpoll()
    defer epoll.Close()

    events := make([]syscall.EpollEvent, 1024)

    for {
        conn, _ := ln.Accept()
        epoll.Add(conn)

        n, _ := epoll.Wait(events, 1000)
        for i := 0; i < n; i++ {
            fd := int(events[i].Fd)
            // 处理事件...
        }
    }
}

使用 evio/evnet 库

package main

import (
    "log"

    "github.com/tidwall/evio"
)

// evio 高性能事件循环
func main() {
    var events evio.Events

    // 连接建立
    events.Opened = func(c evio.Conn) (out []byte, action evio.Action) {
        log.Printf("连接：%v", c.RemoteAddr())
        return nil, evio.None
    }

    // 连接关闭
    events.Closed = func(c evio.Conn, err error) (action evio.Action) {
        log.Printf("断开：%v", c.RemoteAddr())
        return evio.None
    }

    // 接收数据
    events.Data = func(c evio.Conn, in []byte) (out []byte, action evio.Action) {
        // echo 服务器
        out = in
        return
    }

    log.Fatal(events.Serve(":8080"))
}

// 配置选项
func serveWithOpts() {
    var events evio.Events

    events.Serve(evio.ServeOptions{
        Network:     "tcp",
        Addr:        ":8080",
        Multicore:   true,     // 多核
        ReusePort:   true,     // SO_REUSEPORT
        KeepAlive:   60,       // TCP keepalive
        NumLoops:    -1,       // 自动检测 CPU 核心
        LoadBalance: evio.LeastConnections,
    })
}

---

零拷贝技术

sendfile 系统调用

package main

import (
    "io"
    "net/http"
    "os"
)

// 传统方式 (多次拷贝)
func serveFileTraditional(w http.ResponseWriter, r *http.Request) {
    file, _ := os.Open("large.mp4")
    defer file.Close()
    io.Copy(w, file)  // 内核->用户->内核->网络
}

// 零拷贝方式
func serveFileZeroCopy(w http.ResponseWriter, r *http.Request) {
    file, _ := os.Open("large.mp4")
    defer file.Close()

    // 使用 http.ServeContent 内部使用 sendfile
    stat, _ := file.Stat()
    http.ServeContent(w, r, "large.mp4", stat.ModTime(), file)
}

// 直接使用 sendfile (Linux)
func sendfile(dst io.Writer, src *os.File) (int64, error) {
    // net/http 内部会检测并使用 sendfile
    // 也可以使用 syscall.Sendfile
    return io.Copy(dst, src)
}

splice 系统调用

package main

import (
    "syscall"
)

// splice 在内核空间移动数据
func spliceData(src, dst int) error {
    const maxSplice = 1 << 31

    for {
        n, err := syscall.Splice(src, nil, dst, nil, maxSplice,
            syscall.SPLICE_F_MOVE|syscall.SPLICE_F_NONBLOCK)
        if n == 0 {
            break
        }
        if err != nil {
            return err
        }
    }
    return nil
}

sync.Pool 对象复用

package main

import (
    "net"
    "sync"
)

// 缓冲区池
var bufferPool = sync.Pool{
    New: func() interface{} {
        return make([]byte, 4096)
    },
}

func handleConnWithPool(conn net.Conn) {
    defer conn.Close()

    buf := bufferPool.Get().([]byte)
    defer bufferPool.Put(buf)

    for {
        n, err := conn.Read(buf)
        if err != nil {
            break
        }
        conn.Write(buf[:n])
    }
}

// 大对象池
var largeBufferPool = sync.Pool{
    New: func() interface{} {
        return make([]byte, 65536)
    },
}

// 连接池
type ConnPool struct {
    pool sync.Pool
    new  func() net.Conn
}

func NewConnPool(newConn func() net.Conn) *ConnPool {
    return &ConnPool{
        new: newConn,
        pool: sync.Pool{
            New: func() interface{} {
                return newConn()
            },
        },
    }
}

func (p *ConnPool) Get() net.Conn {
    return p.pool.Get().(net.Conn)
}

func (p *ConnPool) Put(conn net.Conn) {
    // 重置连接状态
    // conn.SetDeadline(time.Time{})
    p.pool.Put(conn)
}

---

连接池设计

通用连接池

package pool

import (
    "errors"
    "net"
    "sync"
    "time"
)

type Conn interface {
    Close() error
    IsClosed() bool
}

type ConnFactory func() (Conn, error)

type Config struct {
    MaxOpen     int           // 最大连接数
    MaxIdle     int           // 最大空闲连接
    MaxLifetime time.Duration // 连接最大生命周期
    IdleTimeout time.Duration // 空闲超时
}

type ConnPool struct {
    factory    ConnFactory
    config     Config
    mu         sync.Mutex
    opens      int
    idle       []Conn
    closed     bool
    openerCh   chan struct{}
    cleanerCh  chan struct{}
}

func NewConnPool(factory ConnFactory, config Config) (*ConnPool, error) {
    if config.MaxOpen <= 0 {
        return nil, errors.New("MaxOpen must be positive")
    }
    if config.MaxIdle > config.MaxOpen {
        config.MaxIdle = config.MaxOpen
    }

    p := &ConnPool{
        factory:   factory,
        config:    config,
        idle:      make([]Conn, 0, config.MaxIdle),
        openerCh:  make(chan struct{}, config.MaxOpen),
        cleanerCh: make(chan struct{}),
    }

    // 启动连接清理协程
    go p.connectionCleaner()

    return p, nil
}

func (p *ConnPool) Get() (Conn, error) {
    p.mu.Lock()

    if p.closed {
        p.mu.Unlock()
        return nil, errors.New("pool is closed")
    }

    // 尝试从空闲列表获取
    for len(p.idle) > 0 {
        conn := p.idle[len(p.idle)-1]
        p.idle = p.idle[:len(p.idle)-1]

        if conn.IsClosed() {
            p.opens--
            continue
        }

        p.mu.Unlock()
        return conn, nil
    }

    // 检查是否可以创建新连接
    if p.opens >= p.config.MaxOpen {
        p.mu.Unlock()
        // 等待连接释放
        <-p.openerCh
        return p.Get()
    }

    p.opens++
    p.mu.Unlock()

    // 创建新连接
    return p.factory()
}

func (p *ConnPool) Put(conn Conn) {
    p.mu.Lock()

    if p.closed {
        p.mu.Unlock()
        conn.Close()
        return
    }

    // 放入空闲列表
    if len(p.idle) < p.config.MaxIdle {
        p.idle = append(p.idle, conn)
        select {
        case p.openerCh <- struct{}{}:
        default:
        }
    } else {
        p.opens--
    }

    p.mu.Unlock()
}

func (p *ConnPool) Close() error {
    p.mu.Lock()
    defer p.mu.Unlock()

    if p.closed {
        return nil
    }

    p.closed = true
    close(p.cleanerCh)

    for _, conn := range p.idle {
        conn.Close()
    }

    return nil
}

func (p *ConnPool) connectionCleaner() {
    ticker := time.NewTicker(time.Minute)
    defer ticker.Stop()

    for {
        select {
        case <-ticker.C:
            p.cleanIdle()
        case <-p.cleanerCh:
            return
        }
    }
}

func (p *ConnPool) cleanIdle() {
    p.mu.Lock()
    defer p.mu.Unlock()

    now := time.Now()
    valid := make([]Conn, 0, len(p.idle))

    for _, conn := range p.idle {
        // 检查空闲超时和生命周期
        if conn.IsClosed() {
            p.opens--
        } else {
            valid = append(valid, conn)
        }
    }

    p.idle = valid
}

---

TCP 调优

TCP 参数配置

package main

import (
    "net"
    "syscall"
    "time"
)

// 配置 TCP 参数
func dialWithOpts() {
    d := &net.Dialer{
        Timeout:   30 * time.Second,
        KeepAlive: 30 * time.Second,
        Control: func(network, address string, c syscall.RawConn) error {
            return c.Control(func(fd uintptr) {
                // 禁用 Nagle 算法 (低延迟)
                syscall.SetsockoptInt(int(fd), syscall.IPPROTO_TCP, syscall.TCP_NODELAY, 1)

                // 设置 TCP keepalive
                syscall.SetsockoptInt(int(fd), syscall.SOL_SOCKET, syscall.SO_KEEPALIVE, 1)

                // Linux 特有：设置 keepalive 参数
                // syscall.SetsockoptInt(int(fd), syscall.IPPROTO_TCP, syscall.TCP_KEEPIDLE, 10)
                // syscall.SetsockoptInt(int(fd), syscall.IPPROTO_TCP, syscall.TCP_KEEPINTVL, 5)
                // syscall.SetsockoptInt(int(fd), syscall.IPPROTO_TCP, syscall.TCP_KEEPCNT, 3)

                // 设置发送缓冲区
                syscall.SetsockoptInt(int(fd), syscall.SOL_SOCKET, syscall.SO_SNDBUF, 65536)

                // 设置接收缓冲区
                syscall.SetsockoptInt(int(fd), syscall.SOL_SOCKET, syscall.SO_RCVBUF, 65536)

                // 启用快速打开 (Linux)
                // syscall.SetsockoptInt(int(fd), syscall.SOL_TCP, syscall.TCP_FASTOPEN, 1)
            })
        },
    }

    conn, err := d.Dial("tcp", "localhost:8080")
    if err != nil {
        panic(err)
    }
    defer conn.Close()
}

// 监听器配置
func listenWithOpts() error {
    lc := &net.ListenConfig{
        Control: func(network, address string, c syscall.RawConn) error {
            return c.Control(func(fd uintptr) {
                // SO_REUSEADDR
                syscall.SetsockoptInt(int(fd), syscall.SOL_SOCKET, syscall.SO_REUSEADDR, 1)

                // SO_REUSEPORT (Linux 4.5+, macOS)
                syscall.SetsockoptInt(int(fd), syscall.SOL_SOCKET, 0x200, 1) // SO_REUSEPORT

                // 关闭延迟
                syscall.SetsockoptInt(int(fd), syscall.SOL_SOCKET, syscall.SO_LINGER, 0)
            })
        },
    }

    ln, err := lc.Listen(context.Background(), "tcp", ":8080")
    if err != nil {
        return err
    }
    defer ln.Close()

    return nil
}

---

网络编程检查清单

[ ] 使用 goroutine 池控制并发
[ ] 实现连接复用
[ ] 设置合理的超时时间
[ ] 使用零拷贝技术传输大文件
[ ] 使用 sync.Pool 复用缓冲区
[ ] 配置 TCP 参数优化性能
[ ] 实现优雅关闭
[ ] 监控连接数和延迟
[ ] 使用上下文取消控制
[ ] 处理连接泄露