3.1 reactor反应堆模式

对每一个构架模式的分析，我们都使用参考文献的分析风格，着重分析意图、上下文、问题、解决方案、结构和实现 6个方面的内容。

3.1.1 意图

在事件驱动的应用中，将一个或多个客户的服务请求分离（demultiplex）和调度（dispatch）给应用程序。

3.1.2 上下文

在事件驱动的应用中，同步地、有序地处理同时接收的多个服务请求。

3.1.3 问题

在分布式系统尤其是服务器这一类事件驱动应用中，虽然这些请求最终会被序列化地处理，但是必须时刻准备着处理多个同时到来的服务请求。在实际应用 中，这些请求总是通过一个事件（如CONNECTOR、READ、WRITE等）来表示的。在有序地处理这些服务请求之前，应用程序必须先分离和调度这些 同时到达的事件。为了有效地解决这个问题，我们需要做到以下4方面：

为了提高系统的可测量性和反应时间，应用程序不能长时间阻塞在某个事件源上而停止对其他事件的处理，这样会严重降低对客户端的响应度。
为了提高吞吐量，任何没有必要的上下文切换、同步和CPU之间的数据移动都要避免。
引进新的服务或改良已有的服务都要对既有的事件分离和调度机制带来尽可能小的影响。
大量的应用程序代码需要隐藏在复杂的多线程和同步机制之后。

3.1.4. 解决方案

在一个或多个事件源上等待事件的到来，例如，一个已经连接的Socket描述符就是一个事件源。将事件的分离和调度整合到处理它的服务中，而将分离和调度机制从应用程序对特定事件的处理中分离开，也就是说分离和调度机制与特定的应用程序无关。

具体来说，每个应用程序提供的每个服务都有一个独立的事件处理器与之对应。由事件处理器处理来自事件源的特定类型的事件。每个事件处理器都事先注册 到Reactor管理器中。Reactor管理器使用同步事件分离器在一个或多个事件源中等待事件的发生。当事件发生后，同步事件分离器通知 Reactor管理器，最后由Reactor管理器调度和该事件相关的事件处理器来完成请求的服务。

3.1.5. 结构

在Reactor模式中，有5个关键的参与者。

描述符（handle）：由操作系统提供，用于识别每一个事件，如Socket描述符、文件描述符等。在Linux中，它用一个整数来表示。事件可以来自外部，如来自客户端的连接请求、数据等。事件也可以来自内部，如定时器事件。
同步事件分离器（demultiplexer）：是一个函数，用来等待一个或多个事件的发生。调用者会被阻塞，直到分离器分离的描述符集上有事件发生。Linux的select函数是一个经常被使用的分离器。
事件处理器接口（event handler）：是由一个或多个模板函数组成的接口。这些模板函数描述了和应用程序相关的对某个事件的操作。
具体的事件处理器：是事件处理器接口的实现。它实现了应用程序提供的某个服务。每个具体的事件处理器总和一个描述符相关。它使用描述符来识别事件、识别应用程序提供的服务。
Reactor 管理器（reactor）：定义了一些接口，用于应用程序控制事件调度，以及应用程序注册、删除事件处理器和相关的描述符。它是事件处理器的调度核心。 Reactor管理器使用同步事件分离器来等待事件的发生。一旦事件发生，Reactor管理器先是分离每个事件，然后调度事件处理器，最后调用相关的模 板函数来处理这个事件。
通过上述分析，我们注意到，是Reactor管理器而不是应用程序负责等待事件、分离事件和调度事件。实际上，Reactor管理器并没有被具体的 事件处理器调用，而是管理器调度具体的事件处理器，由事件处理器对发生的事件做出处理。这就是类似Hollywood原则的“反向控制”。应用程序要做的 仅仅是实现一个具体的事件处理器，然后把它注册到Reactor管理器中。接下来的工作由管理器来完成。这些参与者的相互关系如图2-1所示。

现在结合第1章分析的框架五元素来看一下Reactor构架模式的参与者与框架五元素之间的关系：Reactor构架模式的具体实现对应了元素1； 事件处理器接口对应元素2；具体的事件处理器对应元素3；Reactor管理器使用了Hollywood原则，可以认为和元素5对应；元素4的功能相对不 明显，没有明确的对应关系。

如果还是没有理解Reactor构架模式，没有关系，源代码会说明所有问题。此时可再分析一遍Reactor构架模式，然后继续以下内容。

libevent-3-反应堆模式.png

libevent-3-反应堆模式02.png

