Reactor 模型-阿里云开发者社区

服务器程序通常需要处理三类事件：IO 事件、信号事件和定时事件。事件处理机制主要有：Reactor（同步 IO 模型）和 Proactor（异步 IO 模型）。

1、网络编程关注的事件

网络编程关注的三个半事件

连接建立
连接断开
消息到达
消息发送

2、网络 IO 的职责

2.1、IO 检测

检测多个 IO 的就绪状态，可检测笼统的事件。

2.1.1、连接建立

EPOLLIN：接收连接，作为服务端，被动接收客户端连接，监听读事件
EPOLLOUT：主动连接，作为客户端，主动连接第三方服务，监听写事件

2.1.2、连接断开

EPOLLRDHUB，读端关闭，半连接状态，如何处理未写完的数据
EPOLLHUB，读写端关闭

2.1.3、消息到达

EPOLLIN：监听 clientfd 的读事件

2.1.4、消息发送

EPOLLOUT: 监听 clientfd 的写事件，通常是 write 没有把用户数据未发送完，注册写事件，等待写事件触发

(epoll_event)ev.events |= EPOLLOUT;
 epoll_ctl(epfd, EPOLL_CTL_ADD, clientfd, &ev);

2.2、IO 操作

只能用 IO 函数来操作，分为阻塞 IO 与非阻塞 IO。IO 函数只能检测一条连接的就绪状态以及操作一条连接的数据，可检测具体的事件 errno。

2.2.1、连接建立

主动连接 connect

EINPROGRESS正在建立
EISCONN已经建立

被动连接

listen(fd, backlog)： EWOULDBLOCK
accept

2.2.2、连接断开

主动断开

close: 关闭读写端口，close(fd) <-> shutdown(fd, SHUT_RDWR)
shutdown

关闭读端，对端写端关闭，shutdown(fd, SHUT_RD)
关闭写端，对端读端关闭，shutdown(fd, SHUT_RD)

被动断开，客户端发送 FIN 包

read = 0：读端关闭，有的需要支持半关闭状态，即仍然可以向客户端发送数据。
write = -1 && errno = EPIPE: 写端关闭。

2.2.3、连接到达

read

EWOULDBLOCK | EAGAIN：ET 模式，数据已读完
EINTER: 被中断打断，重试
ETIMEOUT: tcp 探活超时

2.2.4、消息发送

write

EWOULDBLOCK | EAGAIN：ET 模式，数据未发送完
EINTER: 被中断打断，重试
EPIPE：写端关闭

3、Reactor 模式

3.1、概念

反应堆，one eventloop per thread

组成：非阻塞 IO（IO 操作）+ IO 多路复⽤（IO 检测）
特征：基于事件循环，以事件驱动或事件回调的方式来实现业务逻辑
表述：将连接的 IO 处理转化为事件处理

3.2、面试：Reactor 为什么使用非阻塞 IO

多线程环境：一个事件同时被多线程监听，当该事件就绪，只有一个线程获得控制权，其他线程阻塞，即惊群效应。例：accept 惊群。
边缘触发：在读事件触发时，read 要在一次事件循环中把读缓冲读空。若读缓冲已经读空，继续调用 read 会阻塞。
select bug：有数据到达，select 检测 fd 可读，随后协议栈检验到该数据错误而丢弃，此时调用 read 无数据可读。若该 fd 设置没有设置成非阻塞，则 read 阻塞当前线程。

3.3、事件分类

reactor 基于事件循环，事件分为两类：EPPOLLIN 和 EPOLLOUT

if (EPPOLLIN) fd->cb;   // 读事件，连接建立(accept_cb) + 消息到达(read_cb)
 if (EPOLLOUT) fd->cb;   // 写事件，消息发送(send_cb)

回调函数分为两类：事件回调函数和错误回调函数

void (*event_callback)(int fd, int events, void *arg);
 void (*error_callback)(int fd, char * err);

其中，事件回调函数是必要的，分为：read_cb()，write_cb()，accept_cb()

read_cb()

read / recv()：把数据读到 rbuffer 里
parser()：解析数据，是否处理完数据，关注 io 下一个事件
event_register(): 重新设置 io 状态

write_cb()

把待发送的数据放入 wbuffer 中
wirte()：判断 io 是否可写，关注 io 下一个事件
event_register()：重新设置 io 状态

accept_cb()

accept(): 接收客户端的连接
设置 fd 可读
添加到 epoll 中

3.4、代码实现

reactor.h

