当前位置：首页 > news >正文

Linux笔记---非阻塞IO与多路复用

news 2025/9/18 12:27:14

1. 五种IO模型

在 UNIX/Linux 系统中，IO 模型是处理输入输出操作的核心机制，主要分为以下五种：阻塞 IO、非阻塞 IO、IO 多路复用、信号驱动 IO、异步 IO。它们的核心区别在于 “等待数据准备” 和 “数据复制到用户空间” 两个阶段的处理方式不同。

1.1 阻塞IO（Blocking IO）

最基础的 IO 模型，进程执行 IO 操作时会全程阻塞，直到数据准备完成并复制到用户空间后才返回。

进程发起 IO 请求（如recvfrom）；
内核开始准备数据（如等待网络数据到达），此阶段进程阻塞；
数据准备完成后，内核将数据从内核空间复制到用户空间，此阶段进程继续阻塞；
复制完成，IO 调用返回，进程恢复运行。

优点：实现简单，无需额外处理；
缺点：进程阻塞期间无法做其他事，效率低，不适用于高并发场景；
适用场景：简单小程序（如单机小工具），或并发量极低的场景。

1.2 非阻塞IO（Non-blocking IO）

进程发起 IO 请求后，若数据未准备好，内核会立即返回错误（而非阻塞）；进程可不断轮询（重复发起请求），直到数据准备好并完成复制。

进程发起 IO 请求，若数据未准备好，内核返回EWOULDBLOCK错误，进程不阻塞；
进程可做其他事，之后主动轮询再次发起 IO 请求；
当数据准备好后，内核将数据复制到用户空间（此阶段进程阻塞）；
复制完成，IO 调用返回，进程处理数据。

优点：进程不阻塞在等待数据阶段，可并行处理其他任务；
缺点：轮询会消耗大量 CPU 资源，效率不高；
适用场景：对响应速度要求极高，且并发量小的场景（实际中很少单独使用）。

1.3 IO 多路复用/多路转接（IO Multiplexing）

通过一个监控进程（如select/poll/epoll）同时监控多个 IO 流，当某个 IO 流的数据准备好时，再通知进程处理。进程阻塞在 “监控” 操作上，而非直接阻塞在 IO 请求上。

进程创建监控器（如epoll_create），并将需要监控的 IO 流注册到监控器；
进程调用监控接口（如epoll_wait），进入阻塞状态（等待任一 IO 流就绪）；
当某个 IO 流数据准备好，内核通知监控器，监控接口返回就绪的 IO 流；
进程针对就绪的 IO 流发起 IO 请求，内核将数据复制到用户空间（此阶段进程阻塞）；
复制完成，进程处理数据。

优点：单进程可高效管理多个 IO 流，适合高并发（如服务器同时处理多个客户端的连接）；
缺点：数据复制阶段仍会阻塞，且监控器本身有一定开销（但远小于轮询）；
适用场景：高并发网络服务器（如 Nginx、Redis），是实际中最常用的模型之一。

1.4 信号驱动 IO（Signal-driven IO）

进程通过信号机制被动等待通知：先注册信号处理函数，内核在数据准备好时主动发送 SIGIO 信号，进程收到信号后再处理 IO。

进程注册 SIGIO 信号的处理函数，并告知内核需要监控的 IO 流；
进程不阻塞，可正常执行其他任务；
当数据准备好，内核发送 SIGIO 信号给进程；
进程捕获信号，在信号处理函数中发起 IO 请求，内核将数据复制到用户空间（此阶段进程阻塞）；
复制完成，进程处理数据。

优点：无需轮询，等待阶段不阻塞；
缺点：信号处理逻辑复杂（如信号队列溢出、多信号竞争），实际中很少使用；
适用场景：特殊场景（如某些网络设备驱动），极少用于通用高并发程序。

1.5 异步 IO（Asynchronous IO）

最 “彻底” 的异步模型：进程发起 IO 请求后立即返回，内核会完成 “数据准备” 和 “复制到用户空间” 的全部操作，完成后再通知进程。

进程发起异步 IO 请求（如aio_read），并指定操作完成后的通知方式（如信号或回调）；
进程立即返回，可继续执行其他任务（全程不阻塞）；
内核自动完成数据准备和复制到用户空间的所有操作；
操作全部完成后，内核通过信号或回调通知进程；
进程收到通知后，直接处理用户空间中的数据。

优点：全程无阻塞，理论上效率最高；
缺点：实现复杂，内核支持有限（如 Linux 的io_uring是较新的高效实现）。
适用场景：对性能要求极致的高并发场景（如高性能数据库、分布式存储）。

