17、实现协程版 TCP 服务器：accept、read、write 全异步化

好，咱们今天聊点硬核的。TCP 服务器，大家都会写。但你要让 accept、read、write 这三个家伙全部变成异步的，而且是用协程来驱动——嗯，这才是真正考验功底的地方。

我记得刚入行那会儿，公司有个高并发项目，用的还是传统的 select + 多线程模型。连接数一上去，线程切换开销直接让 CPU 冒烟。后来我试着把核心 IO 操作改成协程异步，效果立竿见影。今天我就把这条路怎么走，给你掰开揉碎讲清楚。

17.1 为什么要把 TCP 服务器全异步化？

说白了，传统阻塞式 IO 太浪费了。你想想看，一个线程调用 accept()，如果没有新连接，它就在那儿傻等。等一个连接来了，read() 又可能阻塞。线程多了，上下文切换开销大；线程少了，并发能力又上不去。

协程异步的思路是什么？就是让这些阻塞操作“看起来”是同步的，但底层自动切换出去干别的事。等数据准备好了，再切回来继续执行。这样，一个线程就能轻松管理成千上万个连接。

17.2 整体架构设计

先别急着写代码，咱们把架构理清楚。一个协程版 TCP 服务器，核心就三层：

• 事件循环层（Event Loop）： 负责监听所有 socket 上的 IO 事件，用 epoll（Linux）或 kqueue（macOS）实现。
• 协程调度层（Scheduler）： 管理所有协程的创建、挂起、恢复。每个协程对应一个连接的处理流程。
• 异步 IO 封装层（Async Wrapper）： 把 accept、read、write 包装成协程可调用的异步函数，内部自动注册到事件循环。

我习惯把这三层拆成独立的模块，这样后期维护和扩展都方便。你想想看，如果哪天你想换事件循环库，只需要改最底层，上层代码完全不用动。

17.3 核心数据结构

要实现全异步化，我们需要几个关键的数据结构。我直接给你看代码，然后逐行解释。

// 协程控制块
typedef struct coroutine {
    void *stack;           // 协程栈空间
    void *stack_top;       // 栈顶指针
    jmp_buf context;       // 寄存器上下文
    int status;            // 状态：READY, RUNNING, SUSPENDED, DEAD
    void (*func)(void *);  // 协程入口函数
    void *arg;             // 入口参数
    struct coroutine *next; // 链表指针，用于调度队列
} coroutine_t;

// 异步 IO 请求
typedef struct async_io_request {
    int fd;                // 文件描述符
    int events;            // 关注的事件：EPOLLIN / EPOLLOUT
    coroutine_t *co;       // 发起请求的协程
    void *buf;             // 数据缓冲区
    size_t len;            // 缓冲区长度
    ssize_t result;        // 操作结果
    int err;               // 错误码
    struct async_io_request *next;
} async_io_request_t;

// 调度器
typedef struct scheduler {
    coroutine_t *running;           // 当前正在执行的协程
    coroutine_t *ready_queue;       // 就绪队列
    int epoll_fd;                   // epoll 实例
    async_io_request_t *io_requests; // 待处理的 IO 请求链表
} scheduler_t;

这里有个细节我特别想强调：每个协程都有自己的独立栈空间。为什么？因为协程切换时，栈里的局部变量必须保留。如果所有协程共用一个栈，那切换回来数据就全乱了。我在项目中遇到过有人为了省内存用共享栈，结果 debug 了整整两天才找到 bug——协程 A 的局部变量被协程 B 覆盖了。

17.4 异步 accept 的实现

好，咱们先攻克第一个山头：accept。传统写法是 sock = accept(listen_fd, ...)，然后阻塞。协程版怎么做？

