Linux网络编程：select和epoll入门

在服务器开发中，怎么构造更加高性能的服务器是经久不衰的话题，其中的select和epoll是经典的IO服用模型。

多路IO转接服务器也叫做多任务IO服务器。该类服务器实现的主旨思想是，不再由应用程序自己监视客户端连接，取而代之由内核替应用程序监视文件。

Select IO复用

最简单的多路IO实现方式是select，它的机制很好懂。

相关函数：

int select(int nfds, fd_set *readfds, fd_set *writefds, fd_set *exceptfds, struct timeval *timeout);

它的参数解释如下：

nfds: 监控的文件描述符集里最大文件描述符加1，因为此参数会告诉内核检测前多少个文件描述符的状态

readfds：监控有读数据到达文件描述符集合，传入传出参数

writefds：监控写数据到达文件描述符集合，传入传出参数

exceptfds：监控异常发生达文件描述符集合,如带外数据到达异常，传入传出参数

timeout：定时阻塞监控时间，3种情况

1.NULL，永远等下去

2.设置timeval，等待固定时间

3.设置timeval里时间均为0，检查描述字后立即返回，轮询

struct timeval {

long tv_sec; /* seconds */

long tv_usec; /* microseconds */

};

这里需要特别注意的是readfds和writefds和exceptfds集合，它们都是bitmap，每一个位置相对于一个文件描述符。这参数是传入传出参数，传入的是要监听的文件描述符集合，传出的是监听完成后有反应的文件描述符集合。

对这三个集合的操作函数如下：

void FD_CLR(int fd, fd_set *set); //把文件描述符集合里fd清0

int FD_ISSET(int fd, fd_set *set); //测试文件描述符集合里fd是否置1

void FD_SET(int fd, fd_set *set); //把文件描述符集合里fd位置1

void FD_ZERO(fd_set *set); //把文件描述符集合里所有位清0

select返回值：就绪描述符的数目，超时返回0，出错返回-1；就是说返回数字说有多少个fd有反应。

