Linux小知识---常见的IO复用技术

IO即为网络I/O，多路即为多个TCP连接，复用即为共用一个线程或者进程，模型最大的优势是系统开销小，不必创建也不必维护过多的线程或进程。
简单来说，就是一个线程，负责处理多路数据。即监视一组描述符，哪个来了数据，我就处理哪个。

常用的方法：select，poll，epoll和kqueue。其中select各种标准系统都有，epoll是linux特有，kqueue是UNIX特有。

对比一下特点吧，先混个脸熟。

方法	特点
select	1.能够监听的文件描述符受系统限制
	2.当数据来临的时候，需要遍历所有fd描述符，才能找到哪个描述符有数据
	3.每次重新监听时，需要再次拷贝默认fd描述符到内核，并且接收到的数据也需要从内核拷贝出来
poll	不在受到描述符数量限制，其他与select一样
epoll	1.不受描述符数量限制
	2.当数据来临的时候，自动通知数据来临的通道，不需要再遍历查找，
	3.通过内核和用户空间共享一块内存来实现数据传递
kqueue	与epoll原理基本一样，实现起来代码更加简单

然后重点讲一下select和epoll，毕竟这两个是linux下多路IO复用的最具代表性的两个方法

select方法

函数原型

#include <sys/select.h>   int select(int maxfdp1, fd_set *readset, fd_set *writeset, fd_set *exceptset,struct timeval *timeout);

参数定义

参数	含义
intmaxfdp	整数值，是指集合中所有文件描述符的范围，即所有文件描述符的最大值加1，不能错
fd_set*readfds	指向fd_set结构的指针，这个集合中应该包括文件描述符，我们是要监视这些文件描述符的读变化的，即我们关心是否可以从这些文件中读取数据了
fd_set*writefds	是指向fd_set结构的指针，这个集合中应该包括文件描述符，我们是要监视这些文件描述符的写变化的，即我们关心是否可以向这些文件中写入数据了
fd_set*errorfds	同上面两个参数的意图，用来监视文件错误异常文件。
structtimeval* timeout	是select的超时时间。
	NULL：是将select置于阻塞状态；
	时间值设为0秒0微妙，就变成一个纯粹的非阻塞函数，不管文件描述符是否有变化，都立刻返回继续执行，文件无变化返回0，有变化返回一个正值
	timeout的值大于0，这就是等待的超时时间，即 select在timeout时间内阻塞，超时时间之内有事件到来就返回了，否则在超时后不管怎样一定返回，返回值同上述。
函数返回	当监视的相应的文件描述符集中满足条件时，返回一个大于0的数。
	当没有满足条件的文件描述符，且设置的timeval监控时间超时时，select函数会返回一个为0的值。
	当select返回负值时，发生错误。

这里的关键，要理解fds这个东西，就像是一个很大数字，每一位都一个用来绑定一个文件描述符，如果我们要监听多个socket，都需要用FD_SET函数配置到这个fds上，就像掩码一样，一旦有数据了，这个fds与我们要监听的socket通过FD_ISSET这么一查找，就知道是不是哪个socket来数据了。

实现范例，这里提供了一个TCP server的例子

