C++网络编程：使用select实现IO多路复用
- 一、什么是 IO 多路复用？
- 二、IO多路复用器 select
- 三、相关接口
- 3.1、fd_set 结构体
- 3.2、宏和函数
- 四、select 实现 TCP 服务器
- 五、总结
一、什么是 IO 多路复用？
在网络编程中，最容易想到的并发模型就是“一请求一线程”模型，逻辑非常容易理解。但是，“一请求一线程”模型对资源消耗非常高，高并发下很容易就达到了瓶颈；那么单线程可不可以实现高并发连接呢？当然是可行的，那就是IO多路复用。
IO多路复用 (I/O Multiplexing) 是一种允许多个 I/O 流共享同一个线程的技术。它通过一个机制，让单线程可以同时监听多个文件描述符（比如网络连接、文件、管道等），一旦某个描述符就绪（例如可以读写数据），系统就通知该线程，从而避免了线程阻塞在单个 I/O 操作上。
想象一下一个电话接线员，他可以同时接听多个电话线，当某个电话响了，他就去接听那个电话。这就是 IO 多路复用的精髓。 不像传统的阻塞式 I/O，每个电话线都需要一个单独的接线员去监听，IO 多路复用只用一个“接线员”就能高效地处理多个“电话”。
关键概念：
- 单线程处理多个连接， 这是 IO 多路复用的核心优势。
- 事件驱动: IO 多路复用依赖于操作系统提供的事件通知机制。当某个文件描述符就绪时，操作系统会通知应用程序，应用程序再根据事件处理相应的 I/O 操作。
- 非阻塞 I/O: 通常与非阻塞 I/O 结合使用。非阻塞 I/O 调用不会阻塞线程，而是立即返回，即使 I/O 操作未完成。
- 负责监听多个文件描述符的事件。
常用实现方式： select、poll、epoll (Linux)、kqueue (BSD)。
优势：
- 高效率: 单线程处理多个连接，减少了上下文切换开销。
- 高并发: 可以处理大量的并发连接。
- 资源消耗低: 相比多线程/多进程模型，资源消耗更低。
劣势：
- 相比传统的阻塞式 I/O，编程复杂度略高。
- 不同的操作系统有不同的 IO 多路复用实现方式。
二、IO多路复用器 select
本文先跟大家介绍一种最常见的IO多路复用机制：select。用于监视多个文件描述符，以便在某个或某些文件描述符就绪时进行相应的处理。
select 函数的基本原型如下：
int select(int nfds, fd_set *readfds, fd_set *writefds, fd_set *exceptfds, struct timeval *timeout);
参数说明：
- nfds: 监视的文件描述符数量（最大值 + 1）。
- readfds: 监视可读的文件描述符集合（读事件）。
- writefds: 监视可写的文件描述符集合（写事件）。
- exceptfds: 监视异常条件的文件描述符集合。
-
timeout: 指定等待事件发生的最大时间。如果设置为
NULL，则select会无限期等待。
select的特点：
-
select是一个 POSIX 标准函数，多数 UNIX/Linux 系统和 Windows 系统都支持select。 -
select有一个文件描述符数目的限制，通常是 1024（取决于实现和系统配置）。 - 虽然
select可以处理多个文件描述符，但其性能在文件描述符数量较多时会降低，因为它需要线性扫描所有文件描述符集合来检测哪个文件描述符处于就绪状态。
从select的第一个参数我们可以看得出，select内部实现是通过循环遍历所有的fd来确定每个fd的状态。所有，设置select的第一个参数时一定要比实际使用的fd还要大。
三、相关接口
在介绍接口之前，我们可以想一个问题：如何去标识一个 IO 的事件？事件一般只有两个状态：有与没有。IO 在Linux中是一个int类型的值。针对事件的两个状态，可以用一个bit标识，比如char类型有8bit，就可以用来标识8个 IO 的一个事件状态，比如一个 IO 的可读事件有还是没有。这就是接下来要讲的fd_set结构体。
3.1、fd_set 结构体
fd_set 是用于处理 I/O 多路复用的结构体，主要在调用 select、FD_SET、FD_CLR、FD_ISSET 等宏时使用，尤其是在使用 select 函数时。它的主要作用是用来表示一组文件描述符（存储文件描述符的集合），这些文件描述符可以是用于网络套接字、文件或其他 I/O 资源。在 <sys/select.h> 头文件中定义。
fd_set 可以概括为一个可以容纳多个文件描述符的位图（bit vector）。每一个 bit 代表一个文件描述符（通常从 0 开始），如果某个 bit 被设置，表示对应的文件描述符在集合中。
3.2、宏和函数
与 fd_set 一起使用的常见宏和操作包括：
-
FD_ZERO(fd_set *set): 清空fd_set集合。 -
FD_SET(int fd, fd_set *set): 将指定的文件描述符fd添加到fd_set集合中（即将对应比特位设置为 1）。 -
FD_CLR(int fd, fd_set *set): 将指定的文件描述符fd从fd_set集合中移除。 -
FD_ISSET(int fd, fd_set *set): 检查指定的文件描述符fd是否在fd_set集合中。
FD_SET(int fd, fd_set *set)将对应比特位设置以后，通过select函数带到内核里面去，然后看它有没有事件发生。
四、select 实现 TCP 服务器
-
使用
FD_ZERO和FD_SET宏来初始化和设置文件描述符集合。 -
传入相关参数，循环调用
select函数。 -
select返回后，可以通过循环检查文件描述符集合来确定哪些文件描述符就绪，并进行相应的处理。
我们这里把select封装到一个类里面，方便调用。
定义一个类（server.h）：
#include <sys/socket.h>
#include <netinet/in.h>
#include <errno.h>
#include <unistd.h>
#include <fcntl.h>
#include <iostream>
#include <cstring>
#include <string>
#define BUFFER_LEN 4096
class TcpServerSelect {
enum IoMode { Blocking, NonBlocking };
public:
TcpServerSelect() : _port(8080), _listenBlock(20), _listenfd(-1),
_ioMode(Blocking), _finished(false) {}
int initializer();
void run();
void setMode(int mode) { _ioMode = mode; }
void setBlock(int num) { _listenBlock = num; }
void setPort(short port) { _port = port; }
void setFinished(bool finished) { _finished = finished; }
protected:
bool setIoMode(int fd, int mode);
void acceptConnect(int& maxfd, fd_set& rfds);
void recvData(int clientfd, fd_set& wfds, fd_set& rfds);
void sendData(int clientfd, fd_set& wfds, fd_set& rfds);
protected:
short _port;
int _listenBlock;
int _listenfd;
int _ioMode;
bool _finished;
std::string _strData;
};
功能实现（server.cpp）：
int
TcpServerSelect::initializer()
{
if (_listenfd > 0) {
std::cout << "The listening port already exists. listen fd " << _listenfd << std::endl;
return 0;
}
// 1. Create socket
_listenfd = socket(AF_INET, SOCK_STREAM, 0);
if (_listenfd == -1) {
std::cout << "socket return " << errno << ", " << strerror(errno) << std::endl;
return -1;
}
// 2. Set the port and bind it.
sockaddr_in serverAddr;
memset(&serverAddr, 0, sizeof(sockaddr_in));
serverAddr.sin_family = AF_INET;
serverAddr.sin_addr.s_addr = htons(INADDR_ANY); // bind ip address.
serverAddr.sin_port = htons(_port); // bind port.
if (bind(_listenfd, (sockaddr*)&serverAddr, sizeof(serverAddr)) == -1) {
std::cout << "bind return " << errno << ", " << strerror(errno) << std::endl;
return -2;
}
if (_ioMode == NonBlocking) {
// set nonblock mode.
setIoMode(_listenfd, O_NONBLOCK);
}
// 3. listening port.
if (listen(_listenfd, _listenBlock) == -1) {
std::cout << "listen return " << errno << ", " << strerror(errno) << std::endl;
return -3;
}
std::cout << "server listening port " << _port << std::endl;
return 0;
}
void
TcpServerSelect::run()
{
if (_listenfd < 0) {
std::cout << "Initialization not completed." << std::endl;
return;
}
fd_set wfds, rfds, exceptfds;
FD_ZERO(&wfds);
FD_ZERO(&exceptfds);
FD_ZERO(&rfds);
FD_SET(_listenfd, &rfds);
struct timeval timeout;
// 设置超时时间
timeout.tv_sec = 5; // 5秒
timeout.tv_usec = 0;
// 小技巧：通过中间变量将判断位和修改位分开。
fd_set wset, rset;
// 遍历多少个fd
int maxfd = _listenfd;
while (!_finished) {
// 4. select
wset = wfds;
rset = rfds;
// 注意，这里的参数是使用的rset和wset，将判断位和修改位分开。
int ret = select(maxfd + 1, &rset, &wset, &exceptfds, &timeout);
if (ret == 0)
continue;
if (ret < 0) {
std::cout << "select return error: " << errno << ". " << strerror(errno) << std::endl;
continue;
}
if (FD_ISSET(_listenfd, &rset))
acceptConnect(maxfd, rfds);
for (int i = _listenfd + 1; i <= maxfd; ++i) {
if (FD_ISSET(i, &rset))
recvData(i, wfds, rfds);
else if (FD_ISSET(i, &wset))
sendData(i, wfds, rfds);
}
}
close(_listenfd);
_listenfd = -1;
}
bool
TcpServerSelect::setIoMode(int fd, int mode)
{
int flag = fcntl(fd, F_GETFL, 0);
if (flag == -1) {
std::cout << "fcntl get flags return " << errno << ", " << strerror(errno) << std::endl;
return false;
}
flag |= mode;
if (fcntl(fd, F_SETFL, flag) == -1) {
std::cout << "fcntl set flags return " << errno << ", " << strerror(errno) << std::endl;
return false;
}
return true;
}
void
TcpServerSelect::acceptConnect(int& maxfd, fd_set& rfds)
{
// 4. accept connect.
sockaddr_in clientAddr;
memset(&clientAddr, 0, sizeof(clientAddr));
socklen_t clienLen = sizeof(clientAddr);
int clientfd = accept(_listenfd, (sockaddr *)&clientAddr, &clienLen);
if (clientfd == -1) {
std::cout << "accept return " << errno << ", " << strerror(errno) << std::endl;
return;
}
std::cout << "client fd " << clientfd << std::endl;
FD_SET(clientfd, &rfds);
if (clientfd > maxfd)
maxfd = clientfd;
}
void
TcpServerSelect::recvData(int clientfd, fd_set& wfds, fd_set& rfds)
{
// 6. recv message
char buffer[BUFFER_LEN];
int ret = recv(clientfd, buffer, BUFFER_LEN, 0);
if (ret == 0) {
std::cout << "client " << clientfd << " connection dropped" << std::endl;
close(clientfd);
// clear read ready.
FD_CLR(clientfd, &rfds);
return;
} else if (ret == -1) {
std::cout << "recv buffer return " << errno << ", " << strerror(errno) << std::endl;
return;
}
std::cout << "recv buffer from client "<< clientfd << ": " << buffer << std::endl;
_strData = buffer;
FD_CLR(clientfd, &rfds);
// set fd send data ready.
FD_SET(clientfd, &wfds);
}
void
TcpServerSelect::sendData(int clientfd, fd_set& wfds, fd_set& rfds)
{
if (_strData.empty())
_strData = "Hello, Client!";
// 5. send message.
if (send(clientfd, _strData.c_str(), _strData.size(), 0) == -1) {
std::cout << "send buffer return " << errno << ", " << strerror(errno) << std::endl;
return;
}
// clear write fd
FD_CLR(clientfd, &wfds);
// set fd send data ready.
FD_SET(clientfd, &rfds);
}
代码中使用了这样一段赋值操作：
fd_set wset, rset;
// ......
wset = wfds;
rset = rfds;
// ......
这样做的目的是为了将判断位和修改位分开，避免直接对同一个变量做操作造成问题；但是，一定要捋清楚判断位和修改位的使用，不要混淆。
另外，要注意到，对于listenfd来说，调用accept后缓冲区就清空了，状态就自动转换为非就绪。
使用：
int main(int argc, char**argv)
{
TcpServerSelect server;
if (server.initializer() != 0)
return -1;
server.run();
}
五、总结
IO多路复用的作用是用来检测 IO是否有事件；这里所谓的“事件”就是“可读”、“可写”。IO 多路复用器select可以管理所有的IO，select 在处理大规模文件描述符时存在性能瓶颈（例如，文件描述符数量有限制）。
通过select我们理解了 IO 的可读事件、可写事件。明白了多个客户端访问一个服务器应该怎么做。
