Day36 IO多路复用技术
day36 IO多路复用技术
核心概念
定义与本质
IO多路复用:单线程或单进程同时监测若干个文件描述符是否可以执行IO操作的能力。
本质:解决用更少的资源(避免线程/进程创建开销、上下文切换成本、资源竞争)完成更多事件流的并发处理。
作用与背景
- 典型场景:
- 操作系统需同时处理键盘/鼠标输入、中断信号等事件流
- Web服务器(如Nginx)需同时处理来自N个客户端的请求
- 并发本质:逻辑控制流在时间上的重叠(通过CPU时分复用实现)
- 传统并发成本:
- 线程/进程创建开销
- CPU上下文切换成本(Context Switch)
- 多线程资源竞争问题
五种IO模型详解
1. 阻塞IO(最常用默认模式)
核心特性:当IO操作无法立即完成时,进程/线程会挂起等待,直到数据准备就绪。
FIFO管道通信示例
写端代码 (01fifo_w):
#include <errno.h> // 错误码定义(如EEXIST)
#include <fcntl.h> // 文件控制宏(O_WRONLY)
#include <stdio.h> // 标准IO函数(perror)
#include <stdlib.h> // 标准库函数
#include <string.h> // 字符串操作(strlen)
#include <sys/stat.h> // mkfifo函数定义
#include <sys/types.h> // 基础数据类型
#include <unistd.h> // 系统调用(write/sleep/close)int main(int argc, char *argv[]) {// 创建命名管道(权限0666),若已存在则忽略错误int ret = mkfifo("myfifo", 0666);if (-1 == ret) {if (EEXIST != errno) { // 非"已存在"错误则退出perror("mkfifo error\n");return 1;}}// 以只写模式打开FIFO(阻塞:等待读端连接)int fd = open("myfifo", O_WRONLY);if (-1 == fd) {perror("open error\n");return 1;}while (1) {char buf[512] = "hello,this is fifo tested...\n";// 向管道写入数据(包含字符串结束符)write(fd, buf, strlen(buf) + 1);sleep(3); // 每3秒写一次}close(fd); // 实际不会执行(无限循环)return 0;
}
理想运行结果:
- 执行
./01fifo_w
后进程阻塞,等待读端连接 - 连接读端后每3秒向管道写入测试字符串
- 终端无输出,持续写入数据
读端代码 (02fifo_r):
#include <errno.h>
#include <fcntl.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <unistd.h>int main(int argc, char *argv[]) {// 创建命名管道(同写端逻辑)int ret = mkfifo("myfifo", 0666);if (-1 == ret) {if (EEXIST != errno) {perror("mkfifo error\n");return 1;}}// 以只读模式打开FIFO(阻塞:等待写端连接)int fd = open("myfifo", O_RDONLY);if (-1 == fd) {perror("open error\n");return 1;}while (1) {char buf[512] = {0};// 从管道读取数据(阻塞等待)read(fd, buf, sizeof(buf));printf("fifo :%s\n", buf); // 显示管道内容bzero(buf, sizeof(buf));// 从终端读取输入(阻塞等待)fgets(buf, sizeof(buf), stdin);printf("terminal:%s", buf);fflush(stdout); // 立即刷新输出}close(fd);return 0;
}
理想运行结果:
- 执行
./02fifo_r
后进程阻塞,等待写端连接 - 连接写端后:
fifo :hello,this is fifo tested...terminal:用户输入内容 // 显示终端输入
阻塞IO特点:
- 读写操作会挂起进程直到数据就绪
- 简单易用但无法同时处理多事件流
- FIFO管道特性:无对端连接时
open()
阻塞
2. 非阻塞IO
核心特性:通过fcntl()
动态设置O_NONBLOCK
标志,使IO操作立即返回(无数据时返回EAGAIN
错误)。
代码改造关键
// 获取当前文件状态标志
int flag = fcntl(fd, F_GETFL, 0);
// 添加非阻塞标志
fcntl(fd, F_SETFL, flag | O_NONBLOCK);
非阻塞FIFO读取示例
#include <errno.h>
#include <fcntl.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <unistd.h>int main(int argc, char *argv[]) {// 创建FIFO(同阻塞版)int ret = mkfifo("myfifo", 0666);if (-1 == ret && EEXIST != errno) {perror("mkfifo error\n");return 1;}int fd = open("myfifo", O_RDONLY);if (-1 == fd) {perror("open error\n");return 1;}// 设置FIFO为非阻塞模式int flag = fcntl(fd, F_GETFL, 0);fcntl(fd, F_SETFL, flag | O_NONBLOCK);// 设置标准输入为非阻塞模式flag = fcntl(fileno(stdin), F_GETFL, 0);fcntl(0, F_SETFL, flag | O_NONBLOCK);while (1) {char buf[512] = {0};// 非阻塞读管道:无数据时read()立即返回-1if (read(fd, buf, sizeof(buf)) > 0) {printf("fifo :%s\n", buf);}bzero(buf, sizeof(buf));// 非阻塞终端输入:无输入时fgets()返回NULLif (fgets(buf, sizeof(buf), stdin)) {printf("terminal:%s", buf);fflush(stdout);}}close(fd);return 0;
}
理想运行结果:
- 无数据时:
terminal:用户输入内容 // 仅显示终端输入
- 有管道数据时:
fifo :hello,this is fifo tested...terminal:用户输入内容
非阻塞IO特点:
- 避免单个IO操作阻塞整个进程
- 需要轮询检查数据就绪状态(忙等待)
- 通过
errno == EAGAIN
判断无数据
3. 信号驱动IO
核心特性:设置O_ASYNC
标志,当IO就绪时内核发送SIGIO
信号。
三步配置
// 1. 添加异步标志
int flag = fcntl(fd, F_GETFL, 0);
fcntl(fd, F_SETFL, flag | O_ASYNC); // 2. 设置信号接收者
fcntl(fd, F_SETOWN, getpid()); // 当前进程接收信号// 3. 注册信号处理函数
signal(SIGIO, myhandle); // myhandle处理SIGIO
信号驱动FIFO示例
#include <errno.h>
#include <fcntl.h>
#include <signal.h> // 信号处理头文件
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <unistd.h>int fd; // 全局描述符供信号处理函数使用// SIGIO信号处理函数
void myhandle(int num) {char buf[512] = {0};read(fd, buf, sizeof(buf)); // 从管道读取数据printf("fifo :%s\n", buf); // 打印管道内容
}int main(int argc, char *argv[]) {signal(SIGIO, myhandle); // 注册信号处理函数// 创建并打开FIFO(同前)int ret = mkfifo("myfifo", 0666);if (-1 == ret && EEXIST != errno) {perror("mkfifo error\n");return 1;}fd = open("myfifo", O_RDONLY);if (-1 == fd) {perror("open error\n");return 1;}// 配置信号驱动IOint flag = fcntl(fd, F_GETFL);fcntl(fd, F_SETFL, flag | O_ASYNC); // 添加异步标志fcntl(fd, F_SETOWN, getpid()); // 设置接收进程while (1) {char buf[512] = {0};bzero(buf, sizeof(buf));fgets(buf, sizeof(buf), stdin); // 阻塞读终端printf("terminal:%s", buf);fflush(stdout);}close(fd);return 0;
}
理想运行结果:
- 管道数据到达时自动触发
myhandle
:fifo :hello,this is fifo tested...
- 终端输入独立处理:
terminal:用户输入内容
信号驱动IO特点:
- 通过信号机制异步通知IO就绪
- 避免轮询开销,但信号处理函数限制多
- 实际应用较少(仅部分场景适用)
4. 并发模型(进程/线程)
核心机制:
- 进程:
fork()
创建子进程处理独立事件流 - 线程:
pthread_create()
创建线程共享内存空间
局限性:
- 资源开销大(创建/销毁成本高)
- 上下文切换消耗CPU资源
- 需处理同步/互斥问题(锁、条件变量等)
5. IO多路复用(核心重点)
核心价值:单线程内高效监控多个文件描述符的状态变化。
select模型
函数原型:
#include <sys/select.h>
int select(int nfds, fd_set *readfds, fd_set *writefds,fd_set *exceptfds, struct timeval *timeout);
关键宏:
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监控FIFO+终端示例:
#include <errno.h>
#include <fcntl.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <unistd.h>
#include <sys/select.h> // select头文件int main(int argc, char *argv[]) {// 创建并打开FIFO(同前)int ret = mkfifo("myfifo", 0666);if (-1 == ret && EEXIST != errno) {perror("mkfifo error\n");return 1;}int fd = open("myfifo", O_RDONLY);if (-1 == fd) {perror("open error\n");return 1;}fd_set rd_set, tmp_set; // 读事件集合FD_ZERO(&rd_set);FD_ZERO(&tmp_set);// 添加监控对象:标准输入(0) + FIFOFD_SET(0, &tmp_set);FD_SET(fd, &tmp_set);while (1) {rd_set = tmp_set; // 每次循环重置(select会修改集合)// 阻塞等待事件(NULL=永久等待)select(fd + 1, &rd_set, NULL, NULL, NULL);char buf[512] = {0};// 检查FIFO是否就绪if (FD_ISSET(fd, &rd_set)) {read(fd, buf, sizeof(buf));printf("fifo :%s\n", buf);}// 检查终端输入是否就绪if (FD_ISSET(0, &rd_set)) {bzero(buf, sizeof(buf));fgets(buf, sizeof(buf), stdin);printf("terminal:%s", buf);fflush(stdout);}}close(fd);return 0;
}
理想运行结果:
- 任意事件到达时立即响应:
fifo :hello,this is fifo tested... // 管道数据到达terminal:用户输入内容 // 终端输入到达
select局限性:
- 描述符数量限制(
FD_SETSIZE=1024
) - 每次调用需传递整个描述符集合(用户/内核空间拷贝)
- 返回后需遍历所有描述符检查就绪状态(O(n)复杂度)
epoll模型(高性能替代方案)
核心函数:
int epoll_create(int size); // 创建epoll实例
int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event); // 管理描述符
int epoll_wait(int epfd, struct epoll_event *events, int maxevents, int timeout); // 等待事件
epoll监控FIFO+终端示例:
#include <errno.h>
#include <fcntl.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/epoll.h> // epoll头文件
#include <sys/stat.h>
#include <sys/types.h>
#include <unistd.h>// 添加描述符到epoll实例
int add_fd(int epfd, int fd) {struct epoll_event ev;ev.events = EPOLLIN; // 监听可读事件ev.data.fd = fd; // 存储描述符用于识别// 添加到epoll实例if (epoll_ctl(epfd, EPOLL_CTL_ADD, fd, &ev) == -1) {perror("add fd");return 1;}return 0;
}int main(int argc, char *argv[]) {// 创建并打开FIFO(同前)int ret = mkfifo("myfifo", 0666);if (-1 == ret && EEXIST != errno) {perror("mkfifo error\n");return 1;}int fd = open("myfifo", O_RDONLY);if (-1 == fd) {perror("open error\n");return 1;}struct epoll_event rev[2]; // 存储就绪事件// 创建epoll实例int epfd = epoll_create(2);if (epfd == -1) {perror("epoll_create");return 1;}// 添加监控对象add_fd(epfd, 0); // 标准输入add_fd(epfd, fd); // FIFOwhile (1) {char buf[512] = {0};// 等待事件(-1=永久阻塞)int ep_ret = epoll_wait(epfd, rev, 2, -1);for (int i = 0; i < ep_ret; i++) {if (rev[i].data.fd == fd) { // FIFO事件read(fd, buf, sizeof(buf));printf("fifo :%s\n", buf);} else if (rev[i].data.fd == 0) { // 终端输入bzero(buf, sizeof(buf));fgets(buf, sizeof(buf), stdin);printf("terminal:%s", buf);fflush(stdout);}}}close(fd);return 0;
}
理想运行结果:
- 事件到达时立即响应(同select版):
fifo :hello,this is fifo tested... terminal:用户输入内容
epoll优势:
- 无描述符数量限制(仅受系统限制)
- 事件就绪时直接返回就绪列表(O(1)复杂度)
- 用户/内核空间共享内存(避免数据拷贝)
- 支持水平触发(LT)和边沿触发(ET)模式
select vs poll vs epoll 核心对比
特性 | select | poll | epoll |
---|---|---|---|
描述符上限 | FD_SETSIZE(通常1024) | 无硬限制(受系统限制) | 无硬限制 |
性能复杂度 | O(n) 遍历所有描述符 | O(n) 遍历所有描述符 | O(1) 仅处理就绪事件 |
数据拷贝 | 每次调用全量拷贝 | 每次调用全量拷贝 | 共享内存(仅第一次拷贝) |
事件返回方式 | 修改输入集合 | 填充就绪数组 | 直接返回就绪事件列表 |
触发模式 | 仅水平触发 | 仅水平触发 | 支持水平/边沿触发 |
适用场景 | 小规模连接(<1000) | 中等规模连接 | 高并发场景(如Web服务器) |
关键结论:
- epoll在高并发场景下性能显著优于select/poll
- epoll优势前提:监听大量描述符 + 每次仅少量事件就绪
- epoll需管理epoll实例(需
close()
释放资源)
TCP应用实例
基于select的TCP服务器 (ser.c)
#include <netinet/in.h> // 定义网络地址结构(如sockaddr_in)及字节序转换函数(如htons)
#include <netinet/ip.h> // IP协议相关定义(本程序未实际使用)
#include <stdio.h> // 标准输入输出函数(如printf、perror)
#include <stdlib.h> // 标准库函数
#include <string.h> // 字符串处理函数(如bzero、strlen、sprintf)
#include <sys/socket.h> // 套接字相关函数(如socket、bind、listen、accept)
#include <sys/types.h> // 基本系统数据类型
#include <time.h> // 时间相关函数(如time、ctime)
#include <unistd.h> // 系统调用函数(如close)
#include <sys/select.h> // 提供select函数及文件描述符集相关定义typedef struct sockaddr *(SA); // 将struct sockaddr*简化为SAint main(int argc, char **argv)
{// 创建监听套接字:AF_INET(IPv4协议),SOCK_STREAM(TCP流式套接字),0(默认协议)int listfd = socket(AF_INET, SOCK_STREAM, 0);if (-1 == listfd) {perror("scoket error\n"); // 套接字创建失败return 1;}// 定义服务器和客户端的地址结构并初始化struct sockaddr_in ser, cli;bzero(&ser, sizeof(ser));bzero(&cli, sizeof(cli));// 设置服务器地址信息ser.sin_family = AF_INET; // 使用IPv4协议ser.sin_port = htons(50000); // 绑定端口50000ser.sin_addr.s_addr = INADDR_ANY; // 绑定到本机所有可用IP地址// 将监听套接字与服务器地址绑定int ret = bind(listfd, (SA)&ser, sizeof(ser));if (-1 == ret) {perror("bind");return 1;}// 将套接字设为监听状态listen(listfd, 3);socklen_t len = sizeof(cli);// 1. 初始化文件描述符集fd_set rd_set, tmp_set;FD_ZERO(&rd_set);FD_ZERO(&tmp_set);// 2. 向临时集合添加初始监听的文件描述符(监听套接字listfd)FD_SET(listfd, &tmp_set);int maxfd = listfd; // 记录当前最大文件描述符time_t tm;while (1){rd_set = tmp_set; // 恢复监听集合// 调用select监听可读事件select(maxfd + 1, &rd_set, NULL, NULL, NULL);// 遍历所有可能的文件描述符for (int i = 0; i < maxfd + 1; i++){// 若事件来自监听套接字listfd(新客户端连接请求)if (FD_ISSET(i, &rd_set) && i == listfd){// 接受客户端连接int conn = accept(listfd, (SA)&cli, &len);if (-1 == conn) {perror("accept");close(conn);continue;}// 将新连接的套接字添加到监听集合FD_SET(conn, &tmp_set);// 更新最大文件描述符if (conn > maxfd) {maxfd = conn;}}// 若事件来自已连接的客户端套接字if (FD_ISSET(i, &rd_set) && i != listfd){int conn = i;char buf[1024] = {0};// 接收客户端发送的数据int ret = recv(conn, buf, sizeof(buf), 0);if (ret <= 0) {printf("client offline\n");FD_CLR(conn, &tmp_set);close(conn);continue;}// 获取当前时间戳time(&tm);// 将接收的数据与当前时间拼接sprintf(buf, "%s %s", buf, ctime(&tm));// 将拼接后的数据回发给客户端send(conn, buf, strlen(buf), 0);}}}close(listfd);return 0;
}
基于epoll的TCP服务器 (ser.c)
#include <netinet/in.h> // 提供Internet地址族相关定义
#include <netinet/ip.h> // 提供IP协议相关定义
#include <stdio.h> // 标准输入输出函数
#include <stdlib.h> // 标准库函数
#include <string.h> // 字符串处理函数
#include <sys/epoll.h> // epoll相关函数
#include <sys/socket.h> // 套接字相关函数
#include <sys/types.h> // 基本系统数据类型
#include <time.h> // 时间相关函数
#include <unistd.h> // Unix标准函数
typedef struct sockaddr *(SA); // 简化sockaddr指针类型/*** 向epoll实例中添加文件描述符* @param epfd epoll实例的文件描述符* @param fd 要添加的文件描述符* @return 0表示成功,1表示失败*/
int add_fd(int epfd, int fd)
{struct epoll_event ev;ev.events = EPOLLIN; // 关注读事件ev.data.fd = fd; // 绑定要监控的文件描述符// 向epoll实例添加文件描述符及对应的事件int ret = epoll_ctl(epfd, EPOLL_CTL_ADD, fd, &ev);if (-1 == ret){perror("add fd");return 1;}return 0;
}/*** 从epoll实例中删除文件描述符* @param epfd epoll实例的文件描述符* @param fd 要删除的文件描述符* @return 0表示成功,1表示失败*/
int del_fd(int epfd, int fd)
{struct epoll_event ev;ev.events = EPOLLIN;ev.data.fd = fd;// 从epoll实例中删除文件描述符int ret = epoll_ctl(epfd, EPOLL_CTL_DEL, fd, &ev);if (-1 == ret){perror("add fd"); // 原代码错误信息,应为"del fd"return 1;}return 0;
}int main(int argc, char **argv)
{// 创建监听套接字(IPv4协议,字节流服务,默认TCP协议)int listfd = socket(AF_INET, SOCK_STREAM, 0);if (-1 == listfd){perror("scoket error\n");return 1;}// 定义服务器和客户端的地址结构体struct sockaddr_in ser, cli;bzero(&ser, sizeof(ser));bzero(&cli, sizeof(cli));ser.sin_family = AF_INET; // 使用IPv4地址族ser.sin_port = htons(50000); // 设置端口号ser.sin_addr.s_addr = INADDR_ANY; // 绑定到所有可用网络接口// 将监听套接字与服务器地址绑定int ret = bind(listfd, (SA)&ser, sizeof(ser));if (-1 == ret){perror("bind");return 1;}// 开始监听listen(listfd, 3);socklen_t len = sizeof(cli);// 1. 创建epoll实例struct epoll_event rev[10]; // 用于存放epoll_wait返回的就绪事件int epfd = epoll_create(10);if (-1 == epfd){perror("epoll_creaete"); // 原代码拼写错误return 1;}// 2. 将监听套接字添加到epoll实例中add_fd(epfd, listfd);time_t tm;while (1){// 等待epoll实例中的事件就绪int ep_ret = epoll_wait(epfd, rev, 10, -1);// 遍历所有就绪事件for (int i = 0; i < ep_ret; i++){// 如果是监听套接字就绪,表示有新的客户端连接请求if (rev[i].data.fd == listfd){// 接受客户端连接int conn = accept(listfd, (SA)&cli, &len);if (-1 == conn){perror("accept");close(conn);continue;}// 将新的连接套接字添加到epoll实例中add_fd(epfd, conn);}else // 其他就绪事件,为客户端连接套接字的读事件{int conn = rev[i].data.fd;char buf[1024] = {0};// 从连接套接字接收数据int ret = recv(conn, buf, sizeof(buf), 0);if (ret <= 0){del_fd(epfd, conn);close(conn);continue;}// 获取当前时间戳time(&tm);// 将接收的数据与当前时间拼接sprintf(buf, "%s %s", buf, ctime(&tm));// 将拼接后的数据发送回客户端send(conn, buf, strlen(buf), 0);}}}close(listfd);return 0;
}
TCP客户端 (cli.c)
#include <netinet/in.h> // 定义网络地址结构
#include <netinet/ip.h> // IP协议相关定义
#include <stdio.h> // 标准输入输出函数
#include <stdlib.h> // 标准库函数
#include <string.h> // 字符串处理函数
#include <sys/socket.h> // 套接字相关函数
#include <sys/types.h> // 基本系统数据类型
#include <unistd.h> // 系统调用函数typedef struct sockaddr *(SA);int main(int argc, char **argv)
{// 创建TCP套接字int conn = socket(AF_INET, SOCK_STREAM, 0);if (-1 == conn){perror("socket");return 1;}// 定义并初始化服务器地址结构struct sockaddr_in ser;bzero(&ser, sizeof(ser));// 设置服务器地址信息ser.sin_family = AF_INET; // 使用IPv4协议ser.sin_port = htons(50000); // 设置目标端口为50000ser.sin_addr.s_addr = INADDR_ANY; // 绑定到本机所有可用IP地址// 尝试与目标服务器建立TCP连接int ret = connect(conn, (SA)&ser, sizeof(ser));if (-1 == ret){perror("connect error\n");return 1;}// 设置循环计数器,控制发送数据的次数(共10次)int i = 10;while (i){// 定义缓冲区并初始化要发送的TCP测试数据char buf[1024] = "hello,this is tcp test";// 向服务器发送数据send(conn, buf, strlen(buf), 0);// 清空缓冲区,准备接收服务器返回的数据bzero(buf, sizeof(buf));// 接收服务器响应数据recv(conn, buf, sizeof(buf), 0);// 打印从服务器接收到的内容printf("from ser:%s\n", buf);sleep(1); // 暂停1秒i--; // 计数器递减}// 关闭套接字close(conn);return 0;
}
TCP通信理想结果:
- 服务器启动后等待连接
- 客户端连接成功后:
from ser:hello,this is tcp test Mon Jun 10 12:30:45 2023
- 服务器返回拼接了时间戳的原始数据
核心回顾
多路IO复用
- 定义:系统提供的IO事件通知机制
- 应用场景:单进程中需要处理多个阻塞IO,希望及时知道哪个设备可读写
IO模型对比
模型 | 核心特点 | 适用场景 |
---|---|---|
阻塞IO | 默认模式,进程挂起等待数据就绪 | 简单场景,单事件流 |
非阻塞IO | 设置O_NONBLOCK ,立即返回+EAGAIN | 需避免阻塞但可接受忙等 |
信号驱动IO | O_ASYNC +SIGIO 信号通知 | 特殊场景(较少使用) |
并发模型 | 进程/线程处理独立事件流 | 中小规模并发 |
多路复用 | 单线程监控多个描述符(select/epoll) | 高并发场景 |
select vs epoll 关键区别
-
描述符数量:
- select:固定上限(通常1024)
- epoll:仅受系统限制(可支持10万+连接)
-
检测机制:
- select:轮询所有描述符(O(n))
- epoll:事件驱动主动上报(O(1))
-
就绪描述符查找:
- select:需遍历原始集合(含未就绪描述符)
- epoll:直接返回就绪事件列表
-
数据拷贝:
- select:每次调用全量拷贝用户/内核空间
- epoll:共享内存(仅首次拷贝)
核心价值:在高并发场景下,epoll通过避免轮询和减少数据拷贝,显著降低CPU开销,成为现代高性能服务器(如Nginx)的基石。