那么服务器服务过程中，我们需要不断select监听得到有反应的fd集合，然后再对有反应的fd做出相应处理（这个过程我们称之为轮询）。可以看出我们每次轮询都要从最下的fd循环到最大的fd，这个过程如果最小最大相差巨大将耗费很多时间，我们可以对这个过程进行小优化：我们把用到的fd用数组记录下来，然后轮询的时候我们就不需要从最小到最大循环了，我们只用查看记录的fd数组就行了。

  1 #include<stdio.h>
  2 #include<stdlib.h>
  3 #include<unistd.h>
  4 #include<string.h>
  5 #include<arpa/inet.h>
  6 #include<ctype.h>
  7 
  8 #include"wrap.h"
  9 
 10 #define SERV_PORT 6666
 11 
 12 int main(int argc,char *argv[])
 13 {
 14     int i,j,n,maxi;
 15     int nready,client[FD_SETSIZE];        //自定义数组client, 防止遍历1024个文件描述符  FD_SETSIZE默认为1024
 16     int maxfd,listenfd,connfd,sockfd;
 17     char buf[BUFSIZ],str[INET_ADDRSTRLEN];    //#define INET_ADDRSTRLEN 16
 18 
 19     struct sockaddr_in clie_addr,serv_addr;
 20     socklen_t clie_addr_len;
 21     fd_set rset,allset;    // rset 读事件文件描述符集合 allset用来暂存
 22 
 23     //创建socket
 24     listenfd=Socket(AF_INET,SOCK_STREAM,0);
 25 
 26     //端口复用
 27     int opt=1;
 28     setsockopt(listenfd,SOL_SOCKET,SO_REUSEADDR,&opt,sizeof(opt));
 29 
 30     //写好结构体并Bind
 31     bzero(&serv_addr,sizeof(serv_addr));
 32     serv_addr.sin_family=AF_INET;
 33     serv_addr.sin_addr.s_addr=htonl(INADDR_ANY);
 34     serv_addr.sin_port=htons(SERV_PORT);
 35     Bind(listenfd,(struct sockaddr *)&serv_addr,sizeof(serv_addr));
 36     //Listen设置监听上限
 37     Listen(listenfd,128);
 38 
 39 
 40     maxfd=listenfd;        //maxfd是轮询fd最大值，起初listenfd为最大
 41 
 42     maxi=-1;    
 43     for (i=0;i<FD_SETSIZE;i++)
 44         client[i]=-1;    //初始化client[]
 45     FD_ZERO(&allset);
 46     FD_SET(listenfd,&allset);
 47 
 48     while (1) {
 49         //因为select函数会改变bitmap，所以复制一份然后再去select
 50         rset=allset;
 51         nready=select(maxfd+1,&rset,NULL,NULL,NULL);
 52         if (nready<0)
 53             perr_exit("select error");
 54 
 55         //看看select结果中有没有listenfd，有的话就是新连接请求
 56         if (FD_ISSET(listenfd,&rset)) {
 57             clie_addr_len=sizeof(clie_addr);
 58             connfd=Accept(listenfd,(struct sockaddr*)&clie_addr,&clie_addr_len);
 59 
 60             printf("received from %s ar PORT %d\n",inet_ntop(AF_INET,&clie_addr.sin_addr,str,sizeof(str)),
 61             ntohs(clie_addr.sin_port));
 62             
 63             for (i=0;i<FD_SETSIZE;i++)    //优化：把刚刚建立连接的socketfd存到client[]数组里，存到从小到达第一个未用位置
 64                 if (client[i]<0) {
 65                     client[i]=connfd;
 66                     break;
 67                 }
 68 
 69             if (i==FD_SETSIZE) {    //连接到达上限，危险
 70                 fputs("too many clients\n",stderr);
 71                 exit(1);
 72             }
 73 
 74             FD_SET(connfd,&allset);    //更新监听bitmap
 75 
 76             if (connfd>maxfd)
 77                 maxfd=connfd;    //用数组优化过版本，maxfd并不是特别有用了
 78 
 79             if (i>maxi) maxi=i;
 80 
 81             if (--nready==0) continue;    //就监听到一个反应时间还是请求连接，后面不用轮询了
 82         }
 83 
 84         //client[i]数组里存的都是已经建立连接的socketfd，那么这里轮询，如果这里的fd有反应就是读写请求
 85         //maxi是一个优化，把fd保存起来，省去每次从0到1024轮询的时间
 86         for (i=0;i<=maxi;i++) {
 87             if ((sockfd=client[i])<0)
 88                 continue;
 89             if (FD_ISSET(sockfd,&rset)) {
 90                 if ((n=Read(sockfd,buf,sizeof(buf)))==0) {    //Read返回0，连接终止
 91                     Close(sockfd);
 92                     FD_CLR(sockfd,&allset);
 93                     client[i]=-1;
 94                 } else if (n>0) {    //从对端读到数据
 95                     for (j=0;j<n;j++)
 96                         buf[j]=toupper(buf[j]);
 97                     Write(sockfd,buf,n);
 98                     Write(STDOUT_FILENO,buf,n);
 99                 }
100                 if (--nready==0) break;        //处理完所有反应了
101             }
102         }
103     }
104     
105     Close(listenfd);
106     return 0;
107 }

select的server代码

 1 #include<stdio.h>
 2 #include<string.h>
 3 #include<unistd.h>
 4 #include<netinet/in.h>
 5 #include<arpa/inet.h>
 6 
 7 #include"wrap.h"
 8 
 9 #define MAXLINE 8192