3.2 Epoll的反应堆模式实现

  1. #include <stdlib.h>
  2. #include <stdio.h> 
  3. #include <stdio.h>
  4. #include <sys/socket.h>
  5. #include <sys/epoll.h>
  6. #include <arpa/inet.h>
  7. #include <fcntl.h>
  8. #include <unistd.h>
  9. #include <errno.h>
  10. #include <string.h>
  11. #include <time.h>
  12. #define MAX_EVENTS  1024
  13. #define BUFLEN 128
  14. #define SERV_PORT   8080
  17. /*
  18.  * status:1表示在监听事件中，0表示不在 
  19.  * last_active:记录最后一次响应时间,做超时处理
  20.  */
  21. struct myevent_s {
  22.     int fd;                 //cfd listenfd
  23.     int events;             //EPOLLIN  EPLLOUT
  24.     void *arg;              //指向自己结构体指针
  25.     void (*call_back)(int fd, int events, void *arg);
  26.     int status;
  27.     char buf[BUFLEN];
  28.     int len;
  29.     long last_active;
  30. };
  32. int g_efd;          /* epoll_create返回的句柄 */
  33. struct myevent_s g_events[MAX_EVENTS+1];   /* +1 最后一个用于 listen fd */
  35. void eventset(struct myevent_s *ev, int fd, void (*call_back)(int, int, void *), void *arg)
  36. {
  37.     ev->fd = fd;
  38.     ev->call_back = call_back;
  39.     ev->events = 0;
  40.     ev->arg = arg;
  41.     ev->status = 0;
  42.     //memset(ev->buf, 0, sizeof(ev->buf));
  43.     //ev->len = 0;
  44.     ev->last_active = time(NULL);
  46.     return;
  47. }
  49. void recvdata(int fd, int events, void *arg);
  50. void senddata(int fd, int events, void *arg);
  52. void eventadd(int efd, int events, struct myevent_s *ev)
  53. {
  54.     struct epoll_event epv = {0, {0}};
  55.     int op;
  56.     epv.data.ptr = ev;
  57.     epv.events = ev->events = events;
  59.     if (ev->status == 1) {
  60.         op = EPOLL_CTL_MOD;
  61.     } 
  62.     else {
  63.         op = EPOLL_CTL_ADD;
  64.         ev->status = 1;
  65.     }
  67.     if (epoll_ctl(efd, op, ev->fd, &epv) < 0)
  68.         printf("event add failed [fd=%d], events[%d]\n", ev->fd, events);
  69.     else
  70.         printf("event add OK [fd=%d], op=%d, events[%0X]\n", ev->fd, op, events);
  72.     return;
  73. }
  75. void eventdel(int efd, struct myevent_s *ev)
  76. {
  77.     struct epoll_event epv = {0, {0}};
  79.     if (ev->status != 1)
  80.         return;
  82.     epv.data.ptr = ev;
  83.     ev->status = 0;
  84.     epoll_ctl(efd, EPOLL_CTL_DEL, ev->fd, &epv);
  86.     return;
  87. }
  90. void acceptconn(int lfd, int events, void *arg)
  91. {
  92.     struct sockaddr_in cin;
  93.     socklen_t len = sizeof(cin);
  94.     int cfd, i;
  96.     if ((cfd = accept(lfd, (struct sockaddr *)&cin, &len)) == -1) {
  97.         if (errno != EAGAIN && errno != EINTR) {
  98.             /* 暂时不做出错处理 */
  99.         }
  100.         printf("%s: accept, %s\n", __func__, strerror(errno));
  101.         return;
  102.     }
  104.     do {
  105.         for (i = 0; i < MAX_EVENTS; i++) {
  106.             if (g_events[i].status == 0)
  107.                 break;
  108.         }
  110.         if (i == MAX_EVENTS) {
  111.             printf("%s: max connect limit[%d]\n", __func__, MAX_EVENTS);
  112.             break;
  113.         }
  115.         int flag = 0;
  116.         if ((flag = fcntl(cfd, F_SETFL, O_NONBLOCK)) < 0)
  117.         {
  118.             printf("%s: fcntl nonblocking failed, %s\n", __func__, strerror(errno));
  119.             break;
  120.         }
  122.         eventset(&g_events[i], cfd, recvdata, &g_events[i]);
  123.         eventadd(g_efd, EPOLLIN, &g_events[i]);
  124.     } while(0);
  126.     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);
  128.     return;
  129. }
  131. void recvdata(int fd, int events, void *arg)
  132. {
  133.     struct myevent_s *ev = (struct myevent_s *)arg;
  134.     int len;
  136.     len = recv(fd, ev->buf, sizeof(ev->buf), 0);
  137.     eventdel(g_efd, ev);
  139.     if (len > 0) {
  140.         ev->len = len;
  141.         ev->buf[len] = '\0';
  142.         printf("C[%d]:%s\n", fd, ev->buf);
  143.         /* 转换为发送事件 */
  144.         eventset(ev, fd, senddata, ev);
  145.         eventadd(g_efd, EPOLLOUT, ev);
  146.     }
  147.     else if (len == 0) {
  148.         close(ev->fd);
  149.         /* ev-g_events 地址相减得到偏移元素位置 */
  150.         printf("[fd=%d] pos[%d], closed\n", fd, (int)(ev - g_events));
  151.     }
  152.     else {
  153.         close(ev->fd);
  154.         printf("recv[fd=%d] error[%d]:%s\n", fd, errno, strerror(errno));
  155.     }
  157.     return;
  158. }
  160. void senddata(int fd, int events, void *arg)
  161. {
  162.     struct myevent_s *ev = (struct myevent_s *)arg;
  163.     int len;
  165.     len = send(fd, ev->buf, ev->len, 0);
  166.     //printf("fd=%d\tev->buf=%s\ttev->len=%d\n", fd, ev->buf, ev->len);
  167.     //printf("send len = %d\n", len);
  169.     eventdel(g_efd, ev);
  170.     if (len > 0) {
  171.         printf("send[fd=%d], [%d]%s\n", fd, len, ev->buf);
  172.         eventset(ev, fd, recvdata, ev);
  173.         eventadd(g_efd, EPOLLIN, ev);
  174.     }
  175.     else {
  176.         close(ev->fd);
  177.         printf("send[fd=%d] error %s\n", fd, strerror(errno));
  178.     }
  180.     return;
  181. }
  183. void initlistensocket(int efd, short port)
  184. {
  185.     int lfd = socket(AF_INET, SOCK_STREAM, 0);
  186.     fcntl(lfd, F_SETFL, O_NONBLOCK);
  187.     eventset(&g_events[MAX_EVENTS], lfd, acceptconn, &g_events[MAX_EVENTS]);
  188.     eventadd(efd, EPOLLIN, &g_events[MAX_EVENTS]);
  190.     struct sockaddr_in sin;
  192.     memset(&sin, 0, sizeof(sin));
  193.     sin.sin_family = AF_INET;
  194.     sin.sin_addr.s_addr = INADDR_ANY;
  195.     sin.sin_port = htons(port);
  197.     bind(lfd, (struct sockaddr *)&sin, sizeof(sin));
  199.     listen(lfd, 20);
  201.     return;
  202. }
  204. int main(int argc, char *argv[])
  205. {
  206.     unsigned short port = SERV_PORT;
  208.     if (argc == 2)
  209.         port = atoi(argv[1]);
  211.     g_efd = epoll_create(MAX_EVENTS+1);
  213.     if (g_efd <= 0)
  214.         printf("create efd in %s err %s\n", __func__, strerror(errno));
  216.     initlistensocket(g_efd, port);
  218.     /* 事件循环 */
  219.     struct epoll_event events[MAX_EVENTS+1];
  221.     printf("server running:port[%d]\n", port);
  222.     int checkpos = 0, i;
  223.     while (1) {
  224.         /* 超时验证，每次测试100个链接，不测试listenfd 当客户端60秒内没有和服务器通信，则关闭此客户端链接 */
  225.         long now = time(NULL);
  226.         for (i = 0; i < 100; i++, checkpos++) {
  227.             if (checkpos == MAX_EVENTS)
  228.                 checkpos = 0;
  229.             if (g_events[checkpos].status != 1)
  230.                 continue;
  231.             long duration = now - g_events[checkpos].last_active;
  232.             if (duration >= 60) {
  233.                 close(g_events[checkpos].fd);
  234.                 printf("[fd=%d] timeout\n", g_events[checkpos].fd);
  235.                 eventdel(g_efd, &g_events[checkpos]);
  236.             }
  237.         }
  238.         /* 等待事件发生 */
  239.         int nfd = epoll_wait(g_efd, events, MAX_EVENTS+1, 1000);
  240.         if (nfd < 0) {
  241.             printf("epoll_wait error, exit\n");
  242.             break;
  243.         }
  244.         for (i = 0; i < nfd; i++) {
  245.             struct myevent_s *ev = (struct myevent_s *)events[i].data.ptr;
  246.             if ((events[i].events & EPOLLIN) && (ev->events & EPOLLIN)) {
  247.                 ev->call_back(ev->fd, events[i].events, ev->arg);
  248.             }
  249.             if ((events[i].events & EPOLLOUT) && (ev->events & EPOLLOUT)) {
  250.                 ev->call_back(ev->fd, events[i].events, ev->arg);
  251.             }
  252.         }
  253.     }
  255.     /* 退出前释放所有资源 */
  256.     return 0;
  257. }