第6章 I/O复用

多路复用

一、导读

6.1 本章学习目标

第6章是全书最核心的章节之一。I/O多路复用是高性能网络服务器的基石——Nginx、Redis、Node.js等现代高性能服务器都建立在I/O多路复用之上。本章将系统讲解 select、poll 和 epoll 三种I/O多路复用机制，从原理到实践，从API到性能，帮助你理解现代网络服务器的核心架构。

本章的核心学习目标包括：

理解I/O多路复用的基本思想：一个线程如何同时监控多个文件描述符？
掌握select函数的使用方法：fd_set的操作、超时处理、返回值分析。
理解select的局限性：FD_SETSIZE限制、O(n)复杂度、每次调用需要重置fd_set。
掌握poll函数的改进：如何克服select的FD数量限制？
深入理解epoll的工作原理：epoll_create、epoll_ctl、epoll_wait三个函数的协作。
理解水平触发（LT）与边缘触发（ET）的区别：各自的编程模型和适用场景。

6.2 为什么本章如此重要

在I/O多路复用出现之前，处理多个并发连接只有两种选择：

迭代模型：一次处理一个连接，其他连接等待。效率极低。

并发模型：每连接一个进程/线程。资源开销大，受限于进程/线程数量。

I/O多路复用提供了第三种选择：一个线程同时监控多个连接，哪个连接就绪就处理哪个。这使得一个线程就能处理成千上万个并发连接，极大地提升了服务器的可扩展性。

传统模型：
  10000个连接 → 10000个线程 → 10000 × 8MB栈 = 80GB内存
  上下文切换开销巨大

I/O多路复用：
  10000个连接 → 1个线程 → 几MB内存
  无上下文切换开销

6.3 核心问题引导

在阅读本章时，请带着以下问题思考：

select 的 fd_set 是如何实现的？为什么有 FD_SETSIZE 的限制？

为什么 select 每次返回后都需要重新设置 fd_set？

poll 相比 select 改进了什么？还有什么不足？

epoll 为什么能实现 O(1) 的复杂度？它的内部数据结构是什么？

边缘触发（ET）模式下，为什么必须使用非阻塞I/O？

为什么Nginx选择epoll而Redis选择单线程+epoll？

二、核心概念详解

6.1 I/O多路复用的基本概念

6.1.1 问题定义

当一个进程需要同时处理多个I/O操作时（如同时监听多个套接字），如何高效地等待这些操作就绪？

朴素方案：轮询（polling）——循环检查每个描述符是否就绪。

// 朴素轮询：浪费CPU
for (;;) {
    for (int i = 0; i < max_fd; i++) {
        if (fd_array[i] 有数据可读) {
            处理(fd_array[i]);
        }
    }
    usleep(1000);  // 短暂休眠，减少CPU占用
}

问题：要么不断轮询浪费CPU，要么休眠导致响应延迟。

I/O多路复用方案：让内核告诉我们哪些描述符就绪了。

// I/O多路复用：高效等待
// 告诉内核：我关注这些描述符 {fd1, fd2, fd3, ...}
// 内核返回：这些描述符就绪了 {fd2, fd5}
// 只处理就绪的描述符

6.1.2 I/O多路复用的三种机制

机制	出现时间	最大FD数	时间复杂度	可移植性
select	1983	FD_SETSIZE (通常1024)	O(n)	所有POSIX系统
poll	1986	无限制	O(n)	所有POSIX系统
epoll	2002	无限制	O(1)	Linux特有

6.2 select函数

6.2.1 函数原型

#include <sys/select.h>

int select(int maxfdp1, fd_set *readfds, fd_set *writefds, fd_set *exceptfds,
           struct timeval *timeout);
// 返回值：就绪的描述符数量，0表示超时，-1表示错误

参数说明：

maxfdp1：最大描述符+1（所有描述符集合中的最大值+1）
readfds：关注可读事件的描述符集合
writefds：关注可写事件的描述符集合
exceptfds：关注异常事件的描述符集合
timeout：超时时间。NULL表示永久等待

6.2.2 fd_set操作宏

