IO多路复用

IO多路复用

---

什么是IO多用复路

IO多路复用（Input/Output Multiplexing）是一种在单个线程中管理多个输入/输出通道的技术。它允许一个线程同时监听多个输入流（例如网络套接字、文件描述符等），并在有数据可读或可写时进行相应的处理，而不需要为每个通道创建一个独立的线程。== 没有文件句柄就绪时会阻塞应用程序，交出cpu。多路是指网络连接，复用指的是同一个线程

💡 IO多路复用的优点是可以有效地管理大量的IO通道，减少线程的创建和销毁开销，提高系统的并发性能。它在网络编程中特别有用，可以用于实现高性能的服务器和客户端应用程序。

为什么有IO多路复用机制？

没有IO多路复用机制时，有BIO、NIO两种实现方式，但有一些问题

同步阻塞（BIO）

服务端采用单线程，当accept一个请求后，在recv或send调用阻塞时，将无法accept其他请求（必须等上一个请求处recv或send完），无法处理并发

服务器端采用多线程，当accept一个请求后，开启线程进行recv，可以完成并发处理，但随着请求数增加需要增加系统线程，大量的线程占用很大的内存空间，并且线程切换会带来很大的开销，10000个线程真正发生读写事件的线程数不会超过20%，每次accept都开一个线程也是一种资源浪费

详细请看上一章服务器的调度策略

TCP协议 - 惠hiuj - 博客园

同步非阻塞（NIO）

服务器端当accept一个请求后，加入fds集合，每次轮询一遍fds集合recv(非阻塞)数据，没有数据则立即返回错误，每次轮询所有fd（包括没有发生读写事件的fd）会很浪费cpu

IO多用复路

服务器端采用单线程通过select/epoll等系统调用获取fd列表，遍历有事件的fd进行accept/recv/send，使其能支持更多的并发连接请求

---

select函数接口的使用

select Linux内核监测这些套接字状态，读状态，写状态
检测的文件描述符并不是客户端socket的文件描述符
而是连接成功后服务器的accept的文件描述符
accept 的返回值，是服务器的内核维护的
服务器的应用程序并不知道，是那个客户端发送的
只有Linux内核知道，recv是将内核缓冲区拷贝到用户程序的buf中
读就绪状态：可以读
未就绪状态
写就绪状态：封完包状态，放在写的内核缓冲区
内核会判断，内核缓冲区的读写状态，状态达成
才会解除 read/write 的堵塞状态
一直用select监听状态的变化
当select返回值时，会清空集合没有处于就绪态的套接字，只保留需要重新，将文件描述符，加入集合

如果需要监听的数量很多，可以将文件的描述符，的集合存储在（在创建集合前将文件放到）数据结构中，然后遍历数据结构，检测是否还留到集合中，对文件做处理

套接字监听的数量是固定的（1024）

---

#include <sys/select.h>
#include <sys/time.h>
#include <unistd.h>int select(int nfds, fd_set *readfds, fd_set *writefds, fd_set *exceptfds, struct timeval *timeout);

参数说明

• nfds: 监控的文件描述符范围是从 0 到 nfds-1。通常设置为所监控的文件描述符中最大值加 1。
• readfds: 一个 fd_set 结构体，用于监控可读的文件描述符集合。如果不需要监控读操作，可以传入 NULL。
• writefds: 一个 fd_set 结构体，用于监控可写的文件描述符集合。如果不需要监控写操作，可以传入 NULL。
• exceptfds: 一个 fd_set 结构体，用于监控异常条件（如带外数据）的文件描述符集合。如果不需要监控异常情况，可以传入 NULL。
• timeout: 一个 timeval 结构体，指定 select 函数的超时时间。如果传入 NULL，则 select 将永远阻塞，直到某个文件描述符准备就绪。如果设置超时时间为 0，则 select 会立即返回。

返回值

• 返回值为正数：表示准备就绪的文件描述符数量。
• 返回值为 0：表示在指定的超时时间内没有文件描述符准备就绪。
• 返回值为 -1：表示出现错误，同时设置 errno 以指示错误类型。

fd_set 的操作

fd_set 是一个位集合，用于存储文件描述符。常用的宏和函数包括：

• FD_ZERO(fd_set *set): 清空集合。
• FD_SET(int fd, fd_set *set): 将文件描述符 fd 加入集合。
• FD_CLR(int fd, fd_set *set): 将文件描述符 fd 从集合中移除。
• FD_ISSET(int fd, fd_set *set): 检查文件描述符 fd 是否在集合中。

select缺点

• 单个进程所打开的FD是有限制的，通过FD_SETSIZE设置，默认1024
• 每次调用select，都需要把fd集合从用户态拷贝到内核态，这个开销在fd很多时会很大
• 对socket扫描时是线性扫描，采用轮询的方法，效率较低（高并发时）

---

poll函数接口

poll与select相比，只是没有fd的限制，其它基本一样

poll 函数是另一个用于监控多个文件描述符的系统调用，与 select 类似，但在某些方面更灵活和高效。poll 函数的接口在 C 语言中的定义如下：

#include <poll.h>
// 数据结构
struct pollfd {int fd;                         // 需要监视的文件描述符short events;                   // 需要内核监视的事件short revents;                  // 实际发生的事件
};// API
int poll(struct pollfd fds[], nfds_t nfds, int timeout);

参数说明

