我们可以将 Reactor 模型 + epoll 抽象成一个“事件驱动”的 I/O 框架。

当我们使用简单的阻塞IO时，当有多个客户端进行连接请求时，并不能同一时间都连接上，因此，需要使用循环和多线程的方式将这些连接分布在不同的线程中来各自处理。

但是，这样就引发了一些问题。当已连接的fd没有数据可读时，就会在线程中阻塞。当有大量的连接请求时，就会产生同样多的线程，这样就很浪费系统资源。因此需要我们来进行改进使用的方法，将这些问题尽量解决掉。

解决1

每个连接都会因为数据的读写而阻塞，能否使用一种方式将阻塞大大减少，这样就不会使所有的连接阻塞，只需要在一个或少量的阻塞对象上等待即可。

这里的解决方法便是使用 IO多路复用 ；

解决2

当有大量的连接请求时，就会有大量线程被创建，连接断开，线程也会随之销毁，连接阻塞时，线程也会一直存在。当数量庞大时就会很浪费系统资源。

这里的解决方法便是使用 线程池 ；

这两种方式组合在一起便是 Reactor模型。

1. Reactor

在Reactor中，核心是epoll，

|1.启动|
|main → socket → bind → listen 
      → epoll_create → epoll_ctl(ADD, listen_fd, EPOLLIN) → epoll_wait |
|2.连接|
|epoll_wait(返回) → accept → conn_fd 
      → epoll_ctl(ADD, conn_fd, EPOLLIN|EPOLLET) → epoll_wait |
|3.可读|
|epoll_wait 返回 → on_read → read (循环直到 EAGAIN) → 解析 → 生成响应 → 写应用缓冲区 
      → epoll_ctl(MOD, conn_fd, EPOLLIN|EPOLLOUT) → epoll_wait |
|4.关闭|
|epoll_wait 返回 → read=0 → close(conn_fd) 
      → epoll_ctl(DEL, conn_fd) |
|5.退出|
|收到 SIGINT → close(listen_fd) → close(epoll_fd) → exit |

其中主要的部分启动，连接，（可读）回调的过程如下所示：

2. 使用Reactor构建服务端

使用 epoll + Reactor 构建服务端程序示例：

C++
#include <stdlib.h>
#include <stdio.h> 
#include <stdio.h>
#include <sys/socket.h>
#include <sys/epoll.h>
#include <arpa/inet.h>
#include <fcntl.h>
#include <unistd.h>
#include <errno.h>
#include <string.h>
#include <time.h>

#define MAX_EVENTS  1024
#define BUFLEN 128
#define SERV_PORT   8080
/*
 * status:1表示在监听事件中，0表示不在 
 * last_active:记录最后一次响应时间,做超时处理
 */
struct myevent_s {
    int fd;                 //cfd listenfd
    int events;             //EPOLLIN  EPLLOUT
    void *arg;              //指向自己结构体指针
    void (*call_back)(int fd, int events, void *arg);
    int status;
    char buf[BUFLEN];
    int len;
    long last_active;
};

int g_efd;          /* epoll_create返回的句柄 */
struct myevent_s g_events[MAX_EVENTS+1];   /* +1 最后一个用于 listen fd */

void eventset(struct myevent_s *ev, int fd, void (*call_back)(int, int, void *), void *arg) {
    ev->fd = fd;
    ev->call_back = call_back;
    ev->events = 0;
    ev->arg = arg;
    ev->status = 0;
    //memset(ev->buf, 0, sizeof(ev->buf));
    //ev->len = 0;
    ev->last_active = time(NULL);
    return;
}

void recvdata(int fd, int events, void *arg);
void senddata(int fd, int events, void *arg);

void eventadd(int efd, int events, struct myevent_s *ev) {
    struct epoll_event epv = {0, {0}};
    int op;
    epv.data.ptr = ev;
    epv.events = ev->events = events;
    if (ev->status == 1) {
        op = EPOLL_CTL_MOD;
    } 
    else {
        op = EPOLL_CTL_ADD;
        ev->status = 1;
    }
    if (epoll_ctl(efd, op, ev->fd, &epv) < 0) {
        printf("event add failed [fd=%d], events[%d]\n", ev->fd, events);
    }
    else {
        printf("event add OK [fd=%d], op=%d, events[%0X]\n", ev->fd, op, events);
    }
    return;
}

void eventdel(int efd, struct myevent_s *ev) {
    struct epoll_event epv = {0, {0}};
    if (ev->status != 1) {
        return;
    }
    epv.data.ptr = ev;
    ev->status = 0;
    epoll_ctl(efd, EPOLL_CTL_DEL, ev->fd, &epv);
    return;
}

