IO多路复用是如何实现高效数据处理的？

本文共计2619个文字，预计阅读时间需要11分钟。

Linux下实现I/O复用的系统调用主要方式：select、poll、epoll。I/O多路复用+Linux下实现I/O复用的系统调用主要方式：select、poll、epoll。select系统调用+select系统调用可在一段指定时间内监听。

Linux下实现I/O复用的系统调用方式主要：select、poll、epoll。 I/O多路复用

Linux下实现I/O复用的系统调用方式主要：select、poll、epoll。

select 系统调用

select系统调用可在一段指定时间内，监听文件描述符上的可读、可写和异常等事件，判断发生的事件需要轮询。

#include <sys/select.h> //select监听文件描述符事件 //nfds： 被监听文件描述符中最大值+1 //readfds： 可读事件对应的文件描述符集，对应位置1；会被内核修改，返回时无事件的置0。 //writefds： 可写事件对应的文件描述符集，对应位置1；会被内核修改，返回时无事件的置0。 //exceptfds：异常事件对应的文件描述符集，对应位置1；会被内核修改，返回时无事件的置0。 //timeout //return： 返回就绪文件描述符中最大值+1 int select(int nfds, fd_set* readfds, fd_set* writefds, fd_set* exceptfds, struct timeval* timeout); //fd_set比特向量操作 FD_ZERO(fd_set *fdset); //清楚fdset中所有比特位 FD_SET(int fd, fd_set *fdset); //设置fdset中比特位fd FD_CLR(int fd, fd_set *fdset); //清除fdset中比特位fd FD_ISSET(int fd, fd_set *fdset); //测试fdset中比特位fd是否被设置，用于判断是否有事件发生 select事件类型

• socket可读

  select可读（readfds）

  • socket内核接收缓存区字节数大于或等于低水位标记SO_RCVLOWAT。读操作返回读取字节数。
  • 读关闭：socket通信的对方关闭连接。读操作返回0。
  • 错误：socket上有未处理的错误。可以使用getsockopt读取和清楚错误。
  • 连接请求：监听socket上有新的连接请求。

• socket可写（writefds）

  • 可写：socket内核发送缓存区中的可用字节数大于或等于低水位标记SO_SNDLOWAT。写操作返回实际写字节数。
  • 写关闭：写操作被关闭，对写关闭的socket执行写操作将触发一个SIGPIPE信号。
  • 错误：socket上有未处理的错误。可以使用getsockopt读取和清楚错误。
  • 连接结果：socket使用非阻塞的connect连接成功或者失败（超时）之后。

• socket异常（exceptfds）

  • 网络程序中，select能处理的异常情况只有一种，即socket上接收到带外数据。

poll 系统调用

poll系统调用与select类似，需要轮询判断监听文件描述符集上的事件，但没有监听文件描述符数量限制。

#include <poll.h> //poll监听文件描述符上指定的事件 //fds： pollfd结构类型的数组 //nfds： pollfd数组长度 //return： 返回就绪文件描述符中最大值+1 int poll(struct pollfd* fds, nfds_t nfds, int timeout); //pollfd结构体，描述文件描述符上的可读、可写、异常等事件 struct pollfd{ int fd; //文件描述符 short events; //注册的事件，一系列事件的按位或 short revents; //实际发生的事件，由内核填充，一些列事件的按位或 } poll事件类型 事件 描述 是否可作为输入 是否可作为输出 POLLIN 普通或优先数据可读 是 是 POLLRDNORM 普通数据可读 是 是 POLLRDBAND 优先级带数据可读(linux不支持) 是 是 POLLPRI 高优先级数据可读，如TCP带外数据 是 是 POLLOUT 普通或优先数据可写 是 是 POLLWRNORM 普通数据可写 是 是 POLLWRBAND 优先级带数据可写 是 是 POLLRDHUB TCP连接被对方关闭，或者对方关闭了写操作。由GNU引入 是 是 POLLERR 错误 否 是 POLLHUB 挂起。比如管道的写端被关闭，读端描述符上将收到POLLHUB事件 否 是 POLLNVAL 文件描述符没有打开 否 是 epoll

