网络IO管理 - 多路复用IO
思考
- 多路复用怎么理解？
select怎么管理fd的？怎么准确的知道哪个fd需要处理？重要的接口怎么理解？- 强大且低调的
epoll强大在什么地方？
推荐学习
推荐一个零声学院免费公开课程，个人觉得老师讲得不错，分享给大家：Linux，Nginx，ZeroMQ，MySQL，Redis，fastdfs，MongoDB，ZK，流媒体，CDN，P2P，K8S，Docker，TCP/IP，协程，DPDK等技术内容，立即学习
网络IO模型
多路复用IO
1. 理解多路复用
多路复用 IO (IO multiplexing) ，IO multiplexing 这个词可能有点陌生，但是提到 select/epoll，大概就都能明白了。有些地方也称这种 IO 方式为事件驱动 IO(event driven IO)。我们都知道，select/epoll 的好处就在于单个 process 就可以同时处理多个网络连接的 IO。它的基本原理就是 select/epoll 这个 function会不断的轮询所负责的所有 socket，当某个 socket 有数据到达了，就通知用户进程。
2. SELECT
当用户进程调用了
select，那么整个进程会被block，而同时，kernel 会“监视”所有 select 负责的 socket，当任何一个 socket 中的数据准备好了，select 就会返回。这个时候用户进程再调用 read 操作，将数据从 kernel 拷贝到用户进程。使用 select 以后最大的优势是用户可以在一个线程内同时处理多个
socket的 IO 请求。用户可以注册多个socket，然后不断地调用select读取被激活的socket，即可达到在同一个线程内同时处理多个 IO 请求的目的。而在同步阻塞模型中，必须通过多线程的方式才能达到这个目的。
3.重要的接口
大部分 Unix/Linux 都支持 select 函数，该函数用于探测多个文件句柄的状态变化。下面给出 select 接口的原型：
FD_ZERO(int fd, fd_set* rfds) FD_SET(int fd, fd_set* rfds) FD_ISSET(int fd, fd_set* frds) FD_CLR(int fd, fd_set* rfds) int select(int nfds, fd_set *readfds, fd_set *writefds, fd_set* exceptfds, struct timeval *timeout)
难点 - 参数形象化理解
这里，fd_set 类型可以简单的理解为按 bit 位标记句柄的队列，例如要在某
fd_set中标记一个值为 16 的句柄，则该fd_set的第 16 个 bit 位被标记为 1。具体的置位、验证可使用 FD_SET、FD_ISSET 等宏实现。最关键的地方是如何动态维护 select()的三个参数
readfds、writefds和exceptfds。作为输入参数，readfds 应该标记所有的需要探测的“可读事件”的句柄，其中永远包括那个探测connect() 的那个“母”句柄；同时，writefds 和 exceptfds 应该标记所有需要探测的“可写事件”和“错误事件”的句柄 ( 使用 FD_SET() 标记 )。
代码展示
#include <errno.h> #include <netinet/in.h> #include <stdio.h> #include <stdlib.h> #include <string.h> #include <sys/socket.h> #include <sys/types.h> #include <unistd.h> #include <sys/poll.h> #include <sys/epoll.h> #include <pthread.h> #define MAXLNE 4096 #define POLL_SIZE 1024 //8m * 4G = 128 , 512 //C10k void *client_routine(void *arg) { // int connfd = *(int *)arg; char buff[MAXLNE]; while (1) { int n = recv(connfd, buff, MAXLNE, 0); if (n > 0) { buff[n] = '\0'; printf("recv msg from client: %s\n", buff); send(connfd, buff, n, 0); } else if (n == 0) { close(connfd); break; } } return NULL; } int main(int argc, char **argv) { int listenfd, connfd, n; struct sockaddr_in servaddr; char buff[MAXLNE]; if ((listenfd = socket(AF_INET, SOCK_STREAM, 0)) == -1) { printf("create socket error: %s(errno: %d)\n", strerror(errno), errno); return 0; } memset(&servaddr, 0, sizeof(servaddr)); servaddr.sin_family = AF_INET; servaddr.sin_addr.s_addr = htonl(INADDR_ANY); servaddr.sin_port = htons(9999); if (bind(listenfd, (struct sockaddr *)&servaddr, sizeof(servaddr)) == -1) { printf("bind socket error: %s(errno: %d)\n", strerror(errno), errno); return 0; } if (listen(listenfd, 10) == -1) { printf("listen socket error: %s(errno: %d)\n", strerror(errno), errno); return 0; } #elif 0 //io多路复用组件select // fd_set rfds, rset, wfds, wset; FD_ZERO(&rfds); //fd清空 FD_SET(listenfd, &rfds); FD_ZERO(&wfds); int max_fd = listenfd; while (1) { rset = rfds; wset = wfds; int nready = select(max_fd+1, &rset, &wset, NULL, NULL); if (FD_ISSET(listenfd, &rset)) { // struct sockaddr_in client; socklen_t len = sizeof(client); if ((connfd = accept(listenfd, (struct sockaddr *)&client, &len)) == -1) { printf("accept socket error: %s(errno: %d)\n", strerror(errno), errno); return 0; } FD_SET(connfd, &rfds); if (connfd > max_fd) max_fd = connfd; if (--nready == 0) continue; } int i = 0; for (i = listenfd+1;i <= max_fd;i ++) { if (FD_ISSET(i, &rset)) { // n = recv(i, buff, MAXLNE, 0); if (n > 0) { buff[n] = '\0'; printf("recv msg from client: %s\n", buff); FD_SET(i, &wfds); //reactor //send(i, buff, n, 0); } else if (n == 0) { // FD_CLR(i, &rfds); //printf("disconnect\n"); close(i); } if (--nready == 0) break; } else if (FD_ISSET(i, &wset)) { send(i, buff, n, 0); FD_SET(i, &rfds); } }
SELECT的不足点
一个select可以做到1024个fd的管理，多开几个线程，每个线程一个
select，多做几个线程可以突破C10k，但是很难突破到C100k(即一百万并发）。
select本身是这样几个集合：rset这个这个集合需要拷贝到内核中去监控这个集合，还需要吧有数据可读的拷贝出来这样的操作，对数量还是有极限的。就需要更加厉害的组件！
4. EPOLL
附加学习文章：徒手造了个轮子 — 实现epoll
回顾与分析
前面说到select很难突破C100k的用户，那这个更厉害的组件就是epoll。你可以想像100万的用户和服务器连接并不是所有的用户都会同时活跃，反而在一个时间端内可能就100万中的很少一部分是活跃的用户，在SELECT的介绍中最后说到有这样集合需要拷贝到内存中(用户态内存到内核态内存的大量复制)，而由操作系统内核去拷贝操作去确定有没有未处理的事件，这样的操作显然会浪费大量的资源与时间，select和poll就是这样做的，因此它们最多只能处理几千个并发连接。而epoll不这样做，它在Linux内核中申请了一个简易的文件系统，把原先的一个select或poll调用分成了3部分：
int epoll_create(int size); 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);
形象的例子
这里举一个形象的例子，
epolll就像投快递，一个快递员专门管理一个小区的所有块快递，这个快递小哥原来是挨家挨户的跑去投快递，拿快递。为了减轻快递小哥的工作量，这里就建立了某巢快递柜。这样不仅减轻了快递小哥的工作量而且提高了效率。这里有两个集合，一、小区所有的人(所有fd引入集合)。二、某巢快递柜(今天需要寄快递的用户在一个集合)。
1. epoll_create()
就像创建了这个小区，用户(即fd)搬进来。再来学术点的解释就是调用epoll_create建立一个epoll对象(在epoll文件系统中给这个句柄分配资源)；
2. epoll_ctl()
如在小区里搬进搬出的，可能从五楼搬到八楼的用户。学术点就是调用epoll_ctl向epoll对象中添加用户连接的套接字。
3. epoll_wait()
快递小哥多久来某巢快递柜来取走快递。学术点说就是调用epoll_wait收集发生事件的连接。
小知识
eopll没出现以前Linux只能做嵌入式，因为并发量不够。服务器的核心点就是一个while(1)循环，不断地监控各个IO里面有没有事件发生，通过eopll来判断socket中有事件没。
代码展示
#include <stdio.h> #include <stdlib.h> #include <string.h> #include <sys/socket.h> #include <sys/types.h> #include <unistd.h> #include <sys/poll.h> #include <sys/epoll.h> #include <pthread.h> #define MAXLNE 4096 #define POLL_SIZE 1024 //8m * 4G = 128 , 512 //C10k void *client_routine(void *arg) { // int connfd = *(int *)arg; char buff[MAXLNE]; while (1) { int n = recv(connfd, buff, MAXLNE, 0); if (n > 0) { buff[n] = '\0'; printf("recv msg from client: %s\n", buff); send(connfd, buff, n, 0); } else if (n == 0) { close(connfd); break; } } return NULL; } int main(int argc, char **argv) { int listenfd, connfd, n; struct sockaddr_in servaddr; char buff[MAXLNE]; if ((listenfd = socket(AF_INET, SOCK_STREAM, 0)) == -1) { printf("create socket error: %s(errno: %d)\n", strerror(errno), errno); return 0; } //poll/select --> // epoll_create // epoll_ctl(ADD, DEL, MOD) // epoll_wait int epfd = epoll_create(1); //int size struct epoll_event events[POLL_SIZE] = {0}; struct epoll_event ev; ev.events = EPOLLIN; ev.data.fd = listenfd; epoll_ctl(epfd, EPOLL_CTL_ADD, listenfd, &ev); while (1) { int nready = epoll_wait(epfd, events, POLL_SIZE, 5); if (nready == -1) { continue; } int i = 0; for (i = 0;i < nready;i ++) { int clientfd = events[i].data.fd; if (clientfd == listenfd) { struct sockaddr_in client; socklen_t len = sizeof(client); if ((connfd = accept(listenfd, (struct sockaddr *)&client, &len)) == -1) { printf("accept socket error: %s(errno: %d)\n", strerror(errno), errno); return 0; } printf("accept\n"); ev.events = EPOLLIN; ev.data.fd = connfd; epoll_ctl(epfd, EPOLL_CTL_ADD, connfd, &ev); } else if (events[i].events & EPOLLIN) { n = recv(clientfd, buff, MAXLNE, 0); if (n > 0) { buff[n] = '\0'; printf("recv msg from client: %s\n", buff); send(clientfd, buff, n, 0); } else if (n == 0) { // ev.events = EPOLLIN; ev.data.fd = clientfd; epoll_ctl(epfd, EPOLL_CTL_DEL, clientfd, &ev); close(clientfd); } } } } close(listenfd); return 0; }
这样只需要在进程启动时建立一个epoll对象，并在需要的时候向它添加或删除连接就可以了，因此，在实际收集事件时，epoll_wait的效率就会非常高，因为调用epoll_wait时并没有向它传递这C100k个连接，内核也不需要去遍历全部的连接。
Posix API总结
让你更好理解
链接: 网络原理 -Posix API.