#ifndef _REACTOR_
 #define _REACTOR_
 #include <stdio.h>
 #include <unistd.h>
 #include <fcntl.h>
 #include <sys/types.h> 
 #include <sys/socket.h>
 #include <errno.h> 
 #include <arpa/inet.h> 
 #include <assert.h> 
 #include <sys/epoll.h>
 #include <stdlib.h> 
 #include <string.h> 
 #include "chainbuffer/buffer.h"
 // #include "ringbuffer/buffer.h"
 #define MAX_EVENT_NUM 512       // 每次用户拷贝事件的最大数目
 #define MAX_CONN ((1<<16)-1)    // 事件对象的最大数目：65535
 typedef struct event_s event_t;
 typedef void (*event_callback_fn)(int fd, int events, void *privdata);
 typedef void (*error_callback_fn)(int fd, char * err);
 // reactor对象，管理io全局变量 
 typedef struct {
     int epfd;        // epfd 
     int listenfd;    // 监听的fd
     int stop;        // 停止循环标记
     event_t *events; // 存储监听的所有事件(event_t)，存储在堆上，记得释放
     int iter;        // 用于遍历events，获取没有被使用的位置
     struct epoll_event fire[MAX_EVENT_NUM]; // 用户态数组，用于拷贝io事件到用户态
 } reactor_t;
 // 事件对象，保存每个fd对应的io状态
 struct event_s {
     int fd;         // 对应的事件 fd
     reactor_t *r;   // 指向 reactor 全局对象
     buffer_t in;    // 读缓冲，待读取
     buffer_t out;   // 写缓冲，待发送
     event_callback_fn read_fn;  // 读回调
     event_callback_fn write_fn; // 写回调
     error_callback_fn error_fn; // 错误回调
 };
 int event_buffer_read(event_t *e);
 int event_buffer_write(event_t *e, void * buf, int sz);
 // 创建 reactor 对象
 reactor_t * create_reactor() {
     // 堆上申请 reactor 对象
     reactor_t *r = (reactor_t *)malloc(sizeof(*r));
     r->epfd = epoll_create(1);
     r->listenfd = 0;
     r->stop = 0;
     r->iter = 0;
     // 堆上申请 reactor 中的events数组
     r->events = (event_t*)malloc(sizeof(event_t)*MAX_CONN);
     memset(r->events, 0, sizeof(event_t)*MAX_CONN);
     memset(r->fire, 0, sizeof(struct epoll_event) * MAX_EVENT_NUM);
     // init_timer();
     return r;
 }
 // 释放 reactor 对象
 void release_reactor(reactor_t * r) {
     free(r->events);    // 释放reactor在堆上申请的events
     close(r->epfd);     // 关闭epoll
     free(r);            // 释放reactor
 }
 // 从 reactor 的事件堆上获取空闲的事件对象
 event_t * _get_event_t(reactor_t *r) {
     r->iter ++;
     // 寻找没有被使用的事件对象
     while (r->events[r->iter & MAX_CONN].fd > 0) {
         r->iter++;
     }
     return &r->events[r->iter];
 }
 // 基于事件的操作
 // 1、创建事件对象
 event_t * new_event(reactor_t *R, int fd,
     event_callback_fn rd,
     event_callback_fn wt,
     error_callback_fn err) {
     assert(rd != 0 || wt != 0 || err != 0);
     // 获取空闲的事件对象
     event_t *e = _get_event_t(R);
     // 初始化事件对象
     e->r = R;
     e->fd = fd;
     buffer_init(&e->in, 1024*16);
     buffer_init(&e->out, 1024*16);
     e->read_fn = rd;
     e->write_fn = wt;
     e->error_fn = err;
     return e;
 }
 // 2、添加事件
 int add_event(reactor_t *R, int events, event_t *e) {
     struct epoll_event ev;
     ev.events = events;
     ev.data.ptr = e;
     if (epoll_ctl(R->epfd, EPOLL_CTL_ADD, e->fd, &ev) == -1) {
         printf("add event err fd = %d\n", e->fd);
         return 1;
     }
     return 0;
 }
 // 释放事件所占空间
 void free_event(event_t *e) {
     buffer_free(&e->in);    
     buffer_free(&e->out);
 }
 // 3、删除事件
 int del_event(reactor_t *R, event_t *e) {
     epoll_ctl(R->epfd, EPOLL_CTL_DEL, e->fd, NULL);
     free_event(e);
     return 0;
 }
 // 4、修改事件，由后面两个参数决定是读事件还是写事件
 int enable_event(reactor_t *R, event_t *e, int readable, int writeable) {
     struct epoll_event ev;
     ev.events = (readable ? EPOLLIN : 0) | (writeable ? EPOLLOUT : 0);
     ev.data.ptr = e;
     if (epoll_ctl(R->epfd, EPOLL_CTL_MOD, e->fd, &ev) == -1) {
         return 1;
     }
     return 0;
 }
 // 一次事件循环
 void eventloop_once(reactor_t * r, int timeout) {
     int n = epoll_wait(r->epfd, r->fire, MAX_EVENT_NUM, timeout);
     for (int i = 0; i < n; ++i) {
         struct epoll_event *e = &r->fire[i];  // 获取事件
         int mask = e->events;                 // 获取事件类型
         // 用 io 函数捕获具体的错误信息
         if (e->events & EPOLLERR) mask |= EPOLLIN | EPOLLOUT;
         // 用 io 函数捕获断开的具体信息
         if (e->events & EPOLLHUP) mask |= EPOLLIN | EPOLLOUT;
         event_t *et = (event_t*) e->data.ptr; // 获取事件关联的用户数据
         // 处理读事件
         if (mask & EPOLLIN) {
             if (et->read_fn) {
                 et->read_fn(et->fd, EPOLLIN, et);   // 执行读回调
             }          
         }
         // 处理写事件
         if (mask & EPOLLOUT) {
             if (et->write_fn) {
                 et->write_fn(et->fd, EPOLLOUT, et); // 执行写回调
             }     
             else {
                 uint8_t *buf = buffer_write_atmost(&et->out);
                 event_buffer_write(et, buf, buffer_len(&et->out));
             }
         }
     }
 }
 // 停止事件循环
 void stop_eventloop(reactor_t * r) {
     r->stop = 1;
 }
 // 事件循环
 void eventloop(reactor_t * r) {
     while (!r->stop) {
         // int timeout = find_nearest_expire_timer();
         eventloop_once(r, /*timeout*/ -1);
         // expire_timer();
     }
 }
 // 设置非阻塞fd
 int set_nonblock(int fd) {
     int flag = fcntl(fd, F_GETFL, 0);
     return fcntl(fd, F_SETFL, flag | O_NONBLOCK);
 }
 // 创建服务器
 int create_server(reactor_t *R, short port, event_callback_fn func) {
     // 1、socket
     int listenfd = socket(AF_INET, SOCK_STREAM, 0);
     if (listenfd < 0) {
         printf("create listenfd error!\n");
         return -1;
     }
     struct sockaddr_in addr;
     memset(&addr, 0, sizeof(struct sockaddr_in));
     addr.sin_family = AF_INET;
     addr.sin_port = htons(port);
     addr.sin_addr.s_addr = INADDR_ANY;
     // 设置地址可重用
     int reuse = 1;
     if (setsockopt(listenfd, SOL_SOCKET, SO_REUSEADDR, (void *)&reuse, sizeof(int)) == -1) {
         printf("reuse address error: %s\n", strerror(errno));
         return -1;
     }
     // 2、bind
     if (bind(listenfd, (struct sockaddr*)&addr, sizeof(struct sockaddr_in)) < 0) {
         printf("bind error %s\n", strerror(errno));
         return -1;
     }
     // 3、listen
     if (listen(listenfd, 5) < 0) {
         printf("listen error %s\n", strerror(errno));
         return -1;
     }
     // 设置 listenfd 非阻塞 
     if (set_nonblock(listenfd) < 0) {
         printf("set_nonblock error %s\n", strerror(errno));
         return -1;
     }
     R->listenfd = listenfd;
     // 注册读事件
     event_t *e = new_event(R, listenfd, func, 0, 0);
     add_event(R, EPOLLIN, e);
     printf("listen port : %d\n", port);
     return 0;
 }
 // 读数据
 int event_buffer_read(event_t *e) {
     int fd = e->fd; 
     int num = 0;    // 读取的数据总量
     while (1) {
         // TODO: dont use char buf[] here
         char buf[1024] = {0};
         int n = read(fd, buf, 1024);
         // 1、read=0, 服务器收到FIN包，半关闭状态
         // Todo: 半关闭状态逻辑处理，参考 skynet
         if (n == 0) { // 
             printf("close connection fd = %d\n", fd);
             if (e->error_fn) {
                 e->error_fn(fd, "close socket");
             }   
             del_event(e->r, e);
             close(fd);
             return 0;
         } 
         // 2、read=-1，读异常
         else if (n < 0) {
             // 2.1、EINTR：中断，重试
             if (errno == EINTR) {
                 continue;
             }
             // 2.2、EWOULDBLOCK：阻塞，读缓冲区为空 
             if (errno == EWOULDBLOCK) {
                 break;
             }
             // 其他错误，执行错误回调，删除该事件，关闭当前连接
             printf("read error fd = %d err = %s\n", fd, strerror(errno));
             if (e->error_fn)
                 e->error_fn(fd, strerror(errno));
             del_event(e->r, e);
             close(fd);
             return 0;
         } 
         // 3、read>0, 正常，读取数据，处理业务逻辑
         else {
             printf("recv data from client:%s", buf);
             buffer_add(&e->in, buf, n);
         }
         num += n;
     }
     return num;
 }
 // 向对端发送数据
 int _write_socket(event_t *e, void * buf, int sz) {
     int fd = e->fd;
     while (1) {
         int n = write(fd, buf, sz);
         // 1、write=-1，写异常
         if (n < 0) {
             // 1.1、EINTR：中断，重试
             if (errno == EINTR) {
                 continue;
             }  
             // 1.2、EWOULDBLOCK：阻塞，需要注册写事件，等待下一次发送未发送完的数据
             if (errno == EWOULDBLOCK) {
                 break;
             }
             // 其他错误，执行错误回调，删除该事件，关闭当前连接   
             if (e->error_fn) {
                 e->error_fn(fd, strerror(errno));
             }     
             del_event(e->r, e);
             close(e->fd);
         }
         return n;
     }
     return 0;
 }
 // 写数据
 int event_buffer_write(event_t *e, void * buf, int sz) {
     // 指向用户写缓冲
     buffer_t *r = &e->out;
     // 1、用户写缓冲已满，开始发送
     if (buffer_len(r) == 0) {
         // 向对端发送数据
         int n = _write_socket(e, buf, sz);
         // 1.1、本次数据未发送完，未发送的数据写入缓冲，并注册写事件
         if (n == 0 || n < sz) {
             // 1.1、将没有发送完的数据写入缓冲区
             buffer_add(&e->out, (char *)buf + n, sz - n);
             // 1.2、注册写事件，等待下次事件触发接着发送
             enable_event(e->r, e, 1, 1);
             return 0;
         } 
         // 1.2、本次没有发送数据
         else if (n < 0) {
             return 0;
         }
         // 1.3、本次数据发送完成    
         return 1;
     }
     // 2、用户写缓冲未满，写入缓冲，等待发送
     buffer_add(&e->out, (char *)buf, sz);
     return 1;
 }
 #endif