epoll相比select更加高效，主要体现在：

• epoll使用内核事件表维护监听的文件描述符集，避免频繁的用户空间与内核空间频繁的拷贝开销。
• epoll直接返回发生事件的文件描述符即，避免了用户程序轮询的开销。
• epoll内部通过注册回调函数的方式，监听特定文件描述符上的时间，避免了内核轮询开销。
• epoll提供了高效的边沿触发模式，边沿触发带来编程上的复杂性。

LT与ET模式

使用epoll监听文件描述符时，有两种事件触发模式：

• 水平触发（LT）：epoll默认使用水平触发。
• 边沿触发（ET）：往epoll内核事件表这种注册一个文件描述符上的EPOLLET时，将进行边沿触发。相比较水平触发，边沿触发更加高效。

系统调用

epoll在内核中使用事件表（红黑树）维护监听的文件描述符，并支持对事件表的增删改。

• 创建事件表：epoll_create，返回一个内核事件表的文件描述符。
• 操作事件表：epoll_ctl，对事件表进行增、删、改。
• 监听事件表：epoll_wait，监听事件表上的事件。

#include <sys/epoll.h> //epoll事件结构体 strcut epoll_event{ _uint32_t events; //epoll事件 epoll_data_t data; //用户数据，包含监听的文件描述符信息 } //用户数据联合体 typedef union epoll_data{ void* ptr; //指定与fd相关的用户数据，用户数据包含监听的文件描述符 int fd; //监听的文件描述符fd（常用） uint32_t u32; uint64_t u64; } epoll_data_t; //创建epoll内核事件表 //size： 提示内核需要的事件表多大 int epoll_create(int size); //操作epoll内核事件表 //epfd： epoll内核事件表 //op： 参数指定操作，包括增（EPOLL_CTL_ADD)、删（EPOLL_CTL_DEL)、改（EPOLL_CTL_MOD) //fd： 被操作的文件描述符 //event：事件 int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event); //监听事件表上文件描述符的事件 //epfd： epoll内核事件表 //events： 就绪事件数组，由内核修改 //maxevents：最多监听事件数 //timeout: //return： 就绪文件描述符个数 int epoll_wait(int epfd, struct epoll_event* events, int maxevents, int timeout); 事件类型

epoll支持的时间类型和poll基本相同，表示epoll时间类型的宏实在poll对应的宏前加上‘E’，如EPOLLIN。

示例 服务端

服务端代码功能：

1. 创建socket
2. 为socket绑定固定地址和端口
3. 监听套接字等待客户端连接
4. 接收客户端连接，获取连接socket
5. 使用select、poll或epoll监听连接socket的可读事件
6. 连接socket可读事件发生时，从连接socket中读取数据

select