10 #define SERV_PORT 6666
11 
12 int main(int argc,char *argv[])
13 {
14     struct sockaddr_in servaddr;
15     char buf[MAXLINE];
16     int sockfd,n;
17 
18     //创建socket
19     sockfd=Socket(AF_INET,SOCK_STREAM,0);
20 
21     //写好servaddr结构体，connect
22     bzero(&servaddr,sizeof(servaddr));
23     servaddr.sin_family=AF_INET;
24     inet_pton(AF_INET,"127.0.0.1",&servaddr.sin_addr);
25     servaddr.sin_port=htons(SERV_PORT);
26 
27     Connect(sockfd,(struct sockaddr *)&servaddr,sizeof(servaddr));
28 
29     //不断从标准输入，输入后写到服务器，然后从服务器读入写到屏幕
30     while (fgets(buf,MAXLINE,stdin)!=NULL) {
31         Write(sockfd,buf,strlen(buf));
32         n=Read(sockfd,buf,MAXLINE);
33         if (n==0) {
34             printf("the other side has been closed.\n");
35             break;
36         } else
37             Write(STDOUT_FILENO,buf,n);
38     }
39 
40     Close(sockfd);
41     return 0;
42 }

select的client端代码

select的4个缺点：

1，单个进程能够监视的文件描述符的数量存在最大限制，通常是1024，当然可以更改数量，但由于select采用轮询的方式扫描文件描述符，文件描述符数量越多，性能越差；(在linux内核头文件中，有这样的定义：#define __FD_SETSIZE 1024)
2，内核 / 用户空间内存拷贝问题，select需要复制大量的句柄数据结构，产生巨大的开销；
3，select返回的是含有整个句柄的数组，应用程序需要遍历整个数组才能发现哪些句柄发生了事件；
4，select的触发方式是水平触发，应用程序如果没有完成对一个已经就绪的文件描述符进行IO操作，那么之后每次select调用还是会将这些文件描述符通知进程。

尽管上面也提到了select的小优化版本，但是它也只解决了第三个缺点，像是拷贝，水平触发的问题是解决不了的。

Epoll IO复用

在epoll出现之前，其实还有一种IO复用模型poll，但是相比select模型，poll使用链表保存文件描述符，没有了监视文件数量的限制，但其他三个缺点依然存在。所以其实poll和select并没有本质区别，如果是监听数量很多但是每次有反应的fd很少这种情况，select和epoll还是性能低下。于是epoll横空出世：

由于epoll的实现机制与select/poll机制完全不同，上面所说的 select的缺点在epoll上不复存在。

epoll是Linux下多路复用IO接口select/poll的增强版本，它能显著提高程序在大量并发连接中只有少量活跃的情况下的系统CPU利用率，因为它会复用文件描述符集合来传递结果而不用迫使开发者每次等待事件之前都必须重新准备要被侦听的文件描述符集合（避免缺点2），另一点原因就是获取事件的时候，它无须遍历整个被侦听的描述符集，只要遍历那些被内核IO事件异步唤醒而加入Ready队列的描述符集合就行了（避免缺点3）。

目前epell是linux大规模并发网络程序中的热门首选模型。

epoll除了提供select/poll那种IO事件的电平触发（Level Triggered）外，还提供了边沿触发（Edge Triggered），这就使得用户空间程序有可能缓存IO状态，减少epoll_wait/epoll_pwait的调用，提高应用程序效率（避免缺点4）。

相关函数：

int epoll_create(int size)

函数作用是：创建一个epoll句柄，参数size用来告诉内核监听的文件描述符的个数，跟内存大小有关。

int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event)

epfd：为epoll_creat的句柄

op：表示动作，用3个宏来表示：

EPOLL_CTL_ADD (注册新的fd到epfd)，

EPOLL_CTL_MOD (修改已经注册的fd的监听事件)，

EPOLL_CTL_DEL (从epfd删除一个fd)；

event：告诉内核需要监听的事件

struct epoll_event {

__uint32_t events; /* Epoll events */

epoll_data_t data; /* User data variable */

};

　　　　　epoll_event其中的events是这样的：