/*接受连接回调*/
void acceptconn(int lfd, int events, void *arg) {
    struct sockaddr_in cin;
    socklen_t len = sizeof(cin);
    int cfd, i;
    if ((cfd = accept(lfd, (struct sockaddr *)&cin, &len)) == -1) {
        if (errno != EAGAIN && errno != EINTR) {
            /* 暂时不做出错处理 */
        }
        printf("%s: accept, %s\n", __func__, strerror(errno));
        return;
    }
    do {
        for (i = 0; i < MAX_EVENTS; i++) {
            if (g_events[i].status == 0)
                break;
        }
        if (i == MAX_EVENTS) {
            printf("%s: max connect limit[%d]\n", __func__, MAX_EVENTS);
            break;
        }
        int flag = 0;
        if ((flag = fcntl(cfd, F_SETFL, O_NONBLOCK)) < 0) {
            printf("%s: fcntl nonblocking failed, %s\n", __func__, strerror(errno));
            break;
        }
        eventset(&g_events[i], cfd, recvdata, &g_events[i]);
        eventadd(g_efd, EPOLLIN, &g_events[i]);
    } while(0);
    printf("new connect [%s:%d][time:%ld], pos[%d]\n", inet_ntoa(cin.sin_addr), ntohs(cin.sin_port), g_events[i].last_active, i);
    return;
}

/*读取数据回调*/
void recvdata(int fd, int events, void *arg) {
    struct myevent_s *ev = (struct myevent_s *)arg;
    int len;
    len = recv(fd, ev->buf, sizeof(ev->buf), 0);
    eventdel(g_efd, ev);
    if (len > 0) {
        ev->len = len;
        ev->buf[len] = '\0';
        printf("C[%d]:%s\n", fd, ev->buf);
        /* 转换为发送事件 */
        eventset(ev, fd, senddata, ev);
        eventadd(g_efd, EPOLLOUT, ev);
    }
    else if (len == 0) {
        close(ev->fd);
        /* ev-g_events 地址相减得到偏移元素位置 */
        printf("[fd=%d] pos[%d], closed\n", fd, (int)(ev - g_events));
    }
    else {
        close(ev->fd);
        printf("recv[fd=%d] error[%d]:%s\n", fd, errno, strerror(errno));
    }
    return;
}

/*发送数据回调*/
void senddata(int fd, int events, void *arg) {
    struct myevent_s *ev = (struct myevent_s *)arg;
    int len;
    len = send(fd, ev->buf, ev->len, 0);
    //printf("fd=%d\tev->buf=%s\ttev->len=%d\n", fd, ev->buf, ev->len);
    //printf("send len = %d\n", len);
    eventdel(g_efd, ev);
    if (len > 0) {
        printf("send[fd=%d], [%d]%s\n", fd, len, ev->buf);
        eventset(ev, fd, recvdata, ev);
        eventadd(g_efd, EPOLLIN, ev);
    }
    else {
        close(ev->fd);
        printf("send[fd=%d] error %s\n", fd, strerror(errno));
    }
    return;
}

/*初始化监听套接字*/
void initlistensocket(int efd, short port) {
    int lfd = socket(AF_INET, SOCK_STREAM, 0);
    fcntl(lfd, F_SETFL, O_NONBLOCK);
    eventset(&g_events[MAX_EVENTS], lfd, acceptconn, &g_events[MAX_EVENTS]);
    
    eventadd(efd, EPOLLIN, &g_events[MAX_EVENTS]);
    
    struct sockaddr_in sin;
    memset(&sin, 0, sizeof(sin));
    sin.sin_family = AF_INET;
    sin.sin_addr.s_addr = INADDR_ANY;
    sin.sin_port = htons(port);
    
    bind(lfd, (struct sockaddr *)&sin, sizeof(sin));
    listen(lfd, 20);
    return;
}

int main(int argc, char *argv[]) {
    unsigned short port = SERV_PORT;
    if (argc == 2)
        port = atoi(argv[1]);
    
    g_efd = epoll_create(MAX_EVENTS+1);
    if (g_efd <= 0)
        printf("create efd in %s err %s\n", __func__, strerror(errno));
    initlistensocket(g_efd, port);
    /* 事件循环 */
    struct epoll_event events[MAX_EVENTS+1];
    printf("server running:port[%d]\n", port);
    int checkpos = 0, i;
    while (1) {
        /* 超时验证，每次测试100个链接，不测试listenfd 当客户端60秒内没有和服务器通信，则关闭此客户端链接 */
        long now = time(NULL);
        for (i = 0; i < 100; i++, checkpos++) {
            if (checkpos == MAX_EVENTS)
                checkpos = 0;
            if (g_events[checkpos].status != 1)
                continue;
            long duration = now - g_events[checkpos].last_active;
            if (duration >= 60) {
                close(g_events[checkpos].fd);
                printf("[fd=%d] timeout\n", g_events[checkpos].fd);
                eventdel(g_efd, &g_events[checkpos]);
            }
        }
        /* 等待事件发生 */
        int nfd = epoll_wait(g_efd, events, MAX_EVENTS+1, 1000);
        if (nfd < 0) {
            printf("epoll_wait error, exit\n");
            break;
        }
        for (i = 0; i < nfd; i++) {
            struct myevent_s *ev = (struct myevent_s *)events[i].data.ptr;
            if ((events[i].events & EPOLLIN) && (ev->events & EPOLLIN)) {
                ev->call_back(ev->fd, events[i].events, ev->arg);
            }
            if ((events[i].events & EPOLLOUT) && (ev->events & EPOLLOUT)) {
                ev->call_back(ev->fd, events[i].events, ev->arg);
            }
        }
        //
    }
    /* 退出前释放所有资源 */
    return 0;
}

理论