• fds: 一个指向 pollfd 结构体数组的指针，每个 pollfd 结构体描述一个要监控的文件描述符及其事件。
• nfds: fds 数组中的元素个数。
• timeout: 超时时间（以毫秒为单位）。如果 timeout 为负数，poll 将永远阻塞直到有文件描述符准备就绪。如果 timeout 为 0，poll 会立即返回,即不堵塞。

events 和 revents 字段的事件掩码

常用的事件掩码包括：

• POLLIN: 有数据可读。
• POLLRDNORM: 普通数据可读。
• POLLRDBAND: 优先数据可读。
• POLLPRI: 有紧急数据可读。
• POLLOUT: 可以写数据。
• POLLWRNORM: 普通数据可写。
• POLLWRBAND: 优先数据可写。
• POLLERR: 发生错误。
• POLLHUP: 挂起。
• POLLNVAL: 描述符非法。

返回值

• 返回值为正数：表示准备就绪的文件描述符数量。
• 返回值为 0：表示在指定的超时时间内没有文件描述符准备就绪。
• 返回值为 -1：表示出现错误，同时设置 errno 以指示错误类型。

在使用 poll 函数时，pollfd 结构体中的 events 和 revents 字段分别表示要监控的事件和实际发生的事件。它们的具体区别如下：

events 字段

events 字段用于指定你希望 poll 函数监控的事件类型。你可以通过设置 events 字段来告诉 poll 函数你对哪些事件感兴趣。常用的事件掩码包括：

• POLLIN: 有数据可读。
• POLLRDNORM: 普通数据可读。
• POLLRDBAND: 优先数据可读。
• POLLPRI: 有紧急数据可读。
• POLLOUT: 可以写数据。
• POLLWRNORM: 普通数据可写。
• POLLWRBAND: 优先数据可写。
• POLLERR: 发生错误。
• POLLHUP: 挂起。
• POLLNVAL: 描述符非法。

例如，如果你希望监控一个文件描述符是否有数据可读，你可以设置 events 字段为 POLLIN。

revents 字段

revents 字段用于指示 poll 函数返回时实际发生的事件类型。poll 函数会在返回时将 revents 字段设置为实际发生的事件的掩码。你可以通过检查 revents 字段来确定哪些事件在调用 poll 函数期间发生了。

例如，如果 poll 函数返回时 revents 字段包含 POLLIN，这意味着在调用 poll 函数期间，有数据可读。

---

多客户端聊天服务器

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>#define PORT 12345
#define MAX_CLIENTS 10
#define BUFFER_SIZE 1024void error(const char *msg) {perror(msg);exit(1);
}int main() {int server_fd, new_socket, client_socket[MAX_CLIENTS], max_sd, sd, activity, valread;int opt = 1;struct sockaddr_in address;fd_set readfds;char buffer[BUFFER_SIZE];// 初始化所有客户端套接字为0（未使用）for (int i = 0; i < MAX_CLIENTS; i++) {client_socket[i] = 0;}// 创建服务器套接字if ((server_fd = socket(AF_INET, SOCK_STREAM, 0)) == 0) {error("socket failed");}// 设置服务器套接字选项if (setsockopt(server_fd, SOL_SOCKET, SO_REUSEADDR, (char *)&opt, sizeof(opt)) < 0) {error("setsockopt");}// 配置服务器地址address.sin_family = AF_INET;address.sin_addr.s_addr = INADDR_ANY;address.sin_port = htons(PORT);// 绑定套接字到端口if (bind(server_fd, (struct sockaddr *)&address, sizeof(address)) < 0) {error("bind failed");}// 监听连接if (listen(server_fd, 3) < 0) {error("listen");}printf("Listening on port %d \n", PORT);while (1) {// 清空读文件描述符集合FD_ZERO(&readfds);// 将服务器套接字加入集合FD_SET(server_fd, &readfds);max_sd = server_fd;// 添加客户端套接字到集合for (int i = 0; i < MAX_CLIENTS; i++) {sd = client_socket[i];if (sd > 0) {FD_SET(sd, &readfds);}if (sd > max_sd) {max_sd = sd;}}// 等待活动activity = select(max_sd + 1, &readfds, NULL, NULL, NULL);// 检查是否有新的连接if (FD_ISSET(server_fd, &readfds)) {int addrlen = sizeof(address);if ((new_socket = accept(server_fd, (struct sockaddr *)&address, (socklen_t*)&addrlen)) < 0) {error("accept");}printf("New connection, socket fd is %d, ip is: %s, port: %d\n", new_socket, inet_ntoa(address.sin_addr), ntohs(address.sin_port));// 将新套接字加入客户端数组for (int i = 0; i < MAX_CLIENTS; i++) {if (client_socket[i] == 0) {client_socket[i] = new_socket;printf("Adding to list of sockets as %d\n", i);break;}}}// 检查客户端套接字是否有数据for (int i = 0; i < MAX_CLIENTS; i++) {sd = client_socket[i];if (FD_ISSET(sd, &readfds)) {// 检查是否是关闭连接if ((valread = read(sd, buffer, BUFFER_SIZE)) == 0) {printf("Host disconnected, ip %s, port %d\n", inet_ntoa(address.sin_addr), ntohs(address.sin_port));// 关闭套接字并从集合中移除close(sd);client_socket[i] = 0;} else {// 处理客户端发送的数据buffer[valread] = '\0';printf("Received message: %s\n", buffer);// 广播给其他客户端for (int j = 0; j < MAX_CLIENTS; j++) {if (client_socket[j] != 0 && client_socket[j] != sd) {send(client_socket[j], buffer, strlen(buffer), 0);}}}}}}return 0;
}