【epoll】epoll 多路复用和 Reactor 设计思想

目录

1、Reactor 设计思想

文章相关视频讲解：

C/C++ Linux 服务器开发高级架构学习视频点击：C/C++Linux服务器开发/Linux后台架构师-学习视频

epoll原理剖析以及reactor模型应用

linux epoll网络编程细节处理

小前言：

Reactor 必要

传统 OIO 模式

2.2 Reactor 模式

2.3 单线程 Reactor 模式

单 Reactor 多线程模式：

2.4 多线程 Reactor 模式

封装 Epoll 实现并发

Reactor 模式：

封装 Epoll 实现 reactor 模式的高性能并发服务器

epoll 的 api

Reactor 模式:

EPOLL 实现的要点

1、Reactor 设计思想

小前言：

reactor 是对 epoll 的一层封装 ，epoll 是对 io 进行管理，reactor 将对 io 的管理转化为对事件的管理

Reactor 必要

传统 OIO 模式

如图 2.1 所示为传统 IO 模式处理示意图：

图中所示一般是一个请求一个单独的处理线程。

缺点：server 的 accpet 操作是阻塞的，业务处理中的 handler 中的读写请求也是阻塞的。那么这样的一种 IO 模式将会导致一个线程的请求没有处理完成无法处理下一个请求，这样就大大降低了吞吐量，这将是一个严重的问题。

为了解决这种问题就出现了一个经典的模式——Connection Per Thread 即一个线程处理一个请求。

对于每一个新的请求都会分配一个新的线程来处理，这样的好处就是每个 socket 的请求相互之间不受影响，每个请求的业务逻辑相互之间也不影响。任何 socket 的读写操作都不会影响到后面的请求。

缺点：不是每个链接都有请求发生，这样就浪费了很多的线程资源。

这个时候可以采用多路复用 IO 模型的方式来处理 IO 事件，使用 Reactor 将响应 IO 事件和业务处理分开，一个或多个线程来处理 IO 事件，然后将就绪得到事件分发到业务处理 handlers 线程去异步非阻塞处理。

2.2 Reactor 模式

2.3 单线程 Reactor 模式

什么是单线程 Reactor 模式，单线程模式采用一个 Reactor 线程来【处理套接字、新连接的创建】，并且【将接收到的请求分发到处理器 Handler 中】。

如图 2.2 为简单的单线程 Reactor 模式示意图，Reactor 和数据处理(handler)都在一个线程里，图 2.2 参考 doug lea 论文《Scalable IO in Java》论文。

图 2.2 单线程 Reactor 模式示意图

单 Reactor 多线程模式：

2.4 多线程 Reactor 模式

多线程 reactor 模式的设计思想就是将 handler 线程放入到线程次中，在多核的情况下也可以考虑多个 Selector 选择器来处理事件，如图 2.3 为简单的多线程 Reactor 示意图;

图 2.3 多线程 Reactor 模式示意图

关于 C/C++ Linux 后端开发网络底层原理知识 点击 学习资料 获取，内容知识点包括 Linux，Nginx，ZeroMQ，MySQL，Redis，线程池，MongoDB，ZK，Linux 内核，CDN，P2P，epoll，Docker，TCP/IP，协程，DPDK 等等。

封装 Epoll 实现并发

第一次学 epoll 时，容易错误的认为 epoll 可以实现并发，其实正确的说法是借助 epoll 可以实现高性能并发服务器，epoll 只是提供了 IO 复用，在 IO 复用，真正的并发只能通过线程进程实现。

Reactor 模式：

Reactor 模式实现非常简单，使用同步 IO 模型，即业务线程处理数据需要主动等待或询问，主要特点是利用 epoll 监听 listen 描述符是否有响应，及时将客户连接信息放于一个队列，epoll 和队列都是在主进程/线程中，由子进程/线程来接管各个描述符，对描述符进行下一步操作，包括 connect 和数据读写。主程读写就绪事件。

大致流程图如下：

Preactor 模式：

Preactor 模式完全将 IO 处理和业务分离，使用异步 IO 模型，即内核完成数据处理后主动通知给应用处理，主进程/线程不仅要完成 listen 任务，还需要完成内核数据缓冲区的映射，直接将数据 buff 传递给业务线程，业务线程只需要处理业务逻辑即可。

大致流程如下：

封装 Epoll 实现 reactor 模式的高性能并发服务器

epoll 的 api

首先介绍 epoll 的 api

