Redis 源码学习之事件驱动

内容简介：Redis基于多路复用技术实现了一套简单的事件驱动库，代码在ae.h、ae.c以及ae_epoll.c、ae_evport.c和ae_kqueue.c、ae_select.c这几个文件中。其中ae表示的是antirez eventloop的意思。

Redis基于多路复用技术实现了一套简单的事件驱动库，代码在ae.h、ae.c以及ae_epoll.c、ae_evport.c和ae_kqueue.c、ae_select.c这几个文件中。其中ae表示的是antirez eventloop的意思。

Redis里面包含两种事件类型:FileEvent和TimeEvent。

Redis采用IO多路复用技术，所有的事件都是在一个线程中进行处理。Redis的事件驱动模型可以以以下为代码进行表示:

在一个死循环中等待事件的到来，然后对事件进行处理，以此往复。这就是一个最经典的网络编程模型。

1.基本数据结构

aeEventLoop

aeEventLoop是Redis中事件驱动模型的核心，封装了整个事件循环，其中每个字段解释如下：

• maxfd:已经接受的最大的文件描述符。
• setsize:当前循环中所能容纳的文件描述符的数量。
• timeEventNextId:下一个时间事件的ID.
• lastTime:上一次被访问的时间，用来检测系统时钟是否被修改。
• events:指针，指向保存所有注册的事件的数组首地址。
• fired:指针，保存所有已经买被触发的事件的数组首地址。
• timeEventHead:Redis用一个链表来存储所有的时间事件，timeEventHead是指向这个链表的首节点指针。
• stop:停止整个事件循环。
• apiData:指针，指向epoll结构。
• beforeSleep:函数指针。每次实现循环的时候，在阻塞直到时间到来之前，会先调用这个函数。

aeFileEvent和aeTimeEvent

这两个结构分别表示文件事件和时间事件，定义如下

其中mask表示文件事件类型掩码，可以是AE_READABLE表示是可读事件，AE_WRITABLE为可写事件。aeFileProc是函数指针。

aeFiredEvent

aeFiredEvent结构表示一个已经被触发的事件，结果如下:

fd表示事件发生在哪个文件描述符上面，mask用来表示具体事件的类型。

aeApiState

Redis底层采用IO多路复用技术实现高并发，具体实现可以采用kqueue、select、epoll等技术。对于Linux来说，epoll的性能要优于select，所以以epoll为例来进行分析。

aeApiState封装了跟epoll相关的数据，epfd保存epoll_create()返回的文件描述符。

具体实现细节

事件循环启动:aeMain()

事件驱动的启动代码位于ae.c的aeMain()函数中，代码如下:

从aeMain()方法中可以看到，整个事件驱动是在一个while()循环中不停地执行aeProcessEvents()方法，在这个方法中执行从客户端发送过来的请求。

初始化:aeCreateEventLoop()

aeEventLoop的初始化是在aeCreateEventLoop()方法中进行的，这个方法是在server.c中的initServer()中调用的。实现如下:

在这个方法中主要就是给aeEventLoop对象分配内存然后并进行初始化。其中关键的地方有：

1、调用aeApiCreate()初始化epoll相关的数据。aeApiCreate()实现如下:

在aeApiCreate()方法中主要完成以下三件事：
1. 分配aeApiState结构需要的内存。
2. 调用epoll_create()方法生成epoll的文件描述符，并保存在aeApiState.epfd字段中。
3. 把第一步分配的aeApiState的内存地址保存在EventLoop->apidata字段中。

2、初始化events中的mask字段为为AE_NONE。

生成fileEvent:aeCreateFileEvent()

Redis使用aeCreateFileEvent()来生成fileEvent，代码如下:

aeCreateFileEvent()方法主要做了一下三件事:

1. 检查新增的fd是否超过所能容纳最大值。
2. 调用aeApiAddEvent()方法把对应的fd以mask模式添加到epoll监听器中。

设置相应的字段值。其中最关键的步骤是第二步，aeApiAddEvent()方法如下：

生成timeEvent:aeCreateTimeEvent()

aeCreateTimeEvent()方法主要是用来生成timeEvent节点，其实现比较简单，代码如下所示:

处理timeEevnt:processTimeEvents()

Redis在processTimeEvents()方法中来处理所有的timeEvent，实现如下:

在这个方法中会

1. 判断系统时间有没有调整过，如果调整过，则会把timeEvent链表中的所有的timeEvent的触发时间设置为0，表示立即执行。
2. 对timeEvent链表进行遍历，对于每个timeEvent节点，如果有：
   1. 如果已经被标记为删除(AE_DELETED_EVENT_ID)，则立即释放对应节点内存，遍历下个节点。
   2. 如果id大于maxId,则表示当前节点为本次循环中新增节点，咋本次循环中不错处理，继续下个节点。
   3. 如果当前节点的触发时间大于当前时间，则调用对应节点的timeProc()方法执行任务。根据timeProc()方法的返回，又分为两种情况:
      1. 返回为AE_NOMORE，表示当前timeEvent节点属于一次性事件，标记该节点ID为AE_DELETED_EVENT_ID,表示删除节点，该节点将会在下一轮的循环中被删除。
      2. 返回不是AE_NOMORE，表示当前timeEvent节点属于周期性事件，需要多次执行，调用aeAddMillisecondsToNow()方法设置下次被执行时间。