FD_ZERO(fd_set *set);           // 清空集合
FD_SET(int fd, fd_set *set);    // 添加描述符到集合
FD_CLR(int fd, fd_set *set);    // 从集合中移除描述符
FD_ISSET(int fd, fd_set *set);  // 检查描述符是否在集合中

// select的基本用法
fd_set readfds;
FD_ZERO(&readfds);
FD_SET(listenfd, &readfds);
FD_SET(stdin, &readfds);

int maxfd = listenfd > STDIN_FILENO ? listenfd : STDIN_FILENO;

// 注意：select会修改readfds，只保留就绪的描述符
// 所以每次调用前都需要重新设置
FD_SET(listenfd, &readfds);
int nready = select(maxfd + 1, &readfds, NULL, NULL, NULL);

if (FD_ISSET(listenfd, &readfds)) {
    // 有新的连接
    int connfd = accept(listenfd, NULL, NULL);
}
if (FD_ISSET(STDIN_FILENO, &readfds)) {
    // 标准输入有数据
    char buf[256];
    read(STDIN_FILENO, buf, sizeof(buf));
}

6.2.3 超时控制

struct timeval {
    long tv_sec;   // 秒
    long tv_usec;  // 微秒
};

// 方式1：永久等待
select(maxfd + 1, &readfds, NULL, NULL, NULL);

// 方式2：固定超时
struct timeval tv;
tv.tv_sec = 5;
tv.tv_usec = 0;
int nready = select(maxfd + 1, &readfds, NULL, NULL, &tv);
if (nready == 0) {
    printf("Timeout\n");
}

// 方式3：非阻塞（立即返回）
struct timeval tv;
tv.tv_sec = 0;
tv.tv_usec = 0;
select(maxfd + 1, &readfds, NULL, NULL, &tv);

注意：某些系统会修改 timeout 参数，反映剩余的超时时间。不要假设 timeout 的值不变。

6.2.4 select实现并发服务器

// 使用select的并发echo服务器
#define FD_MAX 1024

int main(void) {
    int listenfd = socket(AF_INET, SOCK_STREAM, 0);
    // ... bind, listen ...

    int client[FD_MAX];
    for (int i = 0; i < FD_MAX; i++) {
        client[i] = -1;
    }
    int maxi = -1;  // client数组的最大索引

    fd_set rset, allset;
    FD_ZERO(&allset);
    FD_SET(listenfd, &allset);
    int maxfd = listenfd;

    for (;;) {
        rset = allset;  // select会修改rset，所以每次复制
        int nready = select(maxfd + 1, &rset, NULL, NULL, NULL);

        if (nready < 0) {
            if (errno == EINTR) continue;
            perror("select error");
            break;
        }

        // 检查新的连接
        if (FD_ISSET(listenfd, &rset)) {
            struct sockaddr_in cliaddr;
            socklen_t clilen = sizeof(cliaddr);
            int connfd = accept(listenfd, (struct sockaddr *)&cliaddr, &clilen);

            // 找到空的client槽位
            int i;
            for (i = 0; i < FD_MAX; i++) {
                if (client[i] < 0) {
                    client[i] = connfd;
                    break;
                }
            }

            FD_SET(connfd, &allset);
            if (connfd > maxfd) maxfd = connfd;
            if (i > maxi) maxi = i;

            if (--nready <= 0) continue;  // 没有其他就绪的描述符
        }

        // 检查已连接的客户
        for (int i = 0; i <= maxi; i++) {
            int connfd = client[i];
            if (connfd < 0) continue;

            if (FD_ISSET(connfd, &rset)) {
                char buf[MAXLINE];
                ssize_t n = read(connfd, buf, sizeof(buf));

                if (n <= 0) {
                    // 连接关闭或出错
                    close(connfd);
                    FD_CLR(connfd, &allset);
                    client[i] = -1;
                } else {
                    write(connfd, buf, n);  // 回显
                }

                if (--nready <= 0) break;  // 没有其他就绪的描述符
            }
        }
    }
}

6.2.5 select的局限性

