文章目录

• • 1、网络编程关注的事件
  • 2、网络 IO 的职责
  • • 2.1、IO 检测
    • • 2.1.1、连接建立
      • 2.1.2、连接断开
      • 2.1.3、消息到达
      • 2.1.4、消息发送
    • 2.2、IO 操作
    • • 2.2.1、连接建立
      • 2.2.2、连接断开
      • 2.2.3、连接到达
      • 2.2.4、消息发送
  • 3、Reactor 模式
  • • 3.1、概念
    • 3.2、面试：Reactor 为什么使用非阻塞 IO
    • 3.3、事件分类
    • 3.4、代码实现
  • 4、Reactor 应用
  • • 4.1、redis
    • 4.2、memcached
    • 4.3、nginx

服务器程序通常需要处理三类事件：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 
#include 
#include 
#include  
#include 
#include  
#include  
#include  
#include 
#include  
#include  #include "chainbuffer/buffer.h"
// #include "ringbuffer/buffer.h"#define MAX_EVENT_NUM 512       // 每次用户拷贝事件的最大数目
#define MAX_CONN ((1<<16)-1)    // 事件对象的最大数目：65535typedef 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;    // 监听的fdint 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;         // 对应的事件 fdreactor_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在堆上申请的eventsclose(r->epfd);     // 关闭epollfree(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、socketint 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、bindif (bind(listenfd, (struct sockaddr*)&addr, sizeof(struct sockaddr_in)) < 0) {printf("bind error %s\n", strerror(errno));return -1;}// 3、listenif (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[] herechar buf[1024] = {0};int n = read(fd, buf, 1024);// 1、read=0, 服务器收到FIN包，半关闭状态// Todo: 半关闭状态逻辑处理，参考 skynetif (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 模型。

4.2、memcached

多 reactor 模型

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

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 进程的连接数得到控制，实现了连接平衡。