2. 非阻塞IO

要使用非阻塞IO的方式进行数据的读写非常简单，基本上所有与读写有关的系统调用都会提供一个flag参数，通过对这个参数进行设置，就可以使得该系统调用变为非阻塞式IO。

这里，我们就不对这些接口的flag参数进行详细介绍了，毕竟使用man就可以直接查询。

我们要介绍的是一个通用的设置非阻塞IO的方式，即直接对文件描述符的属性进行设置。

为什么这个方式是通用的呢？因为Linux当中“一切皆文件”。

2.1 fcntl系统调用

在 UNIX/Linux 系统中，fcntl（file control）是一个功能强大的系统调用，用于控制已打开文件描述符的各种属性和行为。

它支持多种操作，包括设置非阻塞模式、管理文件锁、修改文件状态标志等，是系统编程中处理文件和 IO 的核心接口之一。

#include <unistd.h>
#include <fcntl.h>int fcntl(int fd, int cmd, ... /* arg */ );

参数：

fd：需要操作的文件描述符（已通过open/socket等函数打开）。
cmd：操作命令（决定fcntl的行为，如获取 / 设置属性、加锁等）。
常见操作：
- F_GETFL：获取文件状态标志。返回值为当前的状态标志（如O_RDONLY、O_WRONLY、O_NONBLOCK等）。
- F_SETFL：设置文件状态标志。通过arg参数传入新的状态标志（只能修改部分标志，如O_NONBLOCK、O_APPEND等，不可修改读写模式）。
- F_DUPFD：复制文件描述符，返回一个大于等于arg的新描述符（与dup类似，但可指定最小描述符值）。
可选参数arg：根据cmd的不同，可能是整数或结构体（如struct flock）。

返回值：

成功：返回值取决于cmd（如文件状态标志、锁状态等）。
失败：返回-1，并设置errno（如EBADF表示文件描述符无效）。

2.2 设置非阻塞

void SetNoBlock(int fd) 
{// 先获取原本的文件描述符属性int fl = fcntl(fd, F_GETFL); if (fl < 0) { perror("fcntl");return;}// 再将非阻塞属性设置进去fcntl(fd, F_SETFL, fl | O_NONBLOCK); 
}

3. select系统调用实现多路复用

3.1 select系统调用

在 UNIX/Linux 系统中，select是IO 多路复用的经典系统调用，允许进程同时监控多个文件描述符（file descriptor），等待其中一个或多个处于 “就绪” 状态（如可读、可写或发生异常），从而高效处理多 IO 流场景（如同时管理多个网络连接）。

#include <sys/select.h>int select(int nfds, fd_set *readfds, fd_set *writefds, fd_set *exceptfds, struct timeval *timeout);// fd_set为一个位图结构, 在这里用于表示我们希望select帮助我们监控的fd
// 我们需要使用下面的几个宏来对其进行操作
void FD_CLR(int fd, fd_set *set);
int  FD_ISSET(int fd, fd_set *set);
void FD_SET(int fd, fd_set *set);
void FD_ZERO(fd_set *set);

参数：

nfds：需要监控的 “最大文件描述符 + 1”。内核会从 0 到nfds-1遍历检查文件描述符，因此必须正确设置（否则可能遗漏监控）。
readfds：可读事件监控集合。若某个文件描述符在此集合中，且内核检测到其 “可读”（如收到数据），则该描述符会被标记为就绪（直接修改readfds）。
writefds：可写事件监控集合。若某个文件描述符在此集合中，且内核检测到其 “可写”（如发送缓冲区有空间），则该描述符会被标记为就绪。
exceptfds：异常事件监控集合。用于检测文件描述符的异常状态（如带外数据到达）。
timeout：超时设置（struct timeval类型），决定select的阻塞行为：
- NULL：无限阻塞，直到有事件就绪才返回。
- tv_sec=0且tv_usec=0：非阻塞，立即返回（不管是否有事件就绪）。
- 其他值：阻塞等待指定时间（秒 + 微秒），超时后返回。
```
struct timeval {long tv_sec;  // 秒long tv_usec; // 微秒（1秒=1e6微秒）
};
```

注意：除了第一个参数以外，其余参数都是输入/输出型参数。在调用结束之后，fd_set类型的变量会被设置为事件（读/写）就绪的文件描述符集合；timeout中的值为距离超时剩余的事件（超时返回则为0）。

返回值：

成功：返回就绪的文件描述符总数（可读、可写、异常事件的总和）。
超时（未检测到任何就绪事件）：0。
出错（如被信号中断）：-1，并设置errno。