FD_SETSIZE限制：通常1024，无法监控超过此数量的描述符

O(n)复杂度：每次调用都需要遍历所有描述符

fd_set需要重置：每次select后fd_set被修改，下次调用前需要重新设置

内核/用户空间拷贝：每次调用都需要将整个fd_set从用户空间拷贝到内核空间

6.3 poll函数

6.3.1 函数原型

#include <poll.h>

int poll(struct pollfd *fds, nfds_t nfds, int timeout);
// 返回值：就绪的描述符数量，0表示超时，-1表示错误

struct pollfd {
    int   fd;       // 文件描述符
    short events;   // 关注的事件（输入）
    short revents;  // 发生的事件（输出）
};

// 常用事件标志
#define POLLIN    0x0001  // 有数据可读
#define POLLOUT   0x0004  // 可以写入
#define POLLERR   0x0008  // 错误
#define POLLHUP   0x0010  // 挂起（连接关闭）
#define POLLNVAL  0x0020  // 描述符无效

6.3.2 poll的使用示例

// 使用poll的并发服务器
#define MAX_CLIENTS 10000

struct pollfd client_fds[MAX_CLIENTS + 1];
int nfds = 1;  // 当前监控的描述符数量

// 初始化
client_fds[0].fd = listenfd;
client_fds[0].events = POLLIN;

for (;;) {
    int nready = poll(client_fds, nfds, -1);

    if (nready < 0) {
        if (errno == EINTR) continue;
        perror("poll error");
        break;
    }

    // 检查新的连接
    if (client_fds[0].revents & POLLIN) {
        struct sockaddr_in cliaddr;
        socklen_t clilen = sizeof(cliaddr);
        int connfd = accept(listenfd, (struct sockaddr *)&cliaddr, &clilen);

        // 添加到监控列表
        client_fds[nfds].fd = connfd;
        client_fds[nfds].events = POLLIN;
        nfds++;
    }

    // 检查已连接的客户
    for (int i = 1; i < nfds; i++) {
        if (client_fds[i].revents & POLLIN) {
            char buf[MAXLINE];
            ssize_t n = read(client_fds[i].fd, buf, sizeof(buf));

            if (n <= 0) {
                close(client_fds[i].fd);
                // 用最后一个填充空位
                client_fds[i] = client_fds[nfds - 1];
                nfds--;
                i--;  // 重新检查当前位置
            } else {
                write(client_fds[i].fd, buf, n);
            }
        }
    }
}

6.3.3 poll vs select

特性	select	poll
最大FD数	FD_SETSIZE (1024)	无限制
数据结构	fd_set (位图)	pollfd数组
重置需求	需要重新设置fd_set	不需要，events不变
事件类型	读/写/异常	更丰富的事件类型
性能	O(n)	O(n)
可移植性	所有POSIX	所有POSIX

poll的主要改进是取消了FD数量限制，但性能仍然是O(n)——每次调用都需要遍历所有描述符。当连接数很大但活跃连接很少时，效率低下。

6.4 epoll：Linux的高效I/O复用

6.4.1 三个核心函数

#include <sys/epoll.h>

// 1. 创建epoll实例
int epoll_create1(int flags);
// 返回值：epoll文件描述符，-1表示错误

// 2. 注册/修改/删除感兴趣的事件
int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event);
// op: EPOLL_CTL_ADD (添加), EPOLL_CTL_MOD (修改), EPOLL_CTL_DEL (删除)

// 3. 等待事件发生
int epoll_wait(int epfd, struct epoll_event *events, int maxevents, int timeout);
// 返回值：就绪的事件数量，0表示超时，-1表示错误

struct epoll_event {
    uint32_t events;    // epoll事件
    epoll_data_t data;  // 用户数据
};

typedef union epoll_data {
    void    *ptr;
    int      fd;
    uint32_t u32;
    uint64_t u64;
} epoll_data_t;

// 常用事件
#define EPOLLIN     0x001  // 有数据可读
#define EPOLLOUT    0x004  // 可以写入
#define EPOLLET     (1 << 31)  // 边缘触发
#define EPOLLONESHOT (1 << 30) // 一次性触发