server.c

#include "reactor.h"
 void read_cb(int fd, int events, void *privdata) {
     event_t *e = (event_t *)privdata;
     int n = event_buffer_read(e); // 将网络中读缓冲区的数据拷贝到用户态缓冲区
     if (n > 0) {
         // buffer_search 检测是否是一个完整的数据包
         int len = buffer_search(&e->in, "\n", 1);
         if (len > 0 && len < 1024) {
             char buf[1024] = {0};
             buffer_remove(&e->in, buf, len);
             event_buffer_write(e, buf, len);
         }
     }
 }
 void accept_cb(int fd, int events, void *privdata) {
     event_t *e = (event_t*) privdata;
     struct sockaddr_in addr;
     memset(&addr, 0, sizeof(struct sockaddr_in));
     socklen_t len = sizeof(addr);
     int clientfd = accept(fd, (struct sockaddr*)&addr, &len);
     if (clientfd <= 0) {
         printf("accept failed\n");
         return;
     }
     char str[INET_ADDRSTRLEN] = {0};
     printf("recv from %s at port %d\n", inet_ntop(AF_INET, &addr.sin_addr, str, sizeof(str)),
         ntohs(addr.sin_port));
     event_t *ne = new_event(e->r, clientfd, read_cb, 0, 0);
     add_event(e->r, EPOLLIN, ne);
     set_nonblock(clientfd);
 }
 int main() {
     reactor_t *R = create_reactor();
     if (create_server(R, 8989, accept_cb) != 0) {
         release_reactor(R);
         return 1;
     }
     eventloop(R);
     release_reactor(R);
     return 0;
 }