　　　　　　　　EPOLLIN ：表示对应的文件描述符可以读（包括对端SOCKET正常关闭）

　　　　　　 EPOLLOUT：表示对应的文件描述符可以写

　　　　　　EPOLLPRI：表示对应的文件描述符有紧急的数据可读（这里应该表示有带外数据到来）

　　　　　　EPOLLERR：表示对应的文件描述符发生错误

　　　　　　 EPOLLHUP：表示对应的文件描述符被挂断；

　　　　　　EPOLLET：将EPOLL设为边缘触发(Edge Triggered)模式，这是相对于水平触发(Level Triggered)而言的

　　　　　　EPOLLONESHOT：只监听一次事件，当监听完这次事件之后，如果还需要继续监听这个socket的话，需要再次把这个socket加入到EPOLL队列里

　　　　　epoll_event其中的epoll_data是这样的：

　　　　　　typedef union epoll_data {

　　　　　　void *ptr;

　　　　　　int fd;

　　　　　　uint32_t u32;

　　　　　　uint64_t u64;

　　}

int epoll_wait(int epfd, struct epoll_event *events, int maxevents, int timeout)

events：用来存内核得到事件的集合，

maxevents：告之内核这个events有多大，这个maxevents的值不能大于创建epoll_create()时的size，

timeout：是超时时间

-1：阻塞

0：立即返回，非阻塞

>0：指定毫秒

返回值：成功返回有多少文件描述符就绪，时间到时返回0，出错返回-1