// 异步 accept 的协程入口
void async_accept_coroutine(void *arg) {
    scheduler_t *sched = (scheduler_t *)arg;
    int listen_fd = sched->listen_fd;
    
    while (1) {
        // 1. 把 listen_fd 注册到 epoll，关注 EPOLLIN
        // 2. 挂起当前协程，等待事件
        // 3. 事件到来后，恢复协程，执行 accept
        // 4. 创建新协程处理连接
        
        struct sockaddr_in client_addr;
        socklen_t addr_len = sizeof(client_addr);
        
        // 注册监听
        async_io_request_t req;
        req.fd = listen_fd;
        req.events = EPOLLIN;
        req.co = sched->running;
        add_io_request(sched, &req);
        
        // 挂起协程 —— 关键点
        coroutine_yield(sched);
        
        // 被唤醒后，执行 accept（此时一定可读）
        int client_fd = accept(listen_fd, 
                               (struct sockaddr *)&client_addr, 
                               &addr_len);
        if (client_fd < 0) {
            // 错误处理
            continue;
        }
        
        // 设置非阻塞
        set_nonblock(client_fd);
        
        // 创建新协程处理这个连接
        coroutine_t *client_co = coroutine_create(handle_client, 
                                                   (void *)(intptr_t)client_fd);
        add_to_ready_queue(sched, client_co);
    }
}

你看，核心就两步：注册事件 → 挂起等待 → 事件触发 → 恢复执行。这个模式在 read 和 write 里一模一样。

17.5 异步 read 和 write 的实现

read 和 write 的逻辑跟 accept 类似，但多了一个数据缓冲区的管理。咱们先看 read：

// 异步 read 的封装
ssize_t async_read(scheduler_t *sched, int fd, void *buf, size_t len) {
    // 先尝试非阻塞读一次
    ssize_t n = read(fd, buf, len);
    if (n > 0) {
        return n;  // 数据已经就绪，直接返回
    }
    
    if (n == 0) {
        return 0;  // 连接关闭
    }
    
    // 如果是 EAGAIN，说明数据还没到，需要挂起等待
    if (errno == EAGAIN || errno == EWOULDBLOCK) {
        async_io_request_t req;
        req.fd = fd;
        req.events = EPOLLIN;
        req.co = sched->running;
        req.buf = buf;
        req.len = len;
        req.result = 0;
        req.err = 0;
        
        add_io_request(sched, &req);
        
        // 挂起当前协程
        coroutine_yield(sched);
        
        // 被唤醒后，检查结果
        if (req.err != 0) {
            errno = req.err;
            return -1;
        }
        return req.result;
    }
    
    // 其他错误
    return -1;
}

write 的实现几乎一样，只是把 EPOLLIN 换成 EPOLLOUT。这里有个坑：写操作也可能返回 EAGAIN，特别是当发送缓冲区满了的时候。所以 write 也要异步化。

// 异步 write 的封装
ssize_t async_write(scheduler_t *sched, int fd, const void *buf, size_t len) {
    ssize_t total_written = 0;
    
    while (total_written < len) {
        ssize_t n = write(fd, buf + total_written, len - total_written);
        if (n > 0) {
            total_written += n;
            continue;
        }
        
        if (n == 0) {
            return total_written;
        }
        
        if (errno == EAGAIN || errno == EWOULDBLOCK) {
            // 写缓冲区满了，挂起等待可写事件
            async_io_request_t req;
            req.fd = fd;
            req.events = EPOLLOUT;
            req.co = sched->running;
            add_io_request(sched, &req);
            
            coroutine_yield(sched);
            
            // 唤醒后继续写
            continue;
        }
        
        return -1;
    }
    
    return total_written;
}

这里我用了循环，因为 write 不一定一次能写完所有数据。每次挂起恢复后，继续写剩余的部分。这个模式在项目中非常实用，尤其是处理大文件传输时。

17.6 事件循环与协程调度

有了上面的异步封装，事件循环就简单了。它的核心逻辑就是：

1. 调用 epoll_wait 等待事件
2. 事件到来后，找到对应的协程
3. 把协程加入就绪队列
4. 调度器从就绪队列取出协程，恢复执行