4、Reactor 应用

4.1、redis

单 reactor 模型

场景：KV 内存数据库
特点：单线程处理业务逻辑（命令）
原因：数据结构高效，操作命令速度快
优化：read, decode 和 encode, send 阶段采用多线程 (redis 6.0以后)

skynet 采用的也是单 reactor 模型。

redis

4.2、memcached

多 reactor 模型

场景：KV 内存数据库
特点：多线程处理业务逻辑（命令）
原因：更高程度并发处理业务

memcached

4.3、nginx

多进程

场景：反向代理
特点：多进程处理业务逻辑
原因：业务类型复杂，通过进程隔离运行环境避免加锁

master 进程根据配置文件生成一个监听相应端口的 socket，然后 fork 多个 worker 子进程，每个 worker 进程都可以监听 socket 消息。为了解决 accept 惊群问题，当一个连接到来，每个 worker 都能收到通知，但是只有一个 worker 能够建立连接，其他 worker 连接失败。nginx 通过互斥锁 accept_mutex 控制 worker 进程接收连接，只有获得了 accept_mutex 的进程才会添加 accept 事件。

nginx 使用变量 ngx_accept_disabled 来控制是否去竞争 accept_mutex 锁。

ngx_accept_disabled = nginx单进程的连接总数/8 - 空闲连接数量

当 ngx_accept_disabled > 0，不会尝试获取 accept_mutex 锁，该值越大，让出的机会越多，其他进程获取锁的机会越大。不添加 accept，每个 worker 进程的连接数得到控制，实现了连接平衡。

nginx

Reactor 模型