注意这个函数的events参数，这个参数是传出参数，类型是epoll_event结构体数组。那么在epoll_wait之后，得到的就是有反应的事件数组。是的！从这里可以看出我们不用像select和poll那样去轮询，epoll直接帮我们返回了需要处理的事件，我们直接处理这个数组里的事件就可以了。

  1 #include<stdio.h>
  2 #include<stdlib.h>
  3 #include<unistd.h>
  4 #include<string.h>
  5 #include<arpa/inet.h>
  6 #include<sys/epoll.h>
  7 #include<errno.h>
  8 #include<ctype.h>
  9 
 10 #include"wrap.h"
 11 
 12 #define MAXLINE 8192
 13 #define SERV_PORT 8000
 14 #define OPEN_MAX 5000
 15 
 16 int main(int argc,char *argv[])
 17 {
 18     int i,listenfd,connfd,sockfd;
 19     int n,num=0;
 20     ssize_t nready,efd,res;
 21     char buf[MAXLINE],str[INET_ADDRSTRLEN];
 22     socklen_t clilen;
 23 
 24     struct sockaddr_in cliaddr,servaddr;
 25     struct epoll_event tep,ep[OPEN_MAX];
 26 
 27     //前面这些东西没什么特别的,socket->bind->listen
 28     listenfd=Socket(AF_INET,SOCK_STREAM,0);
 29 
 30     int opt=1;
 31     setsockopt(listenfd,SOL_SOCKET,SO_REUSEADDR,&opt,sizeof(opt));
 32 
 33     bzero(&servaddr,sizeof(servaddr));
 34     servaddr.sin_family=AF_INET;
 35     servaddr.sin_addr.s_addr=htonl(INADDR_ANY);
 36     servaddr.sin_port=htons(SERV_PORT);
 37 
 38     Bind(listenfd,(struct sockaddr*)&servaddr,sizeof(servaddr));
 39 
 40     Listen(listenfd,20);
 41 
 42     //从这里开始epoll特别的地方
 43     //epoll_create创建一棵OPEN_MAX节点大小的 红黑树
 44     efd=epoll_create(OPEN_MAX);
 45     if (efd==-1)
 46         perr_exit("epoll_create error");
 47 
 48     //设置listenfd的epoll_event结构体，并把它添加到监听红黑树上
 49     tep.events=EPOLLIN; 
 50     tep.data.fd=listenfd;
 51     res=epoll_ctl(efd,EPOLL_CTL_ADD,listenfd,&tep);
 52     if (res==-1)
 53         perr_exit("epoll_wait error");
 54 
 55     //服务器监听过程
 56     for ( ; ; ) {
 57         /*epoll_wait()为server阻塞监听事件, ep为struct epoll_event类型数组, OPEN_MAX为数组容量, -1表永久阻塞*/
 58         nready=epoll_wait(efd,ep,OPEN_MAX,-1);
 59         if (nready==-1)
 60             perr_exit("epoll_wait error");
 61 
 62         //从epoll_wait后出来的ep数组就是有反应的事件
 63         for (i=0;i<nready;i++) {
 64             if (!(ep[i].events & EPOLLIN))        //不是"读"事件
 65                 continue;
 66             if (ep[i].data.fd==listenfd) {    //是lfd，是请求连接
 67                 clilen=sizeof(cliaddr);
 68                 connfd=Accept(listenfd,(struct sockaddr*)&cliaddr,&clilen);
 69 
 70                 printf("received from %s at PORT %d\n",inet_ntop(AF_INET,&cliaddr.sin_addr,str,sizeof(str)),ntohs(cliaddr.sin_port));
 71                 printf("cfd %d--client %d\n",connfd,++num);
 72 
 73                 //Accept之后把新连接的读事件添加到监听红黑树上
 74                 tep.events=EPOLLIN; 
 75                 tep.data.fd=connfd;
 76                 res=epoll_ctl(efd,EPOLL_CTL_ADD,connfd,&tep);
 77                 if (res==-1)
 78                     perr_exit("epoll_ctl_ error");
 79             } else {    //不是lfd，是读事件
 80                 sockfd=ep[i].data.fd;
 81                 n=Read(sockfd,buf,MAXLINE);
 82 
 83                 if (n==0) {    //对端关闭链接，从树上取下
 84                     res=epoll_ctl(efd,EPOLL_CTL_DEL,sockfd,NULL);
 85                     if (res==-1)
 86                         perr_exit("epoll ctl error");
 87                     Close(sockfd);
 88                     printf("client[%d] closed connection\n",sockfd);
 89                 } else if (n<0) {    //读错误，取下
 90                     perror("read n<0 error:");
 91                     res=epoll_ctl(efd,EPOLL_CTL_DEL,sockfd,NULL);
 92                     Close(sockfd);
 93                 } else {    //正常读到数据
 94                     for (i=0;i<n;i++)
 95                         buf[i]=toupper(buf[i]);
 96 
 97                     Write(STDOUT_FILENO,buf,n);
 98                     Write(sockfd,buf,n);
 99                 }
100             }    
101         }
102 
103     }
104     Close(listenfd);
105     Close(efd);
106 
107     return 0;
108 }

epoll的server端demo

客户端的跟select一样就不重新贴一遍了。

Reactor模型

上面的epoll只是一个demo，有很多欠考虑的地方结构也不够优美，reactor和proactor是两个著名的服务器IO模型，reactor是同步IO而proactor是异步IO，我们常用epoll能很自然地实现reactor模型（也可以用epoll模拟实现proactor）。Windows下通过IOCP实现了真正的异步 I/O，而在Linux系统下，Linux2.6才引入，并且异步I/O使用epoll实现的，所以还不完善。

reactor消息处理流程（多线程的）：

Reactor对象通过Select监控客户端请求事件，收到事件后通过dispatch进行分发。
如果是建立连接请求事件，则由acceptor通过accept处理连接请求，然后创建一个Handler对象处理连接完成后续的各种事件。
如果不是建立连接事件，则Reactor会分发调用连接对应的Handler来响应。
Handler只负责响应事件，不做具体业务处理，通过Read读取数据后，会分发给后面的Worker线程池进行业务处理。
Worker线程池会分配独立的线程完成真正的业务处理，如何将响应结果发给Handler进行处理。
Handler收到响应结果后通过send将响应结果返回给Client。

proactor消息处理流程：