处理所有事件:aeProcessEvents()

Redis中所有的事件，包括timeEvent和fileEvent都是在aeProcessEvents()方法中进行处理的，刚方法实现如下：

该方法的入参flag表示要处理哪些事件，可以取以下几个值 ：

• AE_ALL_EVENTS:timeEvent和fileEvent都会处理。
• AE_FILE_EVENTS:只处理fileEvent。
• AE_TIME_EVENTS:只处理timeEvent。
• AE_DONT_WAIT:要么立马返回，要么处理完那些不需要等待的事件之后再立马返回。

aeProcessEvents()方法会做下面几件事:

1. 判断传入的flag的值，如果既不包含AE_TIME_EVENTS也不包含AE_FILE_EVENTS则直接返回。
2. 计算如果有aeFileEvent事件需要进行处理，则先计算epoll_wait()方法需要阻塞等待的时间，计算方式如下:
   1. 先从aeTimeEvent事件链表中找到最近的需要被触发的aeTimeEvent节点并计算需要被触发的时间，该被触发时间则为epoll_wait()需要等待的时间。
   2. 如果没有找到最近的aeTimeEvent节点，表示没有aeTimeEvent节点被加入链表，则判断传入的flags是否包含AE_DONT_WAIT选项，则设置epoll_wait()需要等待时间为0，表示立即返回。
   3. 如果没有设置AE_DONT_WAIT,则设置需要等待时间为NULL,表示epoll_wait()一直阻塞等待知道有fileEvent事件到来。
3. 调用aeApiPoll()方法阻塞等待事件的到来，阻塞时间为第二步中计算的时间。aeApiPoll()实现如下：
   
   static int aeApiPoll(aeEventLoop *eventLoop, struct timeval *tvp) { aeApiState *state = eventLoop->apidata; int retval, numevents = 0; // 1. 根据传入的tvp计算需要等待时间，然后调用epoll_wait()进行阻塞等待 retval = epoll_wait(state->epfd,state->events,eventLoop->setsize, tvp ? (tvp->tv_sec*1000 + tvp->tv_usec/1000) : -1); // 有事件到来 if (retval > 0) { int j; numevents = retval; for (j = 0; j < numevents; j++) { int mask = 0; struct epoll_event *e = state->events+j; // 2. 计算到来的event的类型 if (e->events & EPOLLIN) mask |= AE_READABLE; if (e->events & EPOLLOUT) mask |= AE_WRITABLE; if (e->events & EPOLLERR) mask |= AE_WRITABLE; if (e->events & EPOLLHUP) mask |= AE_WRITABLE; // 3. 把有事件发生的fd以及对应的mask类型拷贝到eventloop->fired数组中 eventLoop->fired[j].fd = e->data.fd; eventLoop->fired[j].mask = mask; } } return numevents; }

   1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29

   static int aeApiPoll(aeEventLoop *eventLoop, struct timeval *tvp) {     aeApiState *state = eventLoop->apidata;     int retval, numevents = 0;       // 1. 根据传入的tvp计算需要等待时间，然后调用epoll_wait()进行阻塞等待     retval = epoll_wait(state->epfd,state->events,eventLoop->setsize,             tvp ? (tvp->tv_sec*1000 + tvp->tv_usec/1000) : -1);     // 有事件到来     if (retval > 0) {         int j;           numevents = retval;         for (j = 0; j < numevents; j++) {             int mask = 0;             struct epoll_event *e = state->events+j;               // 2. 计算到来的event的类型             if (e->events & EPOLLIN) mask |= AE_READABLE;             if (e->events & EPOLLOUT) mask |= AE_WRITABLE;             if (e->events & EPOLLERR) mask |= AE_WRITABLE;             if (e->events & EPOLLHUP) mask |= AE_WRITABLE;               // 3. 把有事件发生的fd以及对应的mask类型拷贝到eventloop->fired数组中             eventLoop->fired[j].fd = e->data.fd;             eventLoop->fired[j].mask = mask;         }     }     return numevents; }

   
   aeApiPoll()会做下面几件事:
   1. 根据传入的tvp计算需要阻塞的时间，然后调用epoll_wait()进行阻塞等待。
   2. 有事件到来之后先计算对应事件的类型。
   3. 把事件发生的fd以及对应的类型mask拷贝到fired数组中。
4. 从aeApiPoll()方法返回之后，所有事件已经就绪了的fd以及对应事件的类型mask已经保存在eventLoop->fired[]数组中。依次遍历fired数组，根据mask类型，执行对应的frileProc()或者wfileProce()方法。
5. 如果传入的flags中有AE_TIME_EVENTS,则调用processTimeEvents()执行所有已经到时间了的timeEvent。

以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持 码农网

查看所有标签

本站部分资源来源于网络，本站转载出于传递更多信息之目的，版权归原作者或者来源机构所有，如转载稿涉及版权问题，请联系我们。