#include<stdio.h>
#include<stdlib.h>
#include<assert.h>
#include<unistd.h>
#include<string.h>#include<sys/types.h>
#include<netinet/in.h>
#include<arpa/inet.h>
#include<sys/socket.h>
#include<sys/epoll.h>#define listen_port 6000#define MAX 1024#define CONCURRENT_MAX       8       //应用层同时可以处理的连接
#define HZ_SMG_MAXLEN       1024typedef struct _client_rec
{char ipaddr[64];int fd;
} CLIENT_REC;   static CLIENT_REC client_fds[CONCURRENT_MAX]; int main()
{ int i = 0;char input_msg[HZ_SMG_MAXLEN];    char recv_msg[HZ_SMG_MAXLEN];    const int on=1;//本地地址    struct sockaddr_in server_addr;    server_addr.sin_family = AF_INET;    server_addr.sin_port = htons(listen_port);    server_addr.sin_addr.s_addr = htonl(INADDR_ANY);;    bzero(&(server_addr.sin_zero), 8);  //创建socket    int server_sock_fd = socket(AF_INET, SOCK_STREAM, 0);    if(server_sock_fd == -1)    {    printf("!!!!!socket error\n");    goto msgtcpserver_exit;}setsockopt(server_sock_fd,SOL_SOCKET,SO_REUSEADDR,&on,sizeof(on));//绑定socket    int bind_result = bind(server_sock_fd, (struct sockaddr *)&server_addr, sizeof(server_addr));    if(bind_result == -1)    {    printf("!!!!!bind error\n");    goto msgtcpserver_exit;}    if(listen(server_sock_fd,8) == -1)    {    printf("!!!!!listen error\n"); goto msgtcpserver_exit;}    fd_set server_fd_set;    int max_fd = -1;    struct timeval tv;  //超时时间设置    while(1)    {    tv.tv_sec = 5;    tv.tv_usec = 0;FD_ZERO(&server_fd_set);FD_SET(server_sock_fd, &server_fd_set);    if(max_fd < server_sock_fd)    {    max_fd = server_sock_fd;    }    //客户端连接    for(i =0; i < CONCURRENT_MAX; i++)    {    if(client_fds[i].fd != 0)    {    FD_SET(client_fds[i].fd, &server_fd_set);    if(max_fd < client_fds[i].fd)    {    max_fd = client_fds[i].fd;    }    }    }    int ret = select(max_fd + 1, &server_fd_set, NULL, NULL, &tv);    if(ret < 0)    {    printf("!!!!!select 出错\n");    break;    }    else if(ret == 0)    {    //printf("!!!!!select 超时");    continue;    }    else    {    if(FD_ISSET(server_sock_fd, &server_fd_set))    {    //有新的连接请求    struct sockaddr_in client_address;    socklen_t address_len=sizeof(struct sockaddr_in);    int client_sock_fd = accept(server_sock_fd, (struct sockaddr *)&client_address, &address_len);    if(client_sock_fd > 0)    {    int index = -1; for( i = 0; i < CONCURRENT_MAX; i++)    {    if(client_fds[i].fd == 0)    {    index = i;    client_fds[i].fd = client_sock_fd;strcpy(client_fds[i].ipaddr,inet_ntoa(client_address.sin_addr));break;    }    }    if(index >= 0)    {    printf("新客户端(%d)加入成功 ip:%s, port:%d\n",index,inet_ntoa(client_address.sin_addr),ntohs(client_address.sin_port));   }    else    {    bzero(input_msg, HZ_SMG_MAXLEN);    strcpy(input_msg, "服务器加入的客户端数达到最大值,无法加入!\n");    send(client_sock_fd, input_msg, HZ_SMG_MAXLEN, 0);    printf("!!!!!客户端连接数达到最大值，新客户端加入失败 %s:%d\n", inet_ntoa(client_address.sin_addr), ntohs(client_address.sin_port));    }    }    }    for( i =0; i < CONCURRENT_MAX; i++)    {    if(client_fds[i].fd !=0)    {    if(FD_ISSET(client_fds[i].fd, &server_fd_set))    {    //处理某个客户端过来的消息long byte_num;bzero(recv_msg, HZ_SMG_MAXLEN);byte_num = recv(client_fds[i].fd, recv_msg, HZ_SMG_MAXLEN, 0);    if (byte_num > 0)    {if(byte_num > HZ_SMG_MAXLEN)    {    byte_num = HZ_SMG_MAXLEN;    recv_msg[byte_num-1] = '\0';    }else{recv_msg[byte_num] = '\0'; }printf("客户端(%d):%s\n", i, recv_msg);}    else if(byte_num < 0)    {    printf("!!!!!从客户端(%d)接受消息出错.\n", i);    }    else    {    FD_CLR(client_fds[i].fd, &server_fd_set);    client_fds[i].fd = 0;memset(client_fds[i].ipaddr,0,64);printf("客户端(%d)退出了\n", i);    }    }    }    }    }    } msgtcpserver_exit:if(server_sock_fd>0){close(server_sock_fd);}return 0;
}

epoll方法

epoll方法涉及到了三个函数，

#include<sys/epoll.h>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);

主要工作流程如下

epoll_create 创建一个epoll对象，一般epollfd = epoll_create()；
epoll_ctl epoll_add/epoll_del的合体），往epoll对象中增加/删除某一个流的某一个事件。
比如
epoll_ctl(epollfd, EPOLL_CTL_ADD, socket, evn_EPOLLIN);//有缓冲区内有数据时epoll_wait返回epoll_ctl(epollfd, EPOLL_CTL_DEL, socket, evn_EPOLLOUT);//缓冲区可写入时epoll_wait返回
epoll_wait(epollfd,…)等待直到注册的事件发生

直接看代码，很简单，这里提供了一个TCP server的例子