应用程序初始化一个异步读取操作，然后注册相应的事件处理器，此时事件处理器不关注读取就绪事件，而是关注读取完成事件，这是区别于Reactor的关键。
事件分离器等待读取操作完成事件
在事件分离器等待读取操作完成的时候，操作系统调用内核线程完成读取操作，并将读取的内容放入用户传递过来的缓存区中。这也是区别于Reactor的一点，Proactor中，应用程序需要传递缓存区。
事件分离器捕获到读取完成事件后，激活应用程序注册的事件处理器，事件处理器直接从缓存区读取数据，而不需要进行实际的读取操作。

简单地理解，其实reactor和proactor的区别并不十分大，reactor模型中的事件处理器关心的是读就绪事件，事件处理器自己完成读时间和业务逻辑。而proactor模型的事件处理器只关心读完成事件，亦即proactor中的数据读取不是在事件处理器中完成的，是在内核线程完成的。

引用知乎关于这两个模型的区别一句话：

“ reactor：能收了你跟俺说一声。
proactor: 你给我收十个字节，收好了跟俺说一声。 ”

reactor代码如下，细节见注释。

  1 #include <stdio.h>
  2 #include<sys/socket.h>
  3 #include<unistd.h>
  4 #include<stdlib.h>
  5 #include<sys/epoll.h>
  6 #include <errno.h>
  7 #include<string.h>
  8 #include <fcntl.h>
  9 #include<arpa/inet.h>
 10 #include <time.h>
 11 
 12 #define MAX_EVENTS 1024         //监听上限数
 13 #define BUFLEN 4096                 //缓冲区大小
 14 #define SERV_PORT 8080          //默认端口
 15 
 16 struct myevent_s
 17 {
 18     int fd;             //要监听的文件描述符
 19     int events;      //对应的监听事件
 20     void* arg;
 21     void (*call_back) (int fd, int events, void* arg);      //回调函数
 22     int status;     //是否在监听:1->在红黑树上(监听), 0->不在(不监听)
 23     char buf[BUFLEN];
 24     int len;
 25     long last_active;       //记录每次加入红黑树 g_efd 的时间值
 26 };
 27 
 28 int g_efd;      //epoll_create返回的文件描述符，红黑树的fd
 29 struct myevent_s g_events[MAX_EVENTS + 1];      //结构体数组
 30 
 31 void recvdata(int fd, int events, void* arg);
 32 void senddata(int fd, int events, void* arg);
 33 
 34 /*将结构体 myevent_s  ev的成员变量 初始化*/
 35 //ev是事件结构体，call_back是事件的回调函数,   arg泛型指针
 36 void eventset(myevent_s* ev, int fd, void (*call_back)(int, int, void*), void* arg) {
 37     ev->fd = fd;
 38     ev->call_back = call_back;
 39     ev->events = 0;
 40     ev->arg = arg;
 41     ev->status = 0;
 42     ev->len = 0;
 43     memset(ev->buf, 0, sizeof(ev->buf));
 44     ev->last_active = time(NULL);
 45     return;
 46 }
 47 
 48 /* 向 epoll监听的红黑树 添加一个 文件描述符
 49 efd：红黑树fd，events：事件类型，ev：事件结构体*/
 50 void eventadd(int efd, int events, myevent_s* ev) {
 51     epoll_event epv = { 0,{0} };
 52     int op;
 53     epv.data.ptr = ev;      //这个指针指向我们自定义的  "事件结构体myevent"
 54     epv.events = ev->events = events;
 55 
 56     if (ev->status == 1) {
 57         op = EPOLL_CTL_MOD;
 58     }
 59     else {
 60         op = EPOLL_CTL_ADD;
 61         ev->status = 1;
 62     }
 63     //以上代码都是在创建并填充epoll_ctl所需参数
 64 
 65     //准备完毕，插入红黑树
 66     if (epoll_ctl(efd, op, ev->fd, &epv) < 0)
 67         printf("event add failed [fd=%d], event[%d]\n", ev->fd, events);
 68     else
 69         printf("event add OK [fd=%d], op=%d, event[%0X]\n", ev->fd, op, events);
 70     return;
 71 }
 72 
 73 /* 从epoll 监听的 红黑树中删除一个 文件描述符*/
 74 void eventdel(int efd, myevent_s* ev) {
 75     epoll_event epv = { 0,{0} };
 76 
 77     if (ev->status != 1) return;
 78 
 79     epv.data.ptr = ev;      //epv指针指向 myevent
 80     ev->status = 0;         //修改状态
 81     epoll_ctl(efd, EPOLL_CTL_DEL, ev->fd, &epv);        //从红黑树 efd 上将 ev->fd 摘除
 82 }
 83 
 84 /*  当有文件描述符就绪, epoll返回, 调用该函数 与客户端建立链接 */
 85 void acceptconn(int lfd, int events, void* arg) {
 86     sockaddr_in cin;
 87     socklen_t len = sizeof(cin);
 88     int cfd, i;
 89 
 90     if ((cfd = accept(lfd, (sockaddr*)&cin, &len)) == -1) {
 91         if (errno != EAGAIN && errno != EINTR) {
 92 
 93         }
 94         printf("%s: accept, %d\n", __func__, strerror(errno));
 95         return;
 96     }
 97 
 98     do {
 99         //从g_events中找到第一个空闲元素
100         for (i = 0; i < MAX_EVENTS; i++)
101             if (g_events[i].status == 0) break;
102         //用光了
103         if (i == MAX_EVENTS) {
104             printf("%s: fcntl nonblocking failed, %s\n", __func__, strerror(errno));
105             break;
106         }
107 
108         //给cfd设置一个 myevent_s 结构体, 回调函数 设置为 recvdata
109         //设置好之后添加到红黑树
110         eventset(&g_events[i], cfd, recvdata, &g_events[i]);
111         eventadd(g_efd, EPOLLIN, &g_events[i]);
112     } while (0);
113     
114     printf("new connection [%d:%s][time:%ld], pos[%d]\n",
115         inet_ntoa(cin.sin_addr), ntohs(cin.sin_port), g_events[i].last_active, i);
116     return;
117 }
118 
119 /* 处理读事件 */
120 void recvdata(int fd, int events, void* arg) {
121     myevent_s* ev = (myevent_s*)arg;
122     int len;
123 
124     //读入数据
125     len = recv(fd, ev->buf, sizeof(ev->buf), 0);    
126     //从红黑树把该读事件摘下
127     eventdel(g_efd, ev);
128 
129     if (len > 0) {      //>0，读入成功
130         ev->len = len;
131         ev->buf[len] = '\0';
132         printf("C[%d]:%s\n", fd, ev->buf);
133 
134         eventset(ev, fd, senddata, ev);
135         eventadd(g_efd, EPOLLOUT, ev);
136     }
137     else if (len == 0) {    //等于0，对端关闭
138         close(ev->fd);
139         printf("[fd=%d] pos[%ld],closed\n", fd, ev - g_events);
140     }
141     else {      //<0，读出错
142         close(ev->fd);
143         printf("recv[fd=%d] error[%d]:%s\n", fd, errno, strerror(errno));
144     }
145     return;
146 }
147 
148 /* 处理写事件 */
149 void senddata(int fd, int events, void* arg) {
150     struct myevent_s* ev = (myevent_s*)arg;
151     int len;
152 
153     len = send(fd, ev->buf, ev->len, 0);
154 
155     if (len > 0) {      //写成功
156         printf("send[fd=%d], [%d]%s\n", fd, len, ev->buf);
157         eventdel(g_efd, ev);
158         eventset(ev, fd, recvdata, ev);
159         eventadd(g_efd, EPOLLIN, ev);
160     }
161     else {      //写失败
162         close(ev->fd);
163         eventdel(g_efd, ev);
164         printf("send[fd=%d] error %d\n", fd, strerror(errno));
165     }
166     return;
167 }
168 
169 /*创建 listenfd的socket, 初始化lfd 
170 efd：红黑树fd，port：监听的端口号*/
171 void initlistensocket(int efd, short port) {
172     int lfd = socket(AF_INET, SOCK_STREAM, 0);
173     fcntl(lfd, F_SETFL, O_NONBLOCK);    //将socket设为非阻塞
174 
175     //初始化，回调函数是acceptconn
176     eventset(&g_events[MAX_EVENTS], lfd, acceptconn, &g_events[MAX_EVENTS]);
177     //把监听事件挂到树上
178     eventadd(efd, EPOLLIN, &g_events[MAX_EVENTS]);
179 
180     struct sockaddr_in sockaddr;
181     memset(&sockaddr, 0, sizeof(sockaddr));
182     sockaddr.sin_family = AF_INET;
183     sockaddr.sin_addr.s_addr = INADDR_ANY;
184     sockaddr.sin_port = htons(port);
185 
186     bind(lfd, (struct sockaddr *)&sockaddr, sizeof(sockaddr));
187 
188     listen(lfd, 20);
189 
190     return;
191 }
192 
193 int main(int argc,char *argv[])
194 {
195     unsigned short port = SERV_PORT;    //默认端口号
196     if (argc == 2)
197         port = atoi(argv[1]);       //提供的端口号
198 
199     g_efd = epoll_create(MAX_EVENTS + 1);
200     if (g_efd <= 0)
201         printf("create efd in %s err %s\n", __func__, strerror(errno));
202 
203     //创建listenfd的socket
204     initlistensocket(g_efd, port);
205 
206     //用于存储epoll_wait之后有反应的事件
207     epoll_event events[MAX_EVENTS + 1];
208     printf("server running:port [%d]\n", port);
209 
210     int checkpos = 0, i;
211     while (1) {
212         //每次验证100个已连接的客户端链接，若此客户端60s内没有反应，关闭链接
213         long now = time(NULL);
214         for (i = 0; i < 100; i++, checkpos++) {
215             if (checkpos == MAX_EVENTS) checkpos = 0;
216             if (g_events[checkpos].status != 1) continue;
217 
218             //距离上次有反应到现在间隔事件
219             long duration = now - g_events[checkpos].last_active;
220             //超时了，直接关闭+删除
221             if (duration >= 60) {
222                 close(g_events[checkpos].fd);
223                 printf("[fd=%d] timeout\n", g_events[checkpos].fd);
224                 eventdel(g_efd, &g_events[checkpos]);
225             }
226         }
227        
228         int nfd = epoll_wait(g_efd, events, MAX_EVENTS + 1, 1000);
229         if (nfd < 0) {  //epoll_wait出错
230             printf("epoll_wait error, exit\n");
231             break;
232         }
233 
234         //处理每一个有反应的事件
235         for (i = 0; i < nfd; i++) {
236             myevent_s* ev = (myevent_s*)events[i].data.ptr;
237             //是读事件
238             if ((events[i].events & EPOLLIN) && (ev->events & EPOLLIN)) {
239                 ev->call_back(ev->fd, events[i].events, ev->arg);
240             }
241             //是写事件
242             if ((events[i].events & EPOLLOUT) && (ev->events & EPOLLOUT)) {
243                 ev->call_back(ev->fd, events[i].events, ev->arg);
244             }
245         }
246     }
247 
248     //关闭资源
249     close(g_efd);
250     pause();
251     return 0;
252 }

epoll实现的reactor

参考资料和推荐阅读：

select/poll/epoll：https://www.cnblogs.com/Anker/p/3265058.html

reactor/proactor：https://cloud.tencent.com/developer/article/1488120

原文地址：https://www.cnblogs.com/clno1/p/12950645.html