#include <sys/types.h> #include <sys/socket.h> #include <netinet/in.h> #include <arpa/inet.h> #include <assert.h> #include <stdio.h> #include <unistd.h> #include <errno.h> #include <string.h> #include <fcntl.h> #include <stdlib.h> int main(int argc, char *argv[]) { if (argc <= 2) { printf("usage: %s ip_address port_number\n", basename(argv[0])); return 1; } const char *ip = argv[1]; int port = atoi(argv[2]); printf("ip is %s and port is %d\n", ip, port); int ret = 0; struct sockaddr_in address; bzero(&address, sizeof(address)); address.sin_family = AF_INET; inet_pton(AF_INET, ip, &address.sin_addr); address.sin_port = htons(port); int listenfd = socket(PF_INET, SOCK_STREAM, 0); assert(listenfd >= 0); ret = bind(listenfd, (struct sockaddr *)&address, sizeof(address)); assert(ret != -1); ret = listen(listenfd, 5); assert(ret != -1); struct sockaddr_in client_address; socklen_t client_addrlength = sizeof(client_address); int connfd = accept(listenfd, (struct sockaddr *)&client_address, &client_addrlength); if (connfd < 0) { printf("errno is: %d\n", errno); close(listenfd); } char remote_addr[INET_ADDRSTRLEN]; printf("connected with ip: %s and port: %d\n", inet_ntop(AF_INET, &client_address.sin_addr, remote_addr, INET_ADDRSTRLEN), ntohs(client_address.sin_port)); char buf[1024]; fd_set read_fds; fd_set exception_fds; FD_ZERO(&read_fds); FD_ZERO(&exception_fds); int nReuseAddr = 1; setsockopt(connfd, SOL_SOCKET, SO_OOBINLINE, &nReuseAddr, sizeof(nReuseAddr)); while (1) { memset(buf, '\0', sizeof(buf)); FD_SET(connfd, &read_fds); FD_SET(connfd, &exception_fds); ret = select(connfd + 1, &read_fds, NULL, &exception_fds, NULL); printf("select one\n"); if (ret < 0) { printf("selection failure\n"); break; } if (FD_ISSET(connfd, &read_fds)) { ret = recv(connfd, buf, sizeof(buf) - 1, 0); if (ret <= 0) { printf("can not read\n"); break; } printf("get %d bytes of normal data: %s\n", ret, buf); } else if (FD_ISSET(connfd, &exception_fds)) { ret = recv(connfd, buf, sizeof(buf) - 1, MSG_OOB); if (ret <= 0) { printf("can not read\n"); break; } printf("get %d bytes of oob data: %s\n", ret, buf); } } close(connfd); close(listenfd); return 0; } poll

#include <sys/types.h> #include <sys/socket.h> #include <netinet/in.h> #include <arpa/inet.h> #include <assert.h> #include <stdio.h> #include <unistd.h> #include <errno.h> #include <string.h> #include <fcntl.h> #include <stdlib.h> #include <poll.h> int main(int argc, char *argv[]) { if (argc <= 2) { printf("usage: %s ip_address port_number\n", basename(argv[0])); return 1; } const char *ip = argv[1]; int port = atoi(argv[2]); printf("ip is %s and port is %d\n", ip, port); int ret = 0; struct sockaddr_in address; bzero(&address, sizeof(address)); address.sin_family = AF_INET; inet_pton(AF_INET, ip, &address.sin_addr); address.sin_port = htons(port); int listenfd = socket(PF_INET, SOCK_STREAM, 0); assert(listenfd >= 0); ret = bind(listenfd, (struct sockaddr *)&address, sizeof(address)); assert(ret != -1); ret = listen(listenfd, 5); assert(ret != -1); struct sockaddr_in client_address; socklen_t client_addrlength = sizeof(client_address); int connfd = accept(listenfd, (struct sockaddr *)&client_address, &client_addrlength); if (connfd < 0) { printf("errno is: %d\n", errno); close(listenfd); } char remote_addr[INET_ADDRSTRLEN]; printf("connected with ip: %s and port: %d\n", inet_ntop(AF_INET, &client_address.sin_addr, remote_addr, INET_ADDRSTRLEN), ntohs(client_address.sin_port)); char buf[1024]; pollfd poll_fds[1]; bzero(poll_fds, sizeof(poll_fds)); poll_fds[0].fd = connfd; poll_fds[0].events = POLLIN; int nReuseAddr = 1; setsockopt(connfd, SOL_SOCKET, SO_OOBINLINE, &nReuseAddr, sizeof(nReuseAddr)); while (1) { memset(buf, '\0', sizeof(buf)); ret = poll(poll_fds, 1, -1); printf("select one\n"); if (ret < 0) { printf("selection failure\n"); break; } if (poll_fds[0].events == POLLIN) { ret = recv(connfd, buf, sizeof(buf) - 1, 0); if (ret <= 0) { printf("can not read\n"); break; } printf("get %d bytes of normal data: %s\n", ret, buf); } } close(connfd); close(listenfd); return 0; } epoll