#include<stdio.h>
#include<stdlib.h>
#include<assert.h>
#include<unistd.h>
#include<string.h>#include<sys/types.h>
#include<netinet/in.h>
#include<arpa/inet.h>
#include<sys/socket.h>
#include<sys/epoll.h>#define MAX 1024
#define listen_port 6000int main()
{int listenfd=socket(AF_INET,SOCK_STREAM,0); assert(listenfd!=-1);struct sockaddr_in ser,cli;ser.sin_family=AF_INET;ser.sin_port=htons(listen_port);ser.sin_addr.s_addr=inet_addr("127.0.0.1");int res=bind(listenfd,(struct sockaddr*)&ser,sizeof(ser));assert(res!=-1);listen(listenfd,5);int epfd=epoll_create(1);    //创建内核事件表epfdassert(epfd!=-1);struct epoll_event ev;    ev.events=EPOLLIN;ev.data.fd=listenfd;     //初始化一个关于listenfd的event结构体res=epoll_ctl(epfd,EPOLL_CTL_ADD,listenfd,&ev);  //将关于listenfd的结构体放入内核事件表assert(res!=-1);struct epoll_event event[MAX];    //下面epoll_wait()要将就绪事件都放入该数组中返回回来while(1){int n=epoll_wait(epfd,event,MAX,-1);   //核心函数；返回就绪文件描述符个数if(n==-1)   {printf("error!\n");exit(0);}if(n==0){printf("timeout\n");continue;}int i=0;for(;i<n;++i){int fd=event[i].data.fd;if(event[i].events & EPOLLRDHUP)   //cli输入“end”{printf("break\n");close(fd);epoll_ctl(epfd,EPOLL_CTL_DEL,fd,NULL);   //将关于fd的结构体从epfd中删除continue;}if(event[i].events & EPOLLIN)  {if(fd==listenfd){int len=sizeof(cli);int c=accept(listenfd,(struct sockaddr*)&cli,&len);assert(c!=-1);printf("link succese\n");ev.events=EPOLLIN|EPOLLRDHUP;ev.data.fd=c;res=epoll_ctl(epfd,EPOLL_CTL_ADD,c,&ev);assert(res!=-1);}else{char buff[128]={0};int num=recv(fd,buff,127,0);assert(num!=-1);printf("%d:%s",fd,buff);send(fd,"ok",2,0);}}}}
}

kqueue

这个主要用在非linux的UNIX的系统下，暂时找到了一个官方的例子，可以参考一下。
这是一个监听文件是否被改动的例子

#include <event.h>
#include <sys/types.h>#include <err.h>
#include <fcntl.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>int main(int argc, char **argv)
{struct kevent event;     /* Event we want to monitor */struct kevent tevent;     /* Event triggered */int kq, fd, ret;if (argc != 2)err(EXIT_FAILURE, "Usage: %s path\n", argv[0]);// 打开文件，拿到文件描述符fd = open(argv[1], O_RDONLY);if (fd == -1)err(EXIT_FAILURE, "Failed to open '%s'", argv[1]);/* Create kqueue. */// 创建kqueue队列，返回描述符kq = kqueue();if (kq == -1)err(EXIT_FAILURE, "kqueue() failed");// EV_SET(kev, ident,    filter,    flags, fflags, data, udata);/*初始化kevent结构体ident：为文件描述符EVFILE_VNODE： 用这个filterEV_ADD：添加到kqueueEV_CLEAR：每次事件被取走，状态重置NOTE_WRITE：每当ident指向的文件描述符有写入时返回不用太纠结为什么要用EVFILE_VNODE这个filter，按照官网来说，这个filter就是要用监听文件变化的。*/EV_SET(&event,fd, EVFILT_VNODE, EV_ADD | EV_CLEAR, NOTE_WRITE, 0, NULL);/* Attach event to the  kqueue.    */// 还记得前面的设定么？nevents为0，立即返回，返回的值是kqueue放到eventlist里的事件数量，这里eventlist为NULL，所以返回的ret是0。// 所以这个语句的作用是向kqueue注册要监听的事件，仅此而已ret = kevent(kq, &event, 1, NULL, 0, NULL);if (ret == -1) // 注册失败会返回-1err(EXIT_FAILURE, "kevent register");if (event.flags & EV_ERROR) // 有其他错误，会置flags的EV_RROR位为1，错误数据放在data字段errx(EXIT_FAILURE, "Event error: %s", strerror(event.data));// 开启循环for (;;) {/*    Sleep until something happens. */// 这里nevents不为0，eventlist为这NULL，且timeout为空指针，那会永久阻塞，直到有事件产生ret = kevent(kq, NULL, 0, &tevent,    1, NULL);if (ret == -1) {err(EXIT_FAILURE, "kevent wait");} else if (ret > 0) {// 每当有东西写到文件里了，就会触发事件printf("Something was written in '%s'\n", argv[1]);}}
}

那这么一看，大家都算epoll了，谁他么还用select
我是从网上看到这几句话
1、表面上看epoll的性能最好，但是在连接数少并且连接都十分活跃的情况下，select和poll的性能可能比epoll好，毕竟epoll的通知机制需要很多函数回调。
2、select低效是因为每次它都需要轮询。但低效也是相对的，视情况而定，也可通过良好的设计改善

其实就是说，二者并存，其实是有一定理由的。根据实际情况决定采用哪种。

参考文章
linux select函数详解
select、poll、epoll之间的区别

每天多学习一点知识，就多一分安身立命之本。