6.4.2 epoll的基本用法

// 创建epoll实例
int epfd = epoll_create1(0);

// 添加监听套接字
struct epoll_event ev;
ev.events = EPOLLIN;
ev.data.fd = listenfd;
epoll_ctl(epfd, EPOLL_CTL_ADD, listenfd, &ev);

// 事件循环
struct epoll_event events[MAX_EVENTS];
for (;;) {
    int nready = epoll_wait(epfd, events, MAX_EVENTS, -1);

    for (int i = 0; i < nready; i++) {
        if (events[i].data.fd == listenfd) {
            // 新连接
            int connfd = accept(listenfd, NULL, NULL);
            
            ev.events = EPOLLIN;
            ev.data.fd = connfd;
            epoll_ctl(epfd, EPOLL_CTL_ADD, connfd, &ev);
        } else {
            // 客户数据
            int connfd = events[i].data.fd;
            char buf[MAXLINE];
            ssize_t n = read(connfd, buf, sizeof(buf));
            
            if (n <= 0) {
                close(connfd);
                epoll_ctl(epfd, EPOLL_CTL_DEL, connfd, NULL);
            } else {
                write(connfd, buf, n);
            }
        }
    }
}

6.4.3 epoll为什么高效

epoll的高效来自三个关键设计：

事件驱动而非轮询：epoll_wait只返回就绪的描述符，不需要遍历所有描述符。O(1)复杂度（相对于活跃连接数）。

内核数据结构：epoll在内核中使用红黑树存储监控的描述符，使用就绪链表存储就绪的描述符。添加/删除描述符是O(log n)，获取就绪列表是O(1)。

mmap减少拷贝：epoll使用mmap共享内核和用户空间的数据，避免了select/poll中每次调用都需要拷贝fd_set的开销。

select/poll：
  每次调用 → 拷贝所有FD到内核 → 遍历所有FD → 拷贝就绪FD到用户空间
  复杂度：O(n)，n = 总FD数

epoll：
  epoll_ctl → 添加到红黑树（仅修改一次）
  epoll_wait → 直接返回就绪链表
  复杂度：O(k)，k = 就绪FD数

6.4.4 水平触发（LT）vs 边缘触发（ET）

水平触发（Level-Triggered, LT）：

默认模式
只要描述符的读缓冲区中有数据，epoll_wait就会持续返回该描述符的可读事件
编程模型与select/poll类似，容易使用

边缘触发（Edge-Triggered, ET）：

只有描述符的状态发生变化时（如新数据到达），epoll_wait才返回该描述符的事件
如果一次没读完所有数据，epoll_wait不会再次返回，直到有新数据到达
性能更高，但编程更复杂

// 边缘触发模式的使用
// 1. 必须使用非阻塞I/O
int flags = fcntl(connfd, F_GETFL, 0);
fcntl(connfd, F_SETFL, flags | O_NONBLOCK);

// 2. 注册ET模式
ev.events = EPOLLIN | EPOLLET;
ev.data.fd = connfd;
epoll_ctl(epfd, EPOLL_CTL_ADD, connfd, &ev);

// 3. 必须循环读取直到EAGAIN
while (1) {
    ssize_t n = read(connfd, buf, sizeof(buf));
    if (n > 0) {
        // 处理数据
    } else if (n == -1 && errno == EAGAIN) {
        // 所有数据已读完
        break;
    } else if (n == 0) {
        // 连接关闭
        close(connfd);
        epoll_ctl(epfd, EPOLL_CTL_DEL, connfd, NULL);
        break;
    } else {
        // 错误
        perror("read");
        break;
    }
}

为什么ET模式必须使用非阻塞I/O？

在ET模式下，epoll_wait只在状态变化时通知。如果使用阻塞I/O，read可能在数据已经读完的情况下仍然阻塞（因为epoll_wait不会再次通知）。非阻塞I/O确保read在没有数据时立即返回EAGAIN，不会阻塞。

6.4.5 LT vs ET对比