#include <sys/types.h> #include <sys/socket.h> #include <netinet/in.h> #include <arpa/inet.h> #include <assert.h> #include <stdio.h> #include <unistd.h> #include <errno.h> #include <string.h> #include <fcntl.h> #include <stdlib.h> #include <sys/epoll.h> int main(int argc, char *argv[]) { if (argc <= 2) { printf("usage: %s ip_address port_number\n", basename(argv[0])); return 1; } const char *ip = argv[1]; int port = atoi(argv[2]); printf("ip is %s and port is %d\n", ip, port); int ret = 0; struct sockaddr_in address; bzero(&address, sizeof(address)); address.sin_family = AF_INET; inet_pton(AF_INET, ip, &address.sin_addr); address.sin_port = htons(port); int listenfd = socket(PF_INET, SOCK_STREAM, 0); assert(listenfd >= 0); ret = bind(listenfd, (struct sockaddr *)&address, sizeof(address)); assert(ret != -1); ret = listen(listenfd, 5); assert(ret != -1); struct sockaddr_in client_address; socklen_t client_addrlength = sizeof(client_address); int connfd = accept(listenfd, (struct sockaddr *)&client_address, &client_addrlength); if (connfd < 0) { printf("errno is: %d\n", errno); close(listenfd); } char remote_addr[INET_ADDRSTRLEN]; printf("connected with ip: %s and port: %d\n", inet_ntop(AF_INET, &client_address.sin_addr, remote_addr, INET_ADDRSTRLEN), ntohs(client_address.sin_port)); char buf[1024]; //创建内核事件表 int epfd = epoll_create(10); //注册事件 epoll_event event; event.data.fd = connfd; event.events = EPOLLIN; epoll_ctl(epfd, EPOLL_CTL_ADD, connfd, &event); //就绪事件数组 epoll_event events[10]; int nReuseAddr = 1; setsockopt(connfd, SOL_SOCKET, SO_OOBINLINE, &nReuseAddr, sizeof(nReuseAddr)); while (1) { memset(buf, '\0', sizeof(buf)); ret = epoll_wait(epfd, events, 1, -1); printf("select one\n"); if (ret < 0) { printf("errno: %d\n", errno); printf("selection failure\n"); break; } if (events[0].events == EPOLLIN) { ret = recv(connfd, buf, sizeof(buf) - 1, 0); if (ret <= 0) { printf("can not read\n"); break; } printf("get %d bytes of normal data: %s\n", ret, buf); } } close(connfd); close(listenfd); return 0; } 客户端

客户端代码功能：

1. 创建套接字
2. 请求连接
3. 发送数据
4. 关闭连接。

#include <sys/socket.h> #include <netinet/in.h> #include <arpa/inet.h> #include <assert.h> #include <stdio.h> #include <unistd.h> #include <string.h> #include <stdlib.h> int main(int argc, char *argv[]) { if (argc <= 2) { printf("usage: %s ip_address port_number\n", basename(argv[0])); return 1; } const char *ip = argv[1]; int port = atoi(argv[2]); struct sockaddr_in server_address; bzero(&server_address, sizeof(server_address)); server_address.sin_family = AF_INET; inet_pton(AF_INET, ip, &server_address.sin_addr); server_address.sin_port = htons(port); int sockfd = socket(PF_INET, SOCK_STREAM, 0); assert(sockfd >= 0); if (connect(sockfd, (struct sockaddr *)&server_address, sizeof(server_address)) < 0) { printf("connection failed\n"); } else { printf("send oob data out\n"); const char *oob_data = "abc"; const char *normal_data = "123"; send(sockfd, normal_data, strlen(normal_data), 0); send(sockfd, oob_data, strlen(oob_data), MSG_OOB); send(sockfd, normal_data, strlen(normal_data), 0); } close(sockfd); return 0; }