int epoll_create(int size);  // 创建epfdint epoll_ctl(int epfd, int op, int fd, struct epoll_event *event); //向epfd注册(fd,event)
// epoll_event结构体定义struct epoll_event { __uint32_t events; /* epoll 事件 */ epoll_data_t data; /* 传递的数据,用于处理ready的fd,获得上下文关系 */}// data联合体,一般用其指针域ptr,因为要从这个data读取到上下文信息typedef union epoll_data {    void *ptr;    int fd;    uint32_t u32;    uint64_t u64;} epoll_data_t;
// op宏 /* EPOLL_CTL_ADD(注册新的fd到epfd) * EPOLL_CTL_MOD(修改已经注册的fd的监听事件) * EPOLL_CTL_DEL(epfd删除一个fd) */* * events : {EPOLLIN, EPOLLOUT, EPOLLPRI,             EPOLLHUP, EPOLLET, EPOLLONESHOT} */
int epoll_wait(int epfd, struct epoll_event *event,             int maxevents, int timeout);  //用于轮询注册的fd,若满足相应的注册事件,                                           //  则结束epoll_wait阻塞/* @param timeout 超时时间 *     -1: 永久阻塞 *     0: 立即返回，非阻塞 *     >0: 指定微秒*/

Reactor 模式:

io 模式的历程:

单线程,一般阻塞->多线程,一般阻塞（一条连接一线程）->线程池(减少线程创建销毁开销)->reactor(更小粒度的线程)

所谓更小的粒度的线程是指,传统的多线程是一个连接一个线程,粒度太大,比如可以把一个连接继续细分成三个步骤:read,process,send 三个步骤,每个步骤占一个线程,处理完后交给主线程调度,进入下一个处理模块

EPOLL 实现的要点

0. 创建 epoll_fd = epoll_create(MAX_EVENT+1)

1. 维护 event 数组

1.1 一个交给 epoll_wait 维护(空的放进去,ready 的出来)

1.1.1 特定的结构体 struct epoll_event

1.2 一个自己维护(维持对所有注册事件的监控)(全局)

1.2.1 由于自己维护,想怎么写怎么写,一般的会让 epoll_event.data.ptr = myevent,以作为回调函数的参数,保持一个上下文关系

2. 创建并初始化事件(epoll_event)

2.1 结构体有两个字段 event 和 data , event 是 EPOLLIN 这样的宏,data 的作用是记录一些消息,这样 ready 的时候可以访问这个消息,比如 fd, 回调函数指针,status 等等

3. 向 epfd 注册事件

3.1 epoll_ctl(epfd , op_macro , target_fd , &epoll_event )

3.2 op_macro 是代表 epoll_ctl 的类型,诸如 EPOLL_CTL_ADD

3.3 target_fd 是要监听的 fd,比如 tcp 监听套接字 ls_socket,epoll_event 就是当事件发生时,epoll_wait()里面 event[]将会出现的结构体

4. 写回调函数

4.1 这是 reactor 模式的重点.

4.2 典型的,检测到 ls_socket 可读后(epoll_wait 不再阻塞),进入回调函数(通过 event[i].data.ptr->callback),假如命名为 accept_fn(),在这个函数简单来看要做的事就是

1.conn_socket = accept(ls_socket)

2.创建,初始化 epoll_event 并注册到 epfd(当然还要加进自己维护的 fd 列表),由于 epoll 是轮询的模式,需要将 conn_socket 用 fcntl 设为 O_NONBLOCK,非阻塞.

4.3 这个不像 select,需要每次重新加入 fd 到列表里面,注册一次即可

4.4 conn_socket 被通过事件 EPOLLIN 注册到 epfd,下一步马上的,epoll_wait 检测出 conn_socket 可读(假设确实可读),然后回调进入 recvdata 函数(需要在创建并初始化 epoll_event 指定, 实际上是自定义一个结构体,让 data.ptr 指向这个结构体就行),一般的,就以这个结构体组成自己维护的 fd list,

4.5 recvdata 函数首先调用 recv(fd,my_event->buf,...),把待发送的数据存在 my_event->buf , 然后修改(fd,myevent)到 epfd 为发送事件 EPOLLOUT,以及改变回调函数为 callback_send

4.6 来到 epoll_wait,检测到该 fd 可写,则回调进入 senddata 函数,该函数调用 send(fd,myevent->buf)发送数据,然后修改 fd 的注册事件为 EPOLLIN(即 EPOLL_CTLMOD),清空 my_event->buf....如此反复

注意点:

1. callback 函数如何获得 my_event? 我们在 epoll_event 中能获得 event 和 data,在 data.ptr 中找到 my_event,典型的 my_event 可能包括回调函数指针,event_type,fd,buf[BUFLEN],last_active_time,...

nfd = epoll_wait(epfd , epoll_events , MAX_EVENT+1 , 1000)assert(nfd>0)for i in range(nfd):  my_event* ev = (my_event*) epoll_events[i].data.ptr  // if (epoll_events[i].events& EPOLLIN){}  用按位与的方式检测相等,因为这种flag宏一般都是位不相同的                                    // 用回调函数的方法时,不需要比较event_type,直接调用函数即可  // ev->callback_fn(ev)  不可以    callback_fn的原型应该是void(*callvback_fn)(void*),                                  // 因为定义callback原型的时候,ev还是不完整的类型  ev->callback_fn((void*)ev)