3.2 用select实现一个多路复用的网络服务

下面的代码是结合了博主自己封装的Socket来写的，仅供借鉴，完整代码参考：https://gitee.com/da-guan-mu-lao-sheng/linux-c/tree/master/%E5%A4%9A%E8%B7%AF%E8%BD%AC%E6%8E%A5/Select

#pragma once
#include <sys/select.h>
#include "Common.hpp"
#include "Socket.hpp"
using namespace SocketModule;class SelectServer : public NoCopy
{
private:void ListenSocketHandler(){std::shared_ptr<TCPConnectSocket> connect_socket = _listen_socket->Accept();int connect_sockfd = connect_socket->SockFd();_fds[connect_sockfd] = connect_socket;}void ConnectSocketHandler(int fd){std::string message;int n = _fds[fd]->Receive(message);if(n > 0){// 正常收到消息, 回显std::cout << "Client[" << _fds[fd]->Addr().Info() << "] say# " << message << std::endl;}else if(n == 0){// 客户端断开连接LOG(LogLevel::INFO) << "Client[" << _fds[fd]->Addr().Info() << "]已断开连接...";_fds[fd] = _default_socket_ptr;}else{// 出错LOG(LogLevel::ERROR) << "Receive: 接收Client[" << _fds[fd]->Addr().Info() << "]的数据失败! ";}}void Dispatch(fd_set *readfds){for (int fd = 0; fd < _fds.size(); fd++){if (!_fds[fd] || !FD_ISSET(fd, readfds))continue;if (fd == _listen_socket->SockFd()){// 监听套接字ListenSocketHandler();}else{// 连接套接字ConnectSocketHandler(fd);}}}public:SelectServer(in_port_t port): _isrunning(false), _listen_socket(std::make_shared<TCPListenSocket>(port)), _fds(sizeof(fd_set) * 8, _default_socket_ptr){int listen_sockfd = _listen_socket->SockFd();_fds[listen_sockfd] = _listen_socket;}void Run(){_isrunning = true;while (_isrunning){fd_set readfds;FD_ZERO(&readfds);int max_fd = _listen_socket->SockFd();for (int fd = 0; fd < _fds.size(); fd++){if (_fds[fd]){FD_SET(fd, &readfds);max_fd = std::max(max_fd, fd);}}struct timeval timeout = {10, 0};int n = select(max_fd + 1, &readfds, nullptr, nullptr, &timeout);if (n > 0){// 有事件就绪LOG(LogLevel::INFO) << "有事件就绪, 即将处理...";Dispatch(&readfds);}else if (n == 0){// 超时LOG(LogLevel::DEBUG) << "超时处理...";}else{// 出错LOG(LogLevel::FATAL) << "select: " << strerror(errno);break;}}_isrunning = false;}~SelectServer() {}private:bool _isrunning;std::shared_ptr<TCPListenSocket> _listen_socket; // 监听套接字std::vector<std::shared_ptr<Socket>> _fds;       // 需要select等待的套接字static const std::shared_ptr<Socket> _default_socket_ptr;
};
const std::shared_ptr<Socket> SelectServer::_default_socket_ptr = nullptr;

上面的代码就可以在只有单个执行流的情况下，并发地处理多个客户端发来的请求。

3.3 优缺点

优点：

跨平台支持：几乎所有 UNIX/Linux 系统（及 Windows）都支持，兼容性好。
功能完整：可同时监控可读、可写、异常三类事件。

缺点：

文件描述符数量限制： fd_set大小受FD_SETSIZE（通常为 1024）限制，默认最多监控 1024 个文件描述符（需重新编译内核才能修改，成本高）。
效率随 FD 数量下降：每次调用select，内核需遍历所有监控的 FD（O (n) 复杂度），当 FD 数量庞大时（如 10 万级），效率极低。
集合需重复初始化： select返回后会修改fd_set（只保留就绪的 FD），下次调用前必须重新用FD_ZERO和FD_SET重置，操作繁琐。
内核 / 用户空间复制开销：每次调用select，fd_set需从用户空间复制到内核空间，FD 数量越多，复制开销越大。

select适合FD 数量较少的场景（如几百个以内），例如简单的多客户端网络服务器。但在高并发场景（如需要支持上万连接），已被更高效的epoll（Linux）、kqueue（BSD/macOS）等替代。

4. poll系统调用实现多路复用

4.1 poll系统调用

在 Linux 系统中，poll 也是一个用于 I/O 多路复用的系统调用。

#include <poll.h>int poll(struct pollfd *fds, nfds_t nfds, int timeout);

参数：

fds：指向 struct pollfd 结构体数组的指针，每个结构体描述一个要监视的文件描述符及其关注的事件
nfds：要监视的文件描述符数量
timeout：超时时间（毫秒），-1 表示无限等待，0 表示立即返回

struct pollfd结构体：

struct pollfd {int   fd;         // 要监视的文件描述符，-1 表示忽略此结构体short events;     // 要监视的事件（输入参数）short revents;    // 实际发生的事件（输出参数）
};

返回值：

成功：等待成功的事件个数（超时返回0）；
失败：返回-1，并设置错误码。

4.2 用poll实现一个多路复用的网络服务

完整代码参考：https://gitee.com/da-guan-mu-lao-sheng/linux-c/tree/master/%E5%A4%9A%E8%B7%AF%E8%BD%AC%E6%8E%A5/Poll

#pragma once
#include <poll.h>
#include <list>
#include <unordered_map>
#include "Common.hpp"
#include "Socket.hpp"
using namespace SocketModule;class PollServer : public NoCopy
{
private:void ListenSocketHandler(){std::shared_ptr<TCPConnectSocket> connect_socket = _listen_socket->Accept();if (!connect_socket){ // 检查Accept是否成功LOG(LogLevel::ERROR) << "Accept failed";return;}int connect_sockfd = connect_socket->SockFd();RegisterFd(connect_sockfd, POLLIN);_connect_socket[connect_sockfd] = connect_socket;}void ConnectSocketHandler(int fd){auto it = _connect_socket.find(fd);if (it == _connect_socket.end()){LOG(LogLevel::ERROR) << "Invalid socket fd: " << fd;return;}std::string message;int n = it->second->Receive(message);std::string client = it->second->Addr().Info();if (n > 0){// 正常收到消息std::cout << "Client[" << client << "] say# " << message << std::endl;}else if (n == 0){// 客户端断开连接LOG(LogLevel::INFO) << "Client[" << client << "]已断开连接...";_connect_socket.erase(it);DeleteFd(fd);}else{// 出错LOG(LogLevel::ERROR) << "Receive: 接收Client[" << client << "]的数据失败! ";_connect_socket.erase(it);DeleteFd(fd);}}void Dispatch(){for (auto &poll : _fds){// 跳过无效的fd或没有事件的fdif (poll.fd == _default_socket_fd || !(poll.revents & POLLIN))continue;if (poll.fd == _listen_socket->SockFd()){// 监听套接字ListenSocketHandler();}else{// 连接套接字ConnectSocketHandler(poll.fd);}}}void RegisterFd(int fd, short events){// 尝试找到一个空闲位置for (auto &poll : _fds){if (poll.fd == _default_socket_fd){poll.fd = fd;poll.events = events;return;}}// 如果没有空闲位置，则添加新元素_fds.push_back((pollfd){fd, events, 0});}void DeleteFd(int fd){for (auto &poll : _fds){if (poll.fd == fd){poll.fd = _default_socket_fd;poll.events = 0;poll.revents = 0;break;}}}public:PollServer(in_port_t port): _isrunning(false), _listen_socket(std::make_shared<TCPListenSocket>(port)){if (!_listen_socket || _listen_socket->SockFd() < 0){LOG(LogLevel::FATAL) << "PollServer: 初始化监听套接字失败! ";exit(EXIT_FAILURE);}int listen_sockfd = _listen_socket->SockFd();RegisterFd(listen_sockfd, POLLIN);}void Run(){_isrunning = true;while (_isrunning){// 使用实际有效的fd数量int n = poll(_fds.data(), _fds.size(), 10000);if (n < 0){// 处理错误，EINTR是可恢复的if (errno == EINTR)continue;LOG(LogLevel::FATAL) << "poll error: " << strerror(errno);break;}else if (n == 0){// 超时LOG(LogLevel::DEBUG) << "poll timeout...";}else{// 有事件就绪LOG(LogLevel::INFO) << "有" << n << "个事件就绪, 即将处理...";Dispatch();}}_isrunning = false;}~PollServer(){_isrunning = false;}private:bool _isrunning;std::shared_ptr<TCPListenSocket> _listen_socket;                            // 监听套接字std::unordered_map<int, std::shared_ptr<TCPConnectSocket>> _connect_socket; // 需要poll等待的连接套接字std::vector<pollfd> _fds;                                                   // poll等待列表static const int _default_socket_fd;
};const int PollServer::_default_socket_fd = -1;