特性	LT（水平触发）	ET（边缘触发）
通知时机	只要有数据就通知	只在状态变化时通知
编程难度	简单	复杂（必须循环读写）
I/O模式	阻塞/非阻塞均可	必须非阻塞
性能	略低（可能多次通知）	更高（减少epoll_wait调用）
适用场景	大多数应用	高性能、大量连接

6.5 epoll服务器完整示例

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <sys/epoll.h>
#include <fcntl.h>
#include <errno.h>

#define MAX_EVENTS 64
#define MAXLINE 4096

void set_nonblocking(int fd) {
    int flags = fcntl(fd, F_GETFL, 0);
    fcntl(fd, F_SETFL, flags | O_NONBLOCK);
}

int main(int argc, char **argv) {
    int port = 8080;
    if (argc > 1) port = atoi(argv[1]);

    // 创建监听套接字
    int listenfd = socket(AF_INET, SOCK_STREAM, 0);
    int optval = 1;
    setsockopt(listenfd, SOL_SOCKET, SO_REUSEADDR, &optval, sizeof(optval));

    struct sockaddr_in servaddr;
    memset(&servaddr, 0, sizeof(servaddr));
    servaddr.sin_family = AF_INET;
    servaddr.sin_addr.s_addr = htonl(INADDR_ANY);
    servaddr.sin_port = htons(port);

    bind(listenfd, (struct sockaddr *)&servaddr, sizeof(servaddr));
    listen(listenfd, 128);

    // 创建epoll实例
    int epfd = epoll_create1(0);

    // 添加监听套接字到epoll
    struct epoll_event ev;
    ev.events = EPOLLIN;
    ev.data.fd = listenfd;
    epoll_ctl(epfd, EPOLL_CTL_ADD, listenfd, &ev);

    struct epoll_event events[MAX_EVENTS];
    int num_fds = 1;  // 当前监控的FD数量

    printf("Epoll echo server on port %d\n", port);

    for (;;) {
        int nready = epoll_wait(epfd, events, MAX_EVENTS, -1);
        if (nready < 0) {
            if (errno == EINTR) continue;
            perror("epoll_wait");
            break;
        }

        for (int i = 0; i < nready; i++) {
            if (events[i].data.fd == listenfd) {
                // 新连接
                struct sockaddr_in cliaddr;
                socklen_t clilen = sizeof(cliaddr);
                int connfd = accept(listenfd, (struct sockaddr *)&cliaddr, &clilen);
                if (connfd < 0) {
                    perror("accept");
                    continue;
                }

                set_nonblocking(connfd);
                ev.events = EPOLLIN | EPOLLET;  // 边缘触发
                ev.data.fd = connfd;
                epoll_ctl(epfd, EPOLL_CTL_ADD, connfd, &ev);
                num_fds++;

                char ip[INET_ADDRSTRLEN];
                inet_ntop(AF_INET, &cliaddr.sin_addr, ip, sizeof(ip));
                printf("+ Connection from %s:%d (fds: %d)\n",
                       ip, ntohs(cliaddr.sin_port), num_fds);
            } else {
                // 客户数据
                int connfd = events[i].data.fd;
                char buf[MAXLINE];

                // ET模式必须循环读取
                while (1) {
                    ssize_t n = read(connfd, buf, sizeof(buf));
                    if (n > 0) {
                        // 回显
                        write(connfd, buf, n);
                    } else if (n == -1 && errno == EAGAIN) {
                        // 数据读完
                        break;
                    } else if (n == 0) {
                        // 连接关闭
                        printf("- Connection closed (fd=%d)\n", connfd);
                        epoll_ctl(epfd, EPOLL_CTL_DEL, connfd, NULL);
                        close(connfd);
                        num_fds--;
                        break;
                    } else {
                        perror("read");
                        epoll_ctl(epfd, EPOLL_CTL_DEL, connfd, NULL);
                        close(connfd);
                        num_fds--;
                        break;
                    }
                }
            }
        }
    }

    close(epfd);
    close(listenfd);
    return 0;
}

三、实战案例

6.1 三种机制的性能对比测试