void event_loop(scheduler_t *sched) {
    struct epoll_event events[MAX_EVENTS];
    
    while (1) {
        // 1. 处理就绪队列中的协程
        while (sched->ready_queue != NULL) {
            coroutine_t *co = sched->ready_queue;
            sched->ready_queue = co->next;
            co->next = NULL;
            
            sched->running = co;
            coroutine_resume(co);
            sched->running = NULL;
        }
        
        // 2. 等待 IO 事件
        int nfds = epoll_wait(sched->epoll_fd, events, MAX_EVENTS, -1);
        
        for (int i = 0; i < nfds; i++) {
            async_io_request_t *req = (async_io_request_t *)events[i].data.ptr;
            
            // 把结果写回请求结构体
            if (events[i].events & EPOLLIN) {
                // 对于 read 请求，执行实际读取
                req->result = read(req->fd, req->buf, req->len);
                req->err = (req->result < 0) ? errno : 0;
            }
            
            if (events[i].events & EPOLLOUT) {
                // 对于 write 请求，标记可写
                req->result = 0;
                req->err = 0;
            }
            
            // 把对应的协程加入就绪队列
            add_to_ready_queue(sched, req->co);
        }
    }
}

这里有个设计取舍：到底是在事件循环里执行实际的 read/write，还是让协程恢复后自己执行？ 我个人习惯在事件循环里只做事件分发，实际的 IO 操作交给协程自己。这样事件循环的职责单一，不容易出 bug。

17.7 完整的服务器示例

最后，咱们把这些拼起来，看看一个完整的协程版 TCP 回声服务器长什么样：

// 处理客户端连接的协程
void handle_client(void *arg) {
    int client_fd = (int)(intptr_t)arg;
    scheduler_t *sched = get_scheduler();
    char buf[1024];
    
    while (1) {
        ssize_t n = async_read(sched, client_fd, buf, sizeof(buf));
        if (n <= 0) {
            break;  // 连接关闭或出错
        }
        
        ssize_t written = async_write(sched, client_fd, buf, n);
        if (written != n) {
            break;  // 写入失败
        }
    }
    
    close(client_fd);
}

// 主函数
int main() {
    scheduler_t *sched = scheduler_create();
    
    int listen_fd = socket(AF_INET, SOCK_STREAM, 0);
    // ... 绑定、监听、设置非阻塞 ...
    
    // 创建 accept 协程
    coroutine_t *accept_co = coroutine_create(async_accept_coroutine, sched);
    add_to_ready_queue(sched, accept_co);
    
    // 启动事件循环
    event_loop(sched);
    
    return 0;
}

你看，整个代码写下来，逻辑非常清晰。每个连接的处理都在一个协程里，用同步的方式写异步代码。没有回调地狱，没有复杂的线程同步。

17.8 避坑指南

做这个项目，有几个坑我踩过，给你提个醒：

• 信号安全：epoll_wait 可能被信号中断，返回 -1 且 errno 为 EINTR。一定要处理这种情况，否则服务器会莫名其妙挂掉。
• 协程栈溢出：如果协程里用了大量局部变量或深度递归，栈可能不够。建议在 debug 模式下开启栈保护（-fstack-protector）。
• 文件描述符泄漏：每个协程处理完连接后，一定要 close(fd)。我刚开始写的时候漏了一次，结果服务器跑了半天，文件描述符耗尽，再也无法接受新连接。
• 惊群效应：如果多线程跑多个事件循环，accept 时可能出现惊群。解决方案是用 SO_REUSEPORT 或者加一把全局锁。

嗯，今天就先聊到这儿。全异步化的 TCP 服务器，核心就是这三板斧：事件循环 + 协程调度 + 异步 IO 封装。你把这个框架搭好了，后面加什么协议解析、业务逻辑，都只是往协程里填代码的事。