I/O复用
用途:I/O 复用能同时监听多个文件描述符。
I/O 复用虽然能同时监听多个文件描述符,但它本身是阻塞的。并且当多个文件描述符同时就绪时,如果不采取额外的措施,程序就只能按顺序依处理其中的每一个文件描述符,这使得服务器看起来好像是串行工作 的。
如果要提高并发处理的能力,可以配合使用多线程或多进程等编程方法。
在Linux下,有三种系统调用函数,分别是select,poll,epoll。
1.select
用途:在一段指定时间内,监听用户感兴趣的文件描述符的可读、可写和异常等事件。
1.接口介绍:
1.int select(int maxfd, fd_set *readfds, fd_set *writefds, fd_set *exceptfds, struct timeval *timeout);
1.select 成功时返回就绪(可读、可写和异常)文件描述符的总数。如果在超时时间内没有任何文件描述符就 绪,select 将返回 0。select 失败是返回-1.如果在 select 等待期间,程序接收到信号,则 select 立即返 回-1,并设置 errno 为 EINTR。
2.maxfd 参数:指定的被监听的文件描述符的总数。它通常被设置为 select 监听的所 有文件描述符中的最大值+1 。
3.readfds、writefds 和 exceptfds 参数分别指向可读、可写和异常等事件对应的文件 描述符集合。应用程序调用 select 函数时,通过这 3 个参数传入自己感兴趣的文件 描述符。select 返回时,内核将修改它们来通知应用程序哪些文件描述符已经就绪。
通过下列宏可以访问 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
int FD_ISSET(int fd, fd_set *fdset);// 测试 fdset 的位 fd 是否被设置
4.timeout 参数:用来设置 select 函数的超时时间。它是一个 timeval 结构类型的指 针,采用指针参数是因为 内核将修改它以告诉应用程序 select 等待了多久。timeval 结构的定义如下:
struct timeval {
long tv_sec; //秒数
long tv_usec; // 微秒数
};
如果给 timeout 的两个成员都是 0,则 select 将立即返回。如果 timeout 传递 NULL,则 select 将一直阻 塞,直到某个文件描述符就绪 。
2. fd_set 是用来存放描述符的对应事件的集合。
2.例子
1.用select监听键盘是否有数据输入
#include <stdio.h>
#include <sys/select.h>
#include <stdlib.h>
#include <sys/time.h>
#include <string.h>
#include <unistd.h>
int main()
{char buff[256] = {0};fd_set fdset;while (1){FD_ZERO(&fdset);FD_SET(0, &fdset);struct timeval tim = {5,0}; //每次都需要传入一个新的timeval结构体int n = select(1,&fdset,NULL,NULL,&tim);//select中会改变tim的值if(-1 == n){printf("input error\n");continue;}else if(0 == n){printf("timeout\n");continue;}else {if(FD_ISSET(0,&fdset)){memset(buff,0,256);read(0,buff,255);printf("read:%s\n",buff);}}}
}
运行结果:
2.用select处理多个并发客户端
原理图:
服务端参考代码:
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <string.h>
#include <assert.h>
#include <sys/socket.h>
#include <sys/types.h>
#include <sys/select.h>
#include <sys/time.h>
#include <arpa/inet.h>
#include <netinet/in.h>#define MaxLen 60
void fds_init(int fds[]) //初始化
{for (int i = 0; i < MaxLen; ++i){fds[i] = -1; // 无效的描述符}
}
void fds_add(int fd, int fds[]) //添加描述符
{for (int i = 0; i < MaxLen; ++i){if (-1 == fds[i]){fds[i] = fd;break;}}
}
void fds_del(const int fd, int fds[]) //删除描述符
{for (int i = 0; i < MaxLen; ++i){if (fd == fds[i]){fds[i] = -1;break;}}
}
int main()
{int sockfd = socket(AF_INET, SOCK_STREAM, 0);if (-1 == sockfd){exit(1);}struct sockaddr_in saddr;memset(&saddr, 0, sizeof(saddr));saddr.sin_family = AF_INET;saddr.sin_port = htons(6000); // htons 将主机字节序转换为网络字节saddr.sin_addr.s_addr = inet_addr("127.0.0.1"); // 回环地址int res = bind(sockfd, (struct sockaddr *)&saddr, sizeof(saddr));if (-1 == res){exit(1);}res = listen(sockfd, 5);if (-1 == res){exit(1);}int fds[MaxLen]; // 存放套结字fds_init(fds); //初始化fds_add(sockfd, fds); //加入本地服务器的套结字fd_set fdset;while (1){FD_ZERO(&fdset);int maxfd = -1;for (int i = 0; i < MaxLen; ++i) // 将所有已经建立连的接描述符写入fdset{if (-1 != fds[i]){FD_SET(fds[i], &fdset);maxfd = maxfd > fds[i] ? maxfd : fds[i]; //记录文件描述符最大值}}struct timeval tim = {5, 0};int n = select(maxfd + 1, &fdset, NULL, NULL, &tim);if (-1 == n){printf("error\n");continue;}else if (0 == n){printf("timeout\n");continue;}else{for (int i = 0; i < MaxLen; ++i){if (-1 == fds[i]){continue;}if (FD_ISSET(fds[i], &fdset)){if (fds[i] == sockfd){struct sockaddr_in caddr;int len = sizeof(caddr);int c = accept(sockfd, (struct sockaddr *)&caddr, &len);if (-1 == c){continue;}printf("c:%d\n", c);fds_add(c, fds); //建立连接,保存}else{char buff[256] = {0};int n = recv(fds[i], buff, 255, 0);if (0 >= n){fds_del(fds[i], fds); //断开连接,清除close(fds[i]);printf("one client close");}else{printf("client[%d]:%s\n", fds[i], buff);send(fds[i], "ok", 2, 0);}}}}}}
}
客户端参考代码:
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <string.h>
#include <assert.h>
#include <sys/socket.h>
#include <sys/types.h>
#include <arpa/inet.h>
#include <netinet/in.h>
int main()
{int sockfd = socket(AF_INET, SOCK_STREAM, 0);assert(sockfd != -1);struct sockaddr_in saddr;memset(&saddr, 0, sizeof(saddr));saddr.sin_family = AF_INET;saddr.sin_port = htons(6000);saddr.sin_addr.s_addr = inet_addr("127.0.0.1");int res = connect(sockfd, (struct sockaddr *)&saddr, sizeof(saddr));if (-1 == res){ exit(1);}while (1){char buff[128] = {0};printf("input:\n");fgets(buff, 128, stdin); // 会取得\nif (strncmp(buff, "end", 3) == 0){break;}send(sockfd, buff, strlen(buff), 0);memset(buff, 0, 128);recv(sockfd, buff, 127, 0);printf("buff=%s\n", buff);}close(sockfd);exit(0);
}
运行结果:
2.poll
用途:在指定时间内轮询一定数量的文件描述符,以测试其中是否有就绪的 。
1.接口介绍:
1.int poll(struct pollfd *fds, nfds_t nfds, int timeout);
poll 系统调用成功返回就绪文件描述符的总数,超时返回 0,失败返回-1
nfds 参数:指定被监听事件集合 fds 的大小。
timeout 参数:指定 poll 的超时值,单位是毫秒,timeout 为-1 时,poll 调用将永久 阻塞,直到某个事件发生,timeout 为 0 时,poll 调用将立即返回。
fds 参数:是一个 struct pollfd 结构类型的数组,它指定所有用户感兴趣的文件描述 符上发生的可读、可写和异常等事件。
2.pollfd 结构体
定义如下:
struct pollfd
{
int fd; // 文件描述符
short events; // 注册的关注事件类型
short revents; // 实际发生的事件类型,由内核填充
};
其中,fd 成员指定文件描述符,events 成员告诉 poll 监听 fd 上的哪些事件类型。 它是一系列事件的按位或,revents 成员则由内核修改,通知应用程序 fd 上实际发生了哪些事件。
3.poll支持的事件:
2.例子
poll和select的原理相似;此处不再赘述。
本地服务器端参考代码:
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <string.h>
#include <sys/socket.h>
#include <sys/types.h>
#include <poll.h>
#include <sys/time.h>
#include <arpa/inet.h>
#include <netinet/in.h>
#include <stdbool.h>#define MaxLen 60void fds_init(struct pollfd fds[])
{for (int i = 0; i < MaxLen; ++i){fds[i].fd = -1;fds[i].events = 0;fds[i].revents = 0;}
}
void fds_add(struct pollfd fds[], int fd)
{for (int i = 0; i < MaxLen; ++i){if (-1 == fds[i].fd){fds[i].fd = fd;fds[i].events = POLLIN; // 读事件fds[i].revents = 0;break;}}
}
void fds_del(struct pollfd fds[], int fd)
{for (int i = 0; i < MaxLen; ++i){if (fd == fds[i].fd){fds[i].fd = -1;fds[i].events = 0;fds[i].revents = 0;break;}}
}
int main()
{int sockfd = socket(AF_INET, SOCK_STREAM, 0);if (-1 == sockfd){exit(1);}struct sockaddr_in saddr;memset(&saddr, 0, sizeof(saddr));saddr.sin_family = AF_INET;saddr.sin_port = htons(6000); // htons 将主机字节序转换为网络字节saddr.sin_addr.s_addr = inet_addr("127.0.0.1"); // 回环地址int res = bind(sockfd, (struct sockaddr *)&saddr, sizeof(saddr));if (-1 == res){exit(1);}res = listen(sockfd, 5);if (-1 == res){exit(1);}struct pollfd fds[MaxLen]; //存放套接字描述符fds_init(fds);fds_add(fds, sockfd);while (true){int n = poll(fds, MaxLen, 6000); // 阻塞if (-1 == n){continue;}else if (0 == n){printf("time out\n");continue;}else{for (int i = 0; i < MaxLen; ++i){if (-1 == fds[i].fd){continue;}if (fds->revents & POLLIN){if (sockfd == fds[i].fd){struct sockaddr_in caddr;int len = sizeof(caddr);int c = accept(sockfd, (struct sockaddr *)&caddr, &len);if (-1 == c){continue;}printf("c:%d\n", c);fds_add(c,fds);}else{char buff[256] = {0};int n = recv(fds[i].fd, buff, 255, 0);if (0 >= n){fds_del(fds[i].fd, fds); //断开连接,清除close(fds[i].fd);printf("one client close"); }else{printf("client[%d]:%s\n", fds[i], buff);send(fds[i].fd, "ok", 2, 0);}}}}}}
}
epoll
- epoll 是 Linux 特有的 I/O 复用函数。
- epoll 把用户关心的文件描述符上的事件放在内核里的一个事件表中。从而无需像 select 和 poll 那样每次调用都要重复传入文件描述符或事件集。但 epoll 需要使用一个额外的文件描述符,来唯一标识内核中的这个事件表。
1.接口介绍:
1.epoll_create() 用于创建内核事件表
原型:int epoll_create(int size);
epoll_create()成功返回内核事件表的文件描述符,失败返回-1
size 参数:现在并不起作用,只是给内核一个提示,告诉它事件表需要多大。
时间表实际上是一颗红黑树
2.epoll_ctl()用于操作内核事件表
原型:int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event);
epoll_ctl()成功返回 0,失败返回-1
epfd 参数:指定要操作的内核事件表的文件描述符
fd 参数:指定要操作的文件描述符
op 参数:指定操作类型:
- EPOLL_CTL_ADD 往内核事件表中中添加一个文件描述符(即参数fd),指定监视的事件类型(参数event)
- EPOLL_CTL_MOD 修改内核事件表中已经存在的描述符(即参数fd)对应的监视事件类型(参数event)
- EPOLL_CTL_DEL 将某内核事件表中已经存在的描述符(即参数fd)删除,参数event传NULL
event 需要epoll监视的fd对应的事件类型,它是 epoll_event 结构指针类型,epoll_event 的定义如下:
struct epoll_event
{
_uint32_t events; // epoll 事件
epoll_data_t data; // 用户数据
};
其中,events 成员描述事件类型,epoll 支持的事件类型与 poll 基本相同,表示 epoll 事件的宏是在 poll 对应的宏前加上‘E’,比如 epoll 的数据可读事件是 EPOLLIN。但是 epoll 有两个额外的事件类型–EPOLLET 和 EPOLLONESHOT。 data 成员用于存储用户数据,是一个联合体,其定义如下:
typedef union epoll_data
{
void *ptr;
int fd;
uint32_t u32;
uint64_t u64;
}epoll_data_t;
其中 fd 成员使用的最多,它指定事件所从属的目标文件描述符。
3.epoll_wait()用于在一段超时时间内等待一组文件描述符上的事件
原型:int epoll_wait(int epfd, struct epoll_event *events, int maxevents, int timeout);
epoll_wait()成功返回需要处理的事件数目,失败返回-1,超时返回 0
epfd 参数:指定要操作的内核事件表的文件描述符
events 参数:接口的返回参数;是一个用户数组,epoll把发生的事件的集合从内核复制到 events数组中。events数组是一个用户分配好大小的数组,数组长度大于等于maxevents。(events不可以是空指针,内核只负责把数据复制到这个 events数组中,不会去帮助我们在用户态中分配内存)
maxevents 参数:指定用户数组的大小,即指定最多监听多少个事件,它必须大于 0
timeout 参数:指定超时时间,单位为毫秒,如果 timeout 为 0,则 epoll_wait 会立即 返回,如果 timeout 为-1,则 epoll_wait 会一直阻塞,直到有事件就绪; 0表示不阻塞。
2.例子
本地服务器端参考代码:
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <string.h>
#include <sys/socket.h>
#include <sys/types.h>
#include <sys/epoll.h>
#include <sys/time.h>
#include <arpa/inet.h>
#include <netinet/in.h>
#include <stdbool.h>#define MaxLen 20
void epoll_add(int epfd, int fd, int op)
{struct epoll_event ep;ep.events = op;ep.data.fd = fd;if (epoll_ctl(epfd, EPOLL_CTL_ADD, fd, &ep) == -1){printf("Add fd error\n");}
}
void epoll_del(int epfd, int fd)
{if (epoll_ctl(epfd, EPOLL_CTL_DEL, fd, NULL) == -1){printf("Del fd error\n");}
}
int main()
{int sockfd = socket(AF_INET, SOCK_STREAM, 0);if (-1 == sockfd){exit(1);}struct sockaddr_in saddr;memset(&saddr, 0, sizeof(saddr));saddr.sin_family = AF_INET;saddr.sin_port = htons(6000); // htons 将主机字节序转换为网络字节saddr.sin_addr.s_addr = inet_addr("127.0.0.1"); // 回环地址int res = bind(sockfd, (struct sockaddr *)&saddr, sizeof(saddr));if (-1 == res){exit(1);}res = listen(sockfd, 5);if (-1 == res){exit(1);}int epfd = epoll_create(MaxLen); // 内核事件表的文件描述符if (-1 == epfd){exit(1);}struct epoll_event evs[MaxLen]; // 存放就绪的文件描述符epoll_add(epfd,sockfd,EPOLL_CTL_ADD);while (true){int n = epoll_wait(epfd, evs, MaxLen, 5000);if (-1 == n){continue;}else if (0 == n){printf("time out\n");continue;}else{for (int i = 0; i < n; ++i) // 区别于select和poll{int fd = evs[i].data.fd;if (evs[i].events & EPOLLIN) //读事件发生{if (sockfd == fd){struct sockaddr_in caddr;int len = sizeof(caddr);int c = accept(fd, (struct sockaddr *)&caddr, &len);if (0 >= c){continue;}printf("accept client:%d",c);epoll_add(epfd, c, EPOLL_CTL_ADD);}else{char buff[256] = {0};int c = recv(fd, buff, 255, 0);if (0 >= c){printf("one client close\n");epoll_del(epfd, fd);close(fd);continue;}printf("client(%d):%s\n", fd, buff);send(fd, "OK", 2, 0);}}}}}
}
3.LT 和 ET 模式
1.LT模式
以读事件为例,当缓冲区有数据准备好的时候,此时会触发读事件,**如果我们一直不去读取缓冲区里的数据,epoll模型就会一直通知我们有事件就绪。**LT模式也是epoll模型的默认模式。
2.ET模式
对于读事件 EPOLLIN,当该描述符对应的接受缓冲区的数据准备好的时候,也会触发读事件,但是只会触发一次,如果我们这次没有调用read/recv 读取 或者 没有一次读完,后面就不会通知有读事件就绪了。简单来说,只有当该描述符对应的接受缓冲区里的数据量发生变化的时候,才会通知我们一次,不会像LT模式那样一直通知。
示例图:
3.ET模式下的非阻塞编程:
阻塞状态下epoll存在问题:
只有接收缓冲区的数据变化时才会通知,通知的次数少了自然也会引发一些问题,比如触发读事件后必须把数据收取干净,因为你不一定有下一次机会再收取数据了,即使不采用一次读取干净的方式,也要把这个激活状态记下来,后续接着处理,否则如果数据残留到下一次消息来到时就会造成延迟现象。
解决方法:
1.为避免接收缓冲区中的数据一次取不完,我们采用循环来将缓冲区读取干净,当recv发现缓冲区里没有数据了,此时会默认进入阻塞状态,等待数据就绪,这就严重影响到后面的文件描述符读取/写入内容了。所以ET模式下必须要设为非阻塞。
2.当非阻塞状态下时,当缓冲区没有数据时,会返回error,并且将全局变量( errno ) 置为 EAGAIN 或 EWOULDBLOCK;
置为非阻塞的两种方法:
1.fnctl函数设定
int flag = fcntl(client_fd, F_GETFL);flag |= O_NONBLOCK;int ret = fcntl(client_fd, F_SETFL, flag);if (ret == -1){printf("Set Wait error\n");}
2.recv的参数设定
int c = recv(fd,buff,1,MSG_DONTWAIT);
if(0 >= c)
{if(errno != EAGAIN && errno != EWOULDBLOCK ) {epoll_del(epfd,fd);close(fd);printf("one client close\n");}
}
4.例子
本地服务器端参考代码
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <string.h>
#include <sys/socket.h>
#include <sys/types.h>
#include <sys/epoll.h>
#include <sys/time.h>
#include <arpa/inet.h>
#include <netinet/in.h>
#include <stdbool.h>
#include <fcntl.h> //新增
#include <errno.h> //新增#define MaxLen 20
void set_NoWait(int fd)
{int oldfl = fcntl(fd,F_GETFL); //获取描述符的状态int newfl = oldfl | O_NONBLOCK;int n = fcntl(fd,F_SETFL,newfl); //新增非阻塞状态if(-1 == n){printf("Set NoWait error\n");}
}
void epoll_add(int epfd, int fd, int op)
{struct epoll_event ep;ep.events = op | EPOLLET; //开启ET模式ep.data.fd = fd;set_NoWait(fd); //置为非阻塞if (epoll_ctl(epfd, EPOLL_CTL_ADD, fd, &ep) == -1){printf("Add fd error\n");}
}
void epoll_del(int epfd, int fd)
{if (epoll_ctl(epfd, EPOLL_CTL_DEL, fd, NULL) == -1){printf("Del fd error\n");}
}
int main()
{int sockfd = socket(AF_INET, SOCK_STREAM, 0);if (-1 == sockfd){exit(1);}struct sockaddr_in saddr;memset(&saddr, 0, sizeof(saddr));saddr.sin_family = AF_INET;saddr.sin_port = htons(6000); // htons 将主机字节序转换为网络字节saddr.sin_addr.s_addr = inet_addr("127.0.0.1"); // 回环地址int res = bind(sockfd, (struct sockaddr *)&saddr, sizeof(saddr));if (-1 == res){exit(1);}res = listen(sockfd, 5);if (-1 == res){exit(1);}int epfd = epoll_create(MaxLen); // 内核事件表的文件描述符if (-1 == epfd){exit(1);}struct epoll_event evs[MaxLen]; // 存放就绪的文件描述符epoll_add(epfd,sockfd,EPOLL_CTL_ADD);while (true){int n = epoll_wait(epfd, evs, MaxLen, 5000);if (-1 == n){continue;}else if (0 == n){printf("time out\n");continue;}else{for (int i = 0; i < n; ++i) // 区别于select和poll{int fd = evs[i].data.fd;if (evs[i].events & EPOLLIN) //读事件发生{if (sockfd == fd){struct sockaddr_in caddr;int len = sizeof(caddr);int c = accept(fd, (struct sockaddr *)&caddr, &len);if (0 >= c){continue;}printf("accept client:%d",c);epoll_add(epfd, c, EPOLL_CTL_ADD);}else{char buff[256] = {0};while(true) //采用循环清空接受缓冲区{int c = recv(fd,buff,1,0);if(0 >= c){if(errno != EAGAIN && errno != EWOULDBLOCK ) {epoll_del(epfd,fd);close(fd);printf("one client close\n");}break;}printf("client(%d):%s\n",fd,buff);send(fd,"OK",2,0);}}}}}}
}
客户端参考代码
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <string.h>
#include <assert.h>
#include <sys/socket.h>
#include <sys/types.h>
#include <arpa/inet.h>
#include <netinet/in.h>
int main()
{int sockfd = socket(AF_INET, SOCK_STREAM, 0);assert(sockfd != -1);struct sockaddr_in saddr;memset(&saddr, 0, sizeof(saddr));saddr.sin_family = AF_INET;saddr.sin_port = htons(6000);saddr.sin_addr.s_addr = inet_addr("127.0.0.1");int res = connect(sockfd, (struct sockaddr *)&saddr, sizeof(saddr));if (-1 == res){ exit(1);}while (1){char buff[128] = {0};printf("input:\n");fgets(buff, 128, stdin); // 会取得\nbuff[strlen(buff)-1] = 0; //删除\nif (strncmp(buff, "end", 3) == 0){break;}send(sockfd, buff, strlen(buff), 0);memset(buff, 0, 128);recv(sockfd, buff, 127, 0);printf("buff=%s\n", buff);}close(sockfd);exit(0);
}
运行结果:
select/poll/epoll的区别
select
1.select 调用需要传入 fd 数组,需要拷贝一份到内核,高并发场景下这样的拷贝消耗的资源是惊人的。
2.select 在内核层仍然是通过遍历的方式检查文件描述符的就绪状态,是个同步过程。
3.select 仅仅返回可读文件描述符的个数,具体哪个可读还是要用户自己遍历。
poll
本质上和select没有区别,主要就是去掉了 select 只能监听 1024 个文件描述符的限制。
epoll
1.内核中保存一份文件描述符集合,无需用户每次都重新传入,只需告诉内核修改的部分即可。
2.内核不再通过轮询的方式找到就绪的文件描述符,而是通过异步 IO 事件唤醒。
3.内核仅会将有 IO 事件的文件描述符返回给用户,用户也无需遍历整个文件描述符集合。
总结:
I/O多路复用就通过一种机制,可以监视多个描述符,一旦某个描述符就绪(一般是读就绪或者写就绪),能够通知程序进行相应的读写操作。select,poll,epoll本质上都是同步I/O,因为他们都需要在读写事件就绪后自己负责进行读写,也就是说这个读写过程是阻塞的。