// 简单的性能测试框架
#include <sys/time.h>

double get_time_ms(void) {
    struct timeval tv;
    gettimeofday(&tv, NULL);
    return tv.tv_sec * 1000.0 + tv.tv_usec / 1000.0;
}

// 测试select
void benchmark_select(int num_fds) {
    fd_set readfds;
    FD_ZERO(&readfds);
    for (int i = 3; i < num_fds + 3; i++) {
        FD_SET(i, &readfds);
    }
    
    struct timeval tv = {0, 1000};  // 1ms超时
    double start = get_time_ms();
    for (int i = 0; i < 10000; i++) {
        fd_set tmp = readfds;
        select(num_fds + 3, &tmp, NULL, NULL, &tv);
    }
    double elapsed = get_time_ms() - start;
    printf("select(%d fds, 10000 calls): %.2f ms\n", num_fds, elapsed);
}

// 测试epoll
void benchmark_epoll(int num_fds) {
    int epfd = epoll_create1(0);
    struct epoll_event ev;
    for (int i = 3; i < num_fds + 3; i++) {
        ev.events = EPOLLIN;
        ev.data.fd = i;
        epoll_ctl(epfd, EPOLL_CTL_ADD, i, &ev);
    }
    
    struct epoll_event events[64];
    double start = get_time_ms();
    for (int i = 0; i < 10000; i++) {
        epoll_wait(epfd, events, 64, 1);
    }
    double elapsed = get_time_ms() - start;
    printf("epoll(%d fds, 10000 calls): %.2f ms\n", num_fds, elapsed);
    
    close(epfd);
}

6.2 使用epoll实现HTTP服务器骨架

// 极简HTTP服务器（仅演示epoll的使用）
void handle_http(int connfd) {
    char buf[4096];
    ssize_t n = read(connfd, buf, sizeof(buf) - 1);
    if (n <= 0) return;
    buf[n] = '\0';
    
    // 解析HTTP请求（极简）
    if (strstr(buf, "GET") == buf) {
        char response[] =
            "HTTP/1.1 200 OK\r\n"
            "Content-Type: text/html\r\n"
            "Content-Length: 13\r\n"
            "Connection: close\r\n"
            "\r\n"
            "Hello, World!";
        write(connfd, response, strlen(response));
    }
}

四、本章小结

本章系统讲解了I/O多路复用的三种机制，核心要点回顾：

select：最早的I/O复用机制。使用fd_set位图，有FD_SETSIZE限制（通常1024）。每次调用需要重置fd_set。O(n)复杂度。

poll：使用pollfd数组，取消了FD数量限制。不需要重置事件。但仍然是O(n)复杂度。

epoll：Linux特有的高效I/O复用。使用红黑树+就绪链表，O(1)获取就绪描述符。三个函数：epoll_create1创建实例，epoll_ctl注册事件，epoll_wait等待事件。

LT vs ET：LT（水平触发）只要有数据就通知，编程简单；ET（边缘触发）只在状态变化时通知，性能更高但必须使用非阻塞I/O并循环读写。

选择建议：需要可移植性用select/poll；Linux平台追求高性能用epoll。现代高性能服务器（Nginx、Redis等）都基于epoll。

五、思考与练习

思考题

为什么epoll在大量连接但少量活跃的场景下性能远优于select？

ET模式下如果只读了一半数据就去处理其他事件，会发生什么？如何避免？

epoll的EPOLLONESHOT标志有什么作用？在多线程epoll中如何使用？

为什么epoll需要使用非阻塞I/O？如果使用阻塞I/O在ET模式下会有什么问题？

select的"描述符就绪"和epoll的"事件就绪"有什么本质区别？

编程练习

epoll + 线程池：实现一个epoll服务器，主线程负责epoll_wait，将就绪的连接分发给线程池中的工作线程处理。

支持UDP的epoll服务器：扩展epoll服务器，同时监控TCP监听套接字和UDP套接字。

epoll超时管理：实现一个连接超时机制——如果一个客户连接后60秒内没有发送数据，自动关闭连接。提示：使用timerfd或epoll的timeout参数。