zoukankan html css js c++ java

redis5.0.2源码分析——（AE）事件模型

整个分析的代码多数都在ae.c、ae.h、ae_epoll.c、ae_evport.c、ae_kqueue.c、ae_select.c，某些不在这些文件的函数会特别指出位置

一、Redis的事件模型库

　　Redis服务器是一个事件驱动程序。下面先来简单介绍什么是事件驱动。

　　所谓事件驱动，就是当你输入一条命令并且按下回车，然后消息被组装成Redis协议（RESP协议）的格式发送给Redis服务器，这就会产生一个事件，Redis服务器会接收该命令，处理该命令和发送回复，而当你没有与服务器进行交互时，那么服务器就会处于阻塞等待状态，会让出CPU从而进入睡眠状态，当事件触发时，就会被操作系统唤醒。事件驱动使CPU更高效的利用。

　　事件驱动是一种概括和抽象，也可以称为I/O多路复用（I/O multiplexing），它的实现方式各个系统都不同。

　　大家到网上Google“Redis libevent”就可以搜到Redis为什么没有选择libevent以及libev为其事件模型库，而是自己写了一个事件模型。从代码中可以看到它主要支持了epoll、select、kqueue、以及基于Solaris的event ports。主要提供了对两种类型的事件驱动：

1.IO事件（文件事件），包括有IO的读事件和写事件。
2.定时器事件，包括有一次性定时器和循环定时器。

二、事件的抽象

　　Redis将这两个事件分别抽象成一个数据结构来管理。

2.1 文件事件结构（包括IO读事件和写事件）

 1 /* File event structure 文件事件结构*/
 2 typedef struct aeFileEvent {
 3     // 文件时间类型：AE_NONE，AE_READABLE，AE_WRITABLE
 4     int mask; /* one of AE_(READABLE|WRITABLE|BARRIER) */
 5     // 可读处理函数
 6     aeFileProc *rfileProc;
 7     // 可写处理函数
 8     aeFileProc *wfileProc;
 9     // 客户端传入的数据
10     void *clientData;
11 } aeFileEvent;

　　其中rfileProc和wfileProc成员分别为两个函数指针，他们的原型为

1 //IO读写事件回调函数
2 typedef void aeFileProc(struct aeEventLoop *eventLoop, int fd, void *clientData, int mask);

　　这个函数是回调函数，如果当前文件事件所指定的事件类型发生时，则会调用对应的回调函数处理该事件。

　　当事件就绪时，我们需要知道文件事件的文件描述符还有事件类型才能对于锁定该事件，因此定义了aeFiredEvent结构统一管理：

1 /* A fired event 就绪事件*/
2 typedef struct aeFiredEvent {
3     // 就绪事件的文件描述符
4     int fd;
5     // 就绪事件类型：AE_NONE，AE_READABLE，AE_WRITABLE
6     int mask;
7 } aeFiredEvent;

2.2 时间事件结构

 1 /* Time event structure 定时事件结构体，是一个双向链表*/
 2 typedef struct aeTimeEvent {
 3     // 时间事件的id
 4     long long id; /* time event identifier. */
 5     // 时间事件到达的时间的秒数
 6     long when_sec; /* seconds */
 7     // 时间事件到达的时间的毫秒数
 8     long when_ms; /* milliseconds */
 9     // 时间事件处理的回调函数
10     aeTimeProc *timeProc;
11     // 时间事件删除时的回调函数
12     aeEventFinalizerProc *finalizerProc;
13     // 客户端传入的数据
14     void *clientData;
15     // 指向前一个时间事件
16     struct aeTimeEvent *prev;
17     // 指向下一个时间事件
18     struct aeTimeEvent *next;
19 } aeTimeEvent;

从这个结构中可以看出，时间事件表是一个链表，因为它有一个next指针域，指向下一个时间事件。

和文件事件一样，当时间事件所指定的事件发生时，也会调用对应的回调函数，结构成员timeProc和finalizerProc都是回调函数，函数原型如下：

1 //定时器事件回调函数
2 typedef int aeTimeProc(struct aeEventLoop *eventLoop, long long id, void *clientData);
3 //删除定时事件的回调函数
4 typedef void aeEventFinalizerProc(struct aeEventLoop *eventLoop, void *clientData);

虽然对文件事件和时间事件都做了抽象，Redis仍然需要对事件做整体抽象，于是定义了aeEventLoop结构。

2.3 事件状态结构

2.3.1 aeEventLoop结构

　　aeEventLoop保存了文件事件、定时事件、底层IO多复用模型等诸多信息

 1 /* State of an event based program 事件状态结构*/
 2 typedef struct aeEventLoop {
 3     //当前注册的最大文件描述符
 4     int maxfd;   /* highest file descriptor currently registered */
 5     /**
 6      * 指定事件循环要监听的文件描述符集合的大小。这个值与配置文件中得maxclients有关。
 7      *
 8      * setsize参数表示了eventloop可以监听的网络事件fd的个数（不包含超时事件），
 9      * 如果当前监听的fd个数超过了setsize，eventloop将不能继续注册。
10      */
11     int setsize; /* max number of file descriptors tracked */
12     // 下一个时间事件的ID
13     long long timeEventNextId;
14     // 最后一次执行事件的时间
15     time_t lastTime;     /* Used to detect system clock skew */
16     /**
17      * 存放所有注册的读写事件，是大小为setsize的数组。内核会保证新建连接的fd是当前可用描述符的最小值，
18      * 所以最多监听setsize个描述符，那么最大的fd就是setsize - 1。这种组织方式的好处是，可以以fd为下标，
19      * 索引到对应的事件，在事件触发后根据fd快速查找到对应的事件。
20      */
21     aeFileEvent *events; /* Registered events */
22     //存放触发的读写事件。同样是setsize大小的数组。
23     aeFiredEvent *fired; /* Fired events */
24     //redis将定时器事件组织成链表，这个属性指向表头。
25     aeTimeEvent *timeEventHead;
26     // 事件处理开关
27     int stop;
28     //存放epoll、select等实现相关的数据。
29     void *apidata; /* This is used for polling API specific data */
30     //事件循环在每次迭代前会调用beforesleep执行一些异步处理。
31     aeBeforeSleepProc *beforesleep;
32     // 执行处理事件之后的回调函数
33     aeBeforeSleepProc *aftersleep;
34 } aeEventLoop;

2.3.2 底层IO多路复用模式的选择

　　aeEventLoop结构保存了一个void *类型的万能指针apidata，是用来保存轮询事件的状态的，也就是保存底层调用的多路复用库的事件状态，关于Redis的多路复用库的选择，Redis包装了常见的select epoll evport kqueue，他们在编译阶段，根据不同的系统选择性能最高的一个多路复用库作为Redis的多路复用程序的实现，而且所有库实现的接口名称都是相同的，因此Redis多路复用程序底层实现是可以互换的。具体选择库的源码为

 1 /**
 2  * Include the best multiplexing layer supported by this system.
 3  * The following should be ordered by performances, descending.
 4  * 包括该系统支持的最佳复用层。性能依次下降，IO复用的选择，性能依次下降，Linux支持 "ae_epoll.c" 和 "ae_select.c"
 5  */
 6 #ifdef HAVE_EVPORT
 7 #include "ae_evport.c"
 8 #else
 9     #ifdef HAVE_EPOLL
10     #include "ae_epoll.c"
11     #else
12         #ifdef HAVE_KQUEUE
13         #include "ae_kqueue.c"
14         #else
15         #include "ae_select.c"
16         #endif
17     #endif
18 #endif

2.3.3 查看当前系统所使用的多路复用库

也可以通过Redis客户端的命令来查看当前选择的多路复用库，info server。

1 127.0.0.1:6379> INFO server
2 # Server
3 ……
4 multiplexing_api:epoll
5 ……

2.3.4 epoll多路复用库

2.3.4.1 aeApiState结构

　　既然知道了多路复用库的选择，那么我们来查看一下apidata保存的epoll模型的事件状态结构，实际上就是将需要监听的事件和epoll_create创建的文件描述符epfd封装到了一起，位于ae_epoll.c文件中

1 typedef struct aeApiState {
2     // epoll事件的文件描述符
3     int epfd;
4     // 事件表
5     struct epoll_event *events;
6 } aeApiState;

2.3.4.2 用户层到底层事件类型的转换

　　epoll模型的struct epoll_event的结构中定义这自己的事件类型，例如EPOLLIN POLLOUT等等，但是Redis的文件事件结构aeFileEvent中也在mask中定义了自己的事件类型，例如：AE_READABLE AE_WRITABLE等，于是，就需要实现一个中间层将两者的事件类型相联系起来，这也就是之前提到的ae_epoll.c文件中实现的相同的API，我们列出来：

 1 // 创建一个epoll实例，保存到eventLoop中
 2 static int aeApiCreate(aeEventLoop *eventLoop)
 3 // 调整事件表的大小
 4 static int aeApiResize(aeEventLoop *eventLoop, int setsize)  
 5 // 释放epoll实例和事件表空间
 6 static void aeApiFree(aeEventLoop *eventLoop)
 7 // 在epfd标识的事件表上注册fd的事件
 8 static int aeApiAddEvent(aeEventLoop *eventLoop, int fd, int mask)
 9 // 在epfd标识的事件表上注删除fd的事件
10 static void aeApiDelEvent(aeEventLoop *eventLoop, int fd, int delmask)
11 // 等待所监听文件描述符上有事件发生
12 static int aeApiPoll(aeEventLoop *eventLoop, struct timeval *tvp)
13 // 返回正在使用的IO多路复用库的名字
14 static char *aeApiName(void)

　　这些API都是调用相应的底层多路复用库来将Redis事件状态结构aeEventLoop所关联，就是将epoll的底层函数封装起来，Redis实现事件时，只需调用这些接口即可。我们查看两个重要的函数的源码，看看是如何实现的

2.3.4.2.1 aeApiAddEvent函数

向Redis事件状态结构aeEventLoop的事件表event注册一个事件，对应的是epoll_ctl函数

 1 // 在epfd标识的事件表上注册fd的事件
 2 static int aeApiAddEvent(aeEventLoop *eventLoop, int fd, int mask) {
 3     aeApiState *state = eventLoop->apidata;
 4     struct epoll_event ee = {0}; /* avoid valgrind warning */
 5     /**
 6      * If the fd was already monitored for some event, we need a MOD
 7      * operation. Otherwise we need an ADD operation.
 8      * EPOLL_CTL_ADD，向epfd注册fd的上的event
 9      * EPOLL_CTL_MOD，修改fd已注册的event
10      * #define AE_NONE 0           //未设置
11      * #define AE_READABLE 1       //事件可读
12      * #define AE_WRITABLE 2       //事件可写
13      * 判断fd事件的操作，如果没有设置事件，则进行关联mask类型事件，否则进行修改
14      */
15     int op = eventLoop->events[fd].mask == AE_NONE ?
16             EPOLL_CTL_ADD : EPOLL_CTL_MOD;
17 
18 
19     // struct epoll_event {
20     //      uint32_t     events;      /* Epoll events */
21     //      epoll_data_t data;        /* User data variable */
22     // };
23     ee.events = 0;
24     // 如果是修改事件，合并之前的事件类型
25     mask |= eventLoop->events[fd].mask; /* Merge old events */
26     // 根据mask映射epoll的事件类型
27     if (mask & AE_READABLE) ee.events |= EPOLLIN;//读事件
28     if (mask & AE_WRITABLE) ee.events |= EPOLLOUT;//写事件
29     //设置事件所从属的目标文件描述符
30     ee.data.fd = fd;
31     // 将ee事件注册到epoll中
32     if (epoll_ctl(state->epfd,op,fd,&ee) == -1) return -1;
33     return 0;
34 }

2.3.4.2.1 aeApiPoll函数

等待所监听文件描述符上有事件发生，对应着底层epoll_wait函数

 1 // 等待所监听文件描述符上有事件发生
 2 static int aeApiPoll(aeEventLoop *eventLoop, struct timeval *tvp) {
 3     aeApiState *state = eventLoop->apidata;
 4     int retval, numevents = 0;
 5 
 6     // 监听事件表上是否有事件发生
 7     retval = epoll_wait(state->epfd,state->events,eventLoop->setsize,
 8             tvp ? (tvp->tv_sec*1000 + tvp->tv_usec/1000) : -1);
 9     // 至少有一个就绪的事件
10     if (retval > 0) {
11         int j;
12 
13         numevents = retval;
14         // 遍历就绪的事件表，将其加入到eventLoop的就绪事件表中
15         for (j = 0; j < numevents; j++) {
16             int mask = 0;
17             struct epoll_event *e = state->events+j;
18 
19             // 根据就绪的事件类型，设置mask
20             if (e->events & EPOLLIN) mask |= AE_READABLE;
21             if (e->events & EPOLLOUT) mask |= AE_WRITABLE;
22             if (e->events & EPOLLERR) mask |= AE_WRITABLE;
23             if (e->events & EPOLLHUP) mask |= AE_WRITABLE;
24             // 添加到就绪事件表中
25             eventLoop->fired[j].fd = e->data.fd;
26             eventLoop->fired[j].mask = mask;
27         }
28     }
29     // 返回就绪的事件个数
30     return numevents;
31 }

三、使用实例

3.1 底层多路IO模型实现

　　这里写了一个由标准输入的读事件驱动的echo服务例子，同时用一个5秒的循环定时器每个5秒打印一次服务器状态。这里用了epoll为底层事件接口。具体的代码抽取可以

 1 /**
 2  * Include the best multiplexing layer supported by this system.
 3  * The following should be ordered by performances, descending.
 4  * 包括该系统支持的最佳复用层。性能依次下降
 5  */
 6 #ifdef HAVE_EVPORT
 7 #include "ae_evport.c"
 8 #else
 9     #ifdef HAVE_EPOLL
10     #include "ae_epoll.c"
11     #else
12         #ifdef HAVE_KQUEUE
13         #include "ae_kqueue.c"
14         #else
15         #include "ae_select.c"
16         #endif
17     #endif
18 #endif

3.2 使用实例

　　这里主要是分析Redis的事件模型的封装，因此对于其对socket的包装以及内存管理都不做分析。故采用标准输入，同时需要将这些文件中的内存管理接口"zmalloc()"以及"zfree()"替换成C库中的“malloc()”还有"free()"。可以使用sed或者vim的%s做替换操作。代码如下：

 1 #include "ae.h"
 2  
 3 #include <stdio.h>
 4 #include <assert.h>
 5 #include <unistd.h>
 6 #include <sys/time.h>
 7  
 8  
 9 #define MAXFD 5
10  
11  
12 void loop_init(struct aeEventLoop *l) 
13 {
14         puts("I'm loop_init!!! 
");
15 }
16  
17 void file_cb(struct aeEventLoop *l,int fd,void *data,int mask)
18 {
19         char buf[51] ={0};
20         read(fd,buf,51);
21         printf("I'm file_cb ,here [EventLoop: %p],[fd : %d],[data: %p],[mask: %d] 
",l,fd,data,mask);
22         printf("get %s",buf);
23 }
24  
25 int time_cb(struct aeEventLoop *l,long long id,void *data)
26 {
27         printf("now is %ld
",time(NULL));
28         printf("I'm time_cb,here [EventLoop: %p],[id : %lld],[data: %p] 
",l,id,data);
29         return 5*1000;
30  
31 }
32  
33 void fin_cb(struct aeEventLoop *l,void *data)
34 {
35         puts("call the unknow final function 
");
36 }
37  
38 int main(int argc,char *argv[])
39 {
40         aeEventLoop *l; 
41         char *msg = "Here std say:";
42         char *user_data = malloc(50*sizeof(char));
43         if(! user_data)
44                 assert( ("user_data malloc error",user_data) );
45         memset(user_data,'',50);
46         memcpy(user_data,msg,sizeof(msg));
47  
48         l = aeCreateEventLoop(MAXFD);
49         aeSetBeforeSleepProc(l,loop_init);
50         int res;
51         res = aeCreateFileEvent(l,STDIN_FILENO,AE_READABLE,file_cb,user_data);
52         printf("create file event is ok? [%d]
",res);
53         res = aeCreateTimeEvent(l,5*1000,time_cb,NULL,fin_cb);
54         printf("create time event is ok? [%d]
",!res);
55  
56         aeMain(l);
57  
58         puts("Everything is ok !!!
");
59         return 0;
60 }

View Code

大致逻辑就是注册一个标准输入的读事件，和一个定时器事件。

3.3 回调函数类型

IO读写事件回调函数
定时器事件回调函数
删除定时事件回调函数
进入事件循环等待之前的回调函数

这里要说明的就是在ae.h中定义了读、写、定时器等回调函数的类型，源码如下

1 /* Types and data structures */
2 //IO读写事件回调函数
3 typedef void aeFileProc(struct aeEventLoop *eventLoop, int fd, void *clientData, int mask);
4 //定时器事件回调函数
5 typedef int aeTimeProc(struct aeEventLoop *eventLoop, long long id, void *clientData);
6 //删除定时事件的回调函数
7 typedef void aeEventFinalizerProc(struct aeEventLoop *eventLoop, void *clientData);
8 //进入循环等待之前的回调函数
9 typedef void aeBeforeSleepProc(struct aeEventLoop *eventLoop);

四、事件的源码实现

4.1 整体框架

Redis事件的源码全部定义在ae.c文件中，整体框架如下：

4.2 事件模型的创建与释放

4.2.1 事件模型eEventLoop创建的API函数

Redis 通过以下接口进行 eventloop 的创建和释放。

1 eEventLoop *aeCreateEventLoop(int setsize);
2 void aeDeleteEventLoop(aeEventLoop *eventLoop);

4.2.2 事件模型eEventLoop的创建过程

eventloop 的创建，源码如下：

 1 /**
 2  *  eventloop 的创建
 3  */
 4 aeEventLoop *aeCreateEventLoop(int setsize) {
 5     aeEventLoop *eventLoop;
 6     int i;
 7 
 8     if ((eventLoop = zmalloc(sizeof(*eventLoop))) == NULL) goto err;
 9     //直接创建两个未初始化的数组
10     eventLoop->events = zmalloc(sizeof(aeFileEvent)*setsize);
11     eventLoop->fired = zmalloc(sizeof(aeFiredEvent)*setsize);
12     if (eventLoop->events == NULL || eventLoop->fired == NULL) goto err;
13     /**
14      * setsize:server.maxclients+CONFIG_FDSET_INCR
15      *
16      * 1. maxclients代表用户配置的最大连接数，可在启动时由--maxclients指定，默认为10000。
17      * 2. CONFIG_FDSET_INCR 大小为128。给Redis预留一些安全空间。
18      */
19     eventLoop->setsize = setsize;
20     eventLoop->lastTime = time(NULL);
21     //定时事件
22     eventLoop->timeEventHead = NULL;
23     eventLoop->timeEventNextId = 0;
24     eventLoop->stop = 0;
25     eventLoop->maxfd = -1;
26     eventLoop->beforesleep = NULL;
27     eventLoop->aftersleep = NULL;
28     if (aeApiCreate(eventLoop) == -1) goto err;
29     /**
30      * Events with mask == AE_NONE are not set. So let's initialize the
31      * vector with it.
32      * 可以看到数组长度就是setsize，同时创建之后将每一个event的mask属性置为AE_NONE(即是0)，
33      * mask代表该fd注册了哪些事件。
34      *
35      * 对于eventLoop->events数组来说，fd就是这个数组的下标。 例如，当程序刚刚启动时候，创建监听套接字，
36      * 按照标准规定，该fd的值为3。此时就直接在eventLoop->events下标为3的元素中存放相应event数据。
37      *
38      * 不过也基于文件描述符的这些特点，意味着events数组的前三位一定不会有相应的fd赋值。
39      */
40     for (i = 0; i < setsize; i++)
41         eventLoop->events[i].mask = AE_NONE;
42     return eventLoop;
43 
44 err:
45     if (eventLoop) {
46         zfree(eventLoop->events);
47         zfree(eventLoop->fired);
48         zfree(eventLoop);
49     }
50     return NULL;
51 }

　　Redis 通过将对应事件注册到 eventloop 中，然后不断循环检测有无事件触发。目前 eventloop 支持超时事件和网络 IO 读写事件的注册。我们可以通过 aeCreateEventLoop 来创建一个 eventloop，实际上在redis服务器初始化的时候就会调用server.el = aeCreateEventLoop(server.maxclients+CONFIG_FDSET_INCR);创建一个eventloop，作为后续整个服务器运行期间的事件模型。可以看到在创建 EventLoop 的时候，必须指定一个 setsize 的参数。setsize 参数表示了 eventloop 可以监听的网络事件 fd 的个数（不包含超时事件），如果当前监听的 fd 个数超过了 setsize，eventloop 将不能继续注册，setsize实际上就是server.maxclients+CONFIG_FDSET_INCR，其中CONFIG_FDSET_INCR为128，定义为：

 1 /**
 2  * When configuring the server eventloop, we setup it so that the total number
 3  * of file descriptors we can handle are server.maxclients + RESERVED_FDS +
 4  * a few more to stay safe. Since RESERVED_FDS defaults to 32, we add 96
 5  * in order to make sure of not over provisioning more than 128 fds.
 6  * 在配置服务器事件循环时，我们设置它以便我们可以处理的文件描述符总数是
 7  * server.maxclients + RESERVED_FDS + 一些以确保安全。 由于 RESERVED_FDS 默认为 32，
 8  * 我们添加 96 以确保不会过度配置超过 128 个 fd。
 9  */
10 #define CONFIG_FDSET_INCR (CONFIG_MIN_RESERVED_FDS+96)

　　其中32是为了服务器正常运行预留的一些特殊文件描述符，服务器利用它们完成一些特殊操作，96是为了预留些安全区间，因为后续版本的更新中有可能需要更多的文件描述符。

4.2.3 事件模型eEventLoop如何存储监听事件

我们知道，Linux 内核会给每个进程维护一个文件描述符表。而 POSIX 标准对于文件描述符进行了以下约束：

fd 为 0、1、2 分别表示标准输入、标准输出和错误输出
每次新打开的 fd，必须使用当前进程中最小可用的文件描述符。

Redis 充分利用了文件描述符的这些特点，来存储每个 fd 对应的事件。

在 Redis 的 eventloop 中，直接用了一个连续数组来存储事件信息:

1 eventLoop->events = zmalloc(sizeof(aeFileEvent)*setsize);
2 for (i = 0; i < setsize; i++)
3     eventLoop->events[i].mask = AE_NONE;

　　可以看到数组长度就是 setsize，同时创建之后将每一个 event 的 mask 属性置为 AE_NONE(即是 0)，mask 代表该 fd 注册了哪些事件。对于eventLoop->events数组来说，fd 就是这个数组的下标。

　　例如，当程序刚刚启动时候，创建监听套接字，按照标准规定，该 fd 的值为 3。此时就直接在 eventLoop->events 下标为 3 的元素中存放相应 event 数据。不过也基于文件描述符的这些特点，意味着 events 数组的前三位一定不会有相应的 fd 赋值。

　　也正是因为 Redis 利用了 fd 的这个特点，Redis 只能在完全符合 POSIX 标准的系统中工作。其他的例如 Windows 系统，生成的 fd 或者说 HANDLE 更像是个指针，并不符合 POSIX 标准。

4.3 网络 IO 事件的创建与释放

4.3.1 创建和释放的API

Redis 通过以下接口进行网络 IO 事件的注册和删除。

1 int aeCreateFileEvent(aeEventLoop *eventLoop, int fd, int mask,
2         aeFileProc *proc, void *clientData);
3 
4 void aeDeleteFileEvent(aeEventLoop *eventLoop, int fd, int mask);

aeCreateFileEvent 表示将某个 fd 的某些事件注册到 eventloop 中。

4.3.2 事件类型

目前可注册的事件有三种:

AE_READABLE 可读事件
AE_WRITABLE 可写事件
AE_BARRIER 该事件可以实现读写事件处理顺序的反转。

而 mask 就是这几个事件经过或运算后的掩码。

4.3.3 事件的添加

　　aeCreateFileEvent 在 epoll 的实现中调用了 epoll_ctl 函数。Redis 会根据该事件对应之前的 mask 是否为 AE_NONE，来决定使用 EPOLL_CTL_ADD 还是 EPOLL_CTL_MOD。aeCreateFileEvent 源码如下：

 1 /**
 2  * aeCreateFileEvent函数接受一个套接字描述符、一个事件类型，以及一个事件处理器作为参数，
 3  * 将给定套接字的给定事件加入到I/O多路复用程序的监听范围之内，并对事件和事件处理器进行关联;
 4  *
 5  * mask：指定注册的事件类型，可以是读或写。
 6  * proc：事件处理函数。
 7  */
 8 int aeCreateFileEvent(aeEventLoop *eventLoop, int fd, int mask,
 9         aeFileProc *proc, void *clientData)
10 {
11     //允许添加的事件的最大数
12     if (fd >= eventLoop->setsize) {
13         errno = ERANGE;
14         return AE_ERR;
15     }
16     // 获取文件描述符的事件结构体
17     aeFileEvent *fe = &eventLoop->events[fd];
18 
19     // 添加事件，返回-1，添加失败
20     if (aeApiAddEvent(eventLoop, fd, mask) == -1)
21         return AE_ERR;
22     // 设置事件掩码
23     fe->mask |= mask;
24     // 设置事件的回调函数
25     if (mask & AE_READABLE) fe->rfileProc = proc;
26     if (mask & AE_WRITABLE) fe->wfileProc = proc;
27     // 设置客户端的相关信息
28     fe->clientData = clientData;
29     // 更新当前监控的最大文件描述符下标
30     if (fd > eventLoop->maxfd)
31         eventLoop->maxfd = fd;
32     // 返回添加成功
33     return AE_OK;
34 }

aeApiAddEvent函数真正将事件加入监听，但其具体实现与选择的底层IO模型相关，如果选择的是epoll模型，函数源码如下：

 1 // 在epfd标识的事件表上注册fd的事件
 2 static int aeApiAddEvent(aeEventLoop *eventLoop, int fd, int mask) {
 3     aeApiState *state = eventLoop->apidata;
 4     struct epoll_event ee = {0}; /* avoid valgrind warning */
 5     /**
 6      * If the fd was already monitored for some event, we need a MOD
 7      * operation. Otherwise we need an ADD operation.
 8      * EPOLL_CTL_ADD，向epfd注册fd的上的event
 9      * EPOLL_CTL_MOD，修改fd已注册的event
10      * #define AE_NONE 0           //未设置
11      * #define AE_READABLE 1       //事件可读
12      * #define AE_WRITABLE 2       //事件可写
13      * 判断fd事件的操作，如果没有设置事件，则进行关联mask类型事件，否则进行修改
14      */
15     int op = eventLoop->events[fd].mask == AE_NONE ?
16             EPOLL_CTL_ADD : EPOLL_CTL_MOD;
17 
18 
19     // struct epoll_event {
20     //      uint32_t     events;      /* Epoll events */
21     //      epoll_data_t data;        /* User data variable */
22     // };
23     ee.events = 0;
24     // 如果是修改事件，合并之前的事件类型
25     mask |= eventLoop->events[fd].mask; /* Merge old events */
26     // 根据mask映射epoll的事件类型
27     if (mask & AE_READABLE) ee.events |= EPOLLIN;//读事件
28     if (mask & AE_WRITABLE) ee.events |= EPOLLOUT;//写事件
29     //设置事件所从属的目标文件描述符
30     ee.data.fd = fd;
31     // 将ee事件注册到epoll中
32     if (epoll_ctl(state->epfd,op,fd,&ee) == -1) return -1;
33     return 0;
34 }

4.3.4 事件的删除

同样的，aeDeleteFileEvent 也使用了 epoll_ctl，Redis 判断用户是否是要完全删除该 fd 上所有事件，来决定使用 EPOLL_CTL_DEL 还是 EPOLL_CTL_MOD。

 1 void aeDeleteFileEvent(aeEventLoop *eventLoop, int fd, int mask)
 2 {
 3     //允许添加的事件的最大数
 4     if (fd >= eventLoop->setsize)
 5         return;
 6     // 获取文件描述符的事件结构体
 7     aeFileEvent *fe = &eventLoop->events[fd];
 8     // 如果该文件描述符没有事件被设置
 9     if (fe->mask == AE_NONE)
10         return;
11 
12     /**
13      * We want to always remove AE_BARRIER if set when AE_WRITABLE
14      * is removed.
15      * 如果哦我们准备移除AE_WRITABLE写事件，我们就应该移除AE_BARRIER
16      */
17     if (mask & AE_WRITABLE) mask |= AE_BARRIER;
18 
19     //实现事件的移除，和底层的网络IO的选择相关，如果是epoll模型，则调用的就是aeApiDelEvent函数
20     aeApiDelEvent(eventLoop, fd, mask);
21 
22     fe->mask = fe->mask & (~mask);
23     if (fd == eventLoop->maxfd && fe->mask == AE_NONE) {
24         /* Update the max fd 更新当前监听的最大文件描述符ID*/
25         int j;
26 
27         for (j = eventLoop->maxfd-1; j >= 0; j--)
28             if (eventLoop->events[j].mask != AE_NONE) break;
29         eventLoop->maxfd = j;
30     }
31 }

aeApiDelEvent函数真正将事件加入监听，但其具体实现与选择的底层IO模型相关，如果选择的是epoll模型，函数源码如下：

 1 // 在epfd标识的事件表上删除fd的事件
 2 static void aeApiDelEvent(aeEventLoop *eventLoop, int fd, int delmask) {
 3     aeApiState *state = eventLoop->apidata;
 4     struct epoll_event ee = {0}; /* avoid valgrind warning */
 5     int mask = eventLoop->events[fd].mask & (~delmask);
 6 
 7     ee.events = 0;
 8     if (mask & AE_READABLE) ee.events |= EPOLLIN;
 9     if (mask & AE_WRITABLE) ee.events |= EPOLLOUT;
10     ee.data.fd = fd;
11     if (mask != AE_NONE) {
12         epoll_ctl(state->epfd,EPOLL_CTL_MOD,fd,&ee);
13     } else {
14         /* Note, Kernel < 2.6.9 requires a non null event pointer even for
15          * EPOLL_CTL_DEL. */
16         epoll_ctl(state->epfd,EPOLL_CTL_DEL,fd,&ee);
17     }
18 }

4.4 定时事件

4.4.1 创建和销毁的API函数

Redis 通过以下两个接口进行定时器的注册和取消。

1 long long aeCreateTimeEvent(aeEventLoop *eventLoop, long long milliseconds,
2         aeTimeProc *proc, void *clientData,
3         aeEventFinalizerProc *finalizerProc);
4 int aeDeleteTimeEvent(aeEventLoop *eventLoop, long long id);

4.4.2 定时事件的创建

aeCreateTimeEvent 源码如下：

 1 /**
 2  * 创建定时事件
 3  *
 4  * Proc：事件处理函数。
 5  * finalizerProc：清理函数，在删除定时器时调用。
 6  * clientData：需要传入事件处理函数的参数。
 7  */
 8 long long aeCreateTimeEvent(aeEventLoop *eventLoop, long long milliseconds,
 9         aeTimeProc *proc, void *clientData,
10         aeEventFinalizerProc *finalizerProc)
11 {
12     long long id = eventLoop->timeEventNextId++;
13     aeTimeEvent *te;
14 
15     te = zmalloc(sizeof(*te));
16     if (te == NULL) return AE_ERR;
17     te->id = id;
18     aeAddMillisecondsToNow(milliseconds,&te->when_sec,&te->when_ms);
19     te->timeProc = proc;
20     te->finalizerProc = finalizerProc;
21     te->clientData = clientData;
22     te->prev = NULL;
23     te->next = eventLoop->timeEventHead;
24     if (te->next)
25         te->next->prev = te;
26     eventLoop->timeEventHead = te;
27     return id;
28 }

View Code

4.4.3 回调函数设置

在调用 aeCreateTimeEvent 注册超时事件的时候，调用方需要提供两个 callback: aeTimeProc 和 aeEventFinalizerProc。

aeTimeProc 很简单，就是当超时事件触发时调用的 callback。有点特殊的是，aeTimeProc 需要返回一个 int 值，代表下次该超时事件触发的时间间隔。如果返回 - 1，则说明超时时间不需要再触发了，标记为删除即可。
finalizerProc 当 timer 被删除的时候，会调用这个 callback

4.4.4 管理定时事件的底层结构

　　Redis 的定时器其实做的非常简陋，只是一个普通的双向链表，链表也并不是有序的。每次最新的超时事件，直接插入链表的最头部。当 AE 要遍历当前时刻的超时事件时，也是直接暴力的从头到尾遍历链表，看看有没有超时的事件。一般来说，定时器都会采用最小堆或者时间轮等有序数据结构进行存储，为什么 Redis 的定时器做的这么简陋？Redis 的设计与实现》一书中说，在 Redis 3.0 版本中，只使用到了 serverCon 这一个超时事件。所以这种情况下，也无所谓性能了，虽然是个链表，但其实用起来就只有一个元素，相当于当做一个指针在用。

　　Redis 在注释里面也说明了这事，并且给出了以后的优化方案：用 skiplist 代替现有普通链表，查询的时间复杂度将优化为 O(1), 插入的时间复杂度将变成 O(log(N))。

4.4.5 定时事件的异常情况

定时事件的异常情况：

　　虽然定时器做的这么简陋，但是对于一些时间上的异常情况，Redis 还是做了下基本的处理。具体可见如下代码：

1 if (now < eventLoop->lastTime) {
2         te = eventLoop->timeEventHead;
3         while(te) {
4                 te->when_sec = 0;
5                 te = te->next;
6         }
7 }

　　这段代码的意思是，如果当前时刻小于 lastTime, 那意味着时间有可能被调整了。对于这种情况，Redis 是怎么处理的呢？直接把所有的事件的超时时间都置为 0，te->when_sec = 0。这样的话，接下来检查有哪些超时时间到期的时候，所有的超时事件都会被判定为到期。相当于本次遍历把所有超时事件一次性全部激活。因为 Redis 认为，在这种异常情况下，与其冒着超时事件可能永远无法触发的风险，还不如把事情提前做了。还是基于 Redis 够用就行的原则，这个解决方案在 Redis 中显然是被接受的。

　　但是其实还有更好的做法，比如 libevent 就是通过相对时间的方式进行处理这个问题。为了解决这个几乎不会出现的异常 case，libevent 也花了大量代码进行处理。

4.5 事件的循环

4.5.1 事件循环主函数

主函数aeMain，实现事件模型的底层循环等待

 1 // 事件轮询的主函数
 2 void aeMain(aeEventLoop *eventLoop) {
 3     eventLoop->stop = 0;
 4     // 一直处理事件
 5     while (!eventLoop->stop) {
 6         // 执行处理事件之前的函数，实际上就是server.c中的void beforeSleep(struct aeEventLoop *eventLoop)函数
 7         if (eventLoop->beforesleep != NULL)
 8             eventLoop->beforesleep(eventLoop);
 9         //处理到时的时间事件和就绪的文件事件
10         aeProcessEvents(eventLoop, AE_ALL_EVENTS|AE_CALL_AFTER_SLEEP);
11     }
12 }

4.5.2 事件处理

4.5.2.1 事件处理函数API

　　这个事件的主函数aeMain很清楚的可以看到，如果服务器一直处理事件，那么就是一个死循环，而一个最典型的事件驱动，就是一个死循环。调用处理事件的函数aeProcessEvents，他们参数是一个事件状态结构aeEventLoop和AE_ALL_EVENTS|AE_CALL_AFTER_SLEEP，源码如下：

  1 /* Process every pending time event, then every pending file event
  2  * (that may be registered by time event callbacks just processed).
  3  * Without special flags the function sleeps until some file event
  4  * fires, or when the next time event occurs (if any).
  5  *
  6  * If flags is 0, the function does nothing and returns.
  7  * if flags has AE_ALL_EVENTS set, all the kind of events are processed.
  8  * if flags has AE_FILE_EVENTS set, file events are processed.
  9  * if flags has AE_TIME_EVENTS set, time events are processed.
 10  * if flags has AE_DONT_WAIT set the function returns ASAP until all
 11  * if flags has AE_CALL_AFTER_SLEEP set, the aftersleep callback is called.
 12  * the events that's possible to process without to wait are processed.
 13  *
 14  * The function returns the number of events processed.
 15  *
 16  * 处理到时的时间事件和就绪的文件事件
 17  * 函数返回执行的事件个数
 18  *
 19  * 完成的工作包括：
 20  *      取出最近的一次超时事件。
 21  *      计算该超时事件还有多久才可以触发。
 22  *      等待网络事件触发或者超时。
 23  *      处理触发的各个事件，包括网络事件和超时事件
 24  */
 25 int aeProcessEvents(aeEventLoop *eventLoop, int flags)
 26 {
 27     int processed = 0, numevents;
 28 
 29     /* Nothing to do? return ASAP */
 30     // 如果什么事件都没有设置则直接返回
 31     if (!(flags & AE_TIME_EVENTS) && !(flags & AE_FILE_EVENTS)) return 0;
 32 
 33     /**
 34      * Note that we want call select() even if there are no
 35      * file events to process as long as we want to process time
 36      * events, in order to sleep until the next time event is ready
 37      * to fire.
 38      *
 39      * 请注意，既然我们要处理时间事件，即使没有要处理的文件事件，我们仍要调用select（），
 40      * 以便在下一次事件准备启动之前进行休眠
 41      */
 42     // 当前还没有要处理的文件事件，或者设置了时间时间但是没有设置不阻塞标识
 43     if (eventLoop->maxfd != -1 ||
 44         ((flags & AE_TIME_EVENTS) && !(flags & AE_DONT_WAIT))) {
 45         int j;
 46         aeTimeEvent *shortest = NULL;
 47         struct timeval tv, *tvp;
 48 
 49         // 如果设置了时间事件而没有设置不阻塞标识
 50         if (flags & AE_TIME_EVENTS && !(flags & AE_DONT_WAIT))
 51             // 获取最近到时的时间事件
 52             shortest = aeSearchNearestTimer(eventLoop);
 53         // 获取到了最早到时的时间事件
 54         if (shortest) {
 55             long now_sec, now_ms;
 56 
 57             // 获取当前时间
 58             aeGetTime(&now_sec, &now_ms);
 59             tvp = &tv;
 60 
 61             /**
 62              * How many milliseconds we need to wait for the next
 63              * time event to fire?
 64              * 等待该时间事件到时所需要的时长
 65              */
 66             long long ms =
 67                 (shortest->when_sec - now_sec)*1000 +
 68                 shortest->when_ms - now_ms;
 69 
 70             // 如果没到时
 71             if (ms > 0) {
 72                 // 保存时长到tvp中
 73                 tvp->tv_sec = ms/1000;
 74                 tvp->tv_usec = (ms % 1000)*1000;
 75             } else {
 76                 // 如果已经到时，则将tvp的时间设置为0
 77                 tvp->tv_sec = 0;
 78                 tvp->tv_usec = 0;
 79             }
 80         } else {
 81             // 没有获取到了最早到时的时间事件，时间事件链表为空
 82             /* If we have to check for events but need to return
 83              * ASAP because of AE_DONT_WAIT we need to set the timeout
 84              * to zero */
 85             // 如果设置了不阻塞标识
 86             if (flags & AE_DONT_WAIT) {
 87                 // 将tvp的时间设置为0，就不会阻塞
 88                 tv.tv_sec = tv.tv_usec = 0;
 89                 tvp = &tv;
 90             } else {
 91                 // 阻塞到第一个时间事件的到来
 92                 /* Otherwise we can block */
 93                 tvp = NULL; /* wait forever */
 94             }
 95         }
 96 
 97         /**
 98          * Call the multiplexing API, will return only on timeout or when
 99          * some event fires.
100          * 等待所监听文件描述符上有事件发生
101          * 如果tvp为NULL，则阻塞在此，否则等待tvp设置阻塞的时间，就会有时间事件到时
102          * 返回了就绪文件事件的个数
103          */
104         numevents = aeApiPoll(eventLoop, tvp);
105 
106         /* After sleep callback. 调用主循环之后的回调函数*/
107         if (eventLoop->aftersleep != NULL && flags & AE_CALL_AFTER_SLEEP)
108             eventLoop->aftersleep(eventLoop);
109 
110         //处理触发的各个事件，包括网络事件和超时事件
111         for (j = 0; j < numevents; j++) {
112             // 获取就绪文件事件的地址
113             aeFileEvent *fe = &eventLoop->events[eventLoop->fired[j].fd];
114             // 获取就绪文件事件的类型，文件描述符
115             int mask = eventLoop->fired[j].mask;
116             int fd = eventLoop->fired[j].fd;
117             int fired = 0; /* Number of events fired for current fd. */
118 
119             /* Normally we execute the readable event first, and the writable
120              * event laster. This is useful as sometimes we may be able
121              * to serve the reply of a query immediately after processing the
122              * query.
123              *
124              * However if AE_BARRIER is set in the mask, our application is
125              * asking us to do the reverse: never fire the writable event
126              * after the readable. In such a case, we invert the calls.
127              * This is useful when, for instance, we want to do things
128              * in the beforeSleep() hook, like fsynching a file to disk,
129              * before replying to a client. */
130             int invert = fe->mask & AE_BARRIER;
131 
132             /**
133              * Note the "fe->mask & mask & ..." code: maybe an already
134              * processed event removed an element that fired and we still
135              * didn't processed, so we check if the event is still valid.
136              * 注意“fe->mask & mask & ...”代码：可能是一个已经处理的事件删除了一个被触发的元素，
137              * 我们仍然没有处理，所以我们检查事件是否仍然有效。
138              *
139              * Fire the readable event if the call sequence is not
140              * inverted.
141              * 如果调用顺序未反转，则触发 readable 事件。
142              */
143             if (!invert && fe->mask & mask & AE_READABLE) {
144                 //调用读事件方法处理读事件
145                 fe->rfileProc(eventLoop,fd,fe->clientData,mask);
146                 //当前文件描述符触发的事件数
147                 fired++;
148             }
149 
150             /* Fire the writable event. 文件可写事件发生*/
151             if (fe->mask & mask & AE_WRITABLE) {
152                 // 读写事件的执行发法不同，则执行写事件，避免重复执行相同的方法
153                 if (!fired || fe->wfileProc != fe->rfileProc) {
154                     //调用写事件方法处理读事件
155                     fe->wfileProc(eventLoop,fd,fe->clientData,mask);
156                     //当前文件描述符触发的事件数
157                     fired++;
158                 }
159             }
160 
161             /**
162              * If we have to invert the call, fire the readable event now
163              * after the writable one.
164              * 如果我们反转了读写事件的处理顺序，现在在写事件处理结束后触发读事件
165              */
166             if (invert && fe->mask & mask & AE_READABLE) {
167                 if (!fired || fe->wfileProc != fe->rfileProc) {
168                     fe->rfileProc(eventLoop,fd,fe->clientData,mask);
169                     fired++;
170                 }
171             }
172             //执行的事件次数加1
173             processed++;
174         }
175     }
176     /**
177      * Check time events
178      * 如果存在定时事件，处理时间事件
179      */
180     if (flags & AE_TIME_EVENTS)
181         processed += processTimeEvents(eventLoop);
182 
183     return processed; /* return the number of processed file/time events */
184 }

View Code

4.5.2.2 事件处理函数API使用的参数

刚才提到该函数的一个参数是AE_ALL_EVENTS，他的定义在ae.h中，定义如下：

 1 //文件事件
 2 #define AE_FILE_EVENTS 1
 3 //时间事件
 4 #define AE_TIME_EVENTS 2
 5 //文件和时间事件
 6 #define AE_ALL_EVENTS (AE_FILE_EVENTS|AE_TIME_EVENTS)
 7 //不阻塞等待标识
 8 #define AE_DONT_WAIT 4
 9 //循环等待之后的事件，相当于aeBeforeSleepProc（也就是server.c中的beforeSleep）
10 #define AE_CALL_AFTER_SLEEP 8

很明显，flags是AE_FILE_EVENTS和AE_TIME_EVENTS或的结果，他们的含义如下：

如果flags = 0，函数什么都不做，直接返回
如果flags设置了 AE_ALL_EVENTS ，则执行所有类型的事件
如果flags设置了 AE_FILE_EVENTS ，则执行文件事件
如果flags设置了 AE_TIME_EVENTS ，则执行时间事件
如果flags设置了 AE_DONT_WAIT ，那么函数处理完事件后直接返回，不阻塞等待
如果flags设置了 AE_CALL_AFTER_SLEEP，那么函数在aeApiPoll函数返回之后就执行eventLoop->aftersleep(eventLoop);

4.5.3 事件循环的流程

4.5.3.1 定时事件

　　Redis服务器在没有被事件触发时，就会阻塞等待，因为没有设置AE_DONT_WAIT标识。但是他不会一直的死等待，等待文件事件的到来，因为他还要处理时间事件，因此，在调用aeApiPoll进行监听之前，先从时间事件表中获取一个最近到达的时间时间，根据要等待的时间构建一个struct timeval tv, *tvp结构的变量，这个变量保存着服务器阻塞等待文件事件的最长时间，一旦时间到达而没有触发文件事件，aeApiPoll函数就会停止阻塞，进而调用processTimeEvents处理时间事件，因为Redis服务器设定一个对自身资源和状态进行周期性检查的定时时间事件，而该函数就是timeProc所指向的回调函数serverCron，该定时事件在服务器初始化的时候就会设置，创建位置在server.c的void initServer(void)函数中，源代码如下：

 1 /**
 2      * Create the timer callback, this is our way to process many background
 3      * operations incrementally, like clients timeout, eviction of unaccessed
 4      * expired keys and so forth.
 5      * 创建计时器回调，这是我们逐步处理许多后台操作的方式，例如客户端超时、驱逐未访问的过期密钥等。
 6      * 回调函数是serverCron
 7      */
 8     if (aeCreateTimeEvent(server.el, 1, serverCron, NULL, NULL) == AE_ERR) {
 9         serverPanic("Can't create event loop timers.");
10         exit(1);
11     }

　　回调函数serverCron位于server.c中，原型为：

1 int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData)

　　如果在阻塞等待的最长时间之间，触发了文件事件，就会先执行文件事件，后执行时间事件，因此处理时间事件通常比预设的会晚一点。

4.5.3.2 文件事件回调函数类型

　　而执行文件事件rfileProc和wfileProc也是调用了回调函数，Redis将文件事件的处理分为了好几种，用于处理不同的网络通信需求，下面列出回调函数的原型：

网络客户端连接请求
本地客户端连接请求
（客户端）读事件请求
（服务端）写事件请求

1 // 用于accept client的connect。
2 void acceptTcpHandler(aeEventLoop *el, int fd, void *privdata, int mask);
3 // 用于acceptclient的本地connect。
4 void acceptUnixHandler(aeEventLoop *el, int fd, void *privdata, int mask);
5 // 用于读入client发送的请求。
6 void readQueryFromClient(aeEventLoop *el, int fd, void *privdata, int mask);
7 // 用于向client发送命令回复。
8 void addReplyString(client *c, const char *s, size_t len);

4.5.3.3 获取下一个超时事件

接下来，我们查看获取最快达到的时间事件的函数aeSearchNearestTimer实现

 1 /**
 2  * Search the first timer to fire.
 3  * This operation is useful to know how many time the select can be
 4  * put in sleep without to delay any event.
 5  * If there are no timers NULL is returned.
 6  *
 7  * Note that's O(N) since time events are unsorted.
 8  * Possible optimizations (not needed by Redis so far, but...):
 9  * 1) Insert the event in order, so that the nearest is just the head.
10  *    Much better but still insertion or deletion of timers is O(N).
11  * 2) Use a skiplist to have this operation as O(1) and insertion as O(log(N)).
12  * 寻找第一个快到时的时间事件
13  * 这个操作是有用的知道有多少时间可以选择该事件设置为不用推迟任何事件的睡眠中。
14  * 如果事件链表没有时间将返回NULL。
15  */
16 static aeTimeEvent *aeSearchNearestTimer(aeEventLoop *eventLoop)
17 {
18     // 时间事件头节点地址
19     aeTimeEvent *te = eventLoop->timeEventHead;
20     aeTimeEvent *nearest = NULL;
21 
22     // 遍历所有的时间事件
23     while(te) {
24         // 寻找第一个快到时的时间事件，保存到nearest中
25         if (!nearest || te->when_sec < nearest->when_sec ||
26                 (te->when_sec == nearest->when_sec &&
27                  te->when_ms < nearest->when_ms))
28             nearest = te;
29         te = te->next;
30     }
31     return nearest;
32 }

这个函数就是遍历链表，找到最小值。

4.5.3.4 定时事件的处理

我们重点看执行时间事件的函数processTimeEvents实现

  1 /**
  2  * Process time events
  3  * 执行时间事件
  4  */
  5 static int processTimeEvents(aeEventLoop *eventLoop) {
  6     int processed = 0;
  7     aeTimeEvent *te;
  8     long long maxId;
  9     // 获取当前时间
 10     time_t now = time(NULL);
 11 
 12     /* If the system clock is moved to the future, and then set back to the
 13      * right value, time events may be delayed in a random way. Often this
 14      * means that scheduled operations will not be performed soon enough.
 15      *
 16      * Here we try to detect system clock skews, and force all the time
 17      * events to be processed ASAP when this happens: the idea is that
 18      * processing events earlier is less dangerous than delaying them
 19      * indefinitely, and practice suggests it is. */
 20     /**
 21      * 这段代码的意思是，如果当前时刻小于lastTime, 也就是上一次处理事件的时间比当前时间还要大，
 22      * 那意味着时间有可能被调整了。
 23      *
 24      * 对于这种情况，Redis是怎么处理的呢： 直接把所有的事件的超时时间都置为0, te->when_sec = 0。
 25      * 这样的话，接下来检查有哪些超时时间到期的时候，所有的超时事件都会被判定为到期。
 26      * 相当于本次遍历把所有超时事件一次性全部激活。
 27      *
 28      * 因为Redis认为，在这种异常情况下，与其冒着超时事件可能永远无法触发的风险，还不如把事情提前做了。
 29      *
 30      * 还是基于Redis够用就行的原则，这个解决方案在Redis中显然是被接受的。
 31      *
 32      * 但是其实还有更好的做法，比如libevent就是通过相对时间的方式进行处理这个问题。
 33      * 为了解决这个几乎不会出现的异常case，libevent也花了大量代码进行处理。
 34      */
 35     if (now < eventLoop->lastTime) {
 36         te = eventLoop->timeEventHead;
 37         while(te) {
 38             te->when_sec = 0;
 39             te = te->next;
 40         }
 41     }
 42     // 设置上一次时间事件处理的时间为当前时间
 43     eventLoop->lastTime = now;
 44 
 45     te = eventLoop->timeEventHead;
 46     //当前时间事件表中的最大ID
 47     maxId = eventLoop->timeEventNextId-1;
 48     // 遍历时间事件链表
 49     while(te) {
 50         long now_sec, now_ms;
 51         long long id;
 52 
 53         /* Remove events scheduled for deletion. */
 54         // 如果时间事件已被删除了
 55         if (te->id == AE_DELETED_EVENT_ID) {
 56             aeTimeEvent *next = te->next;
 57             // 从事件链表中删除事件的节点
 58             if (te->prev)
 59                 te->prev->next = te->next;
 60             else
 61                 eventLoop->timeEventHead = te->next;
 62             if (te->next)
 63                 te->next->prev = te->prev;
 64             // 调用时间事件终结方法清楚该事件
 65             if (te->finalizerProc)
 66                 te->finalizerProc(eventLoop, te->clientData);
 67             // 释放内存
 68             zfree(te);
 69             te = next;
 70             continue;
 71         }
 72 
 73         /**
 74          * Make sure we don't process time events created by time events in
 75          * this iteration. Note that this check is currently useless: we always
 76          * add new timers on the head, however if we change the implementation
 77          * detail, this check may be useful again: we keep it here for future
 78          * defense.
 79          * 确保我们不处理在此迭代中由时间事件创建的时间事件。 请注意，此检查目前无效：
 80          * 我们总是在头节点添加新的计时器，但是如果我们更改实施细节，则该检查可能会再次有用：
 81          * 我们将其保留在未来的防御
 82          */
 83         if (te->id > maxId) {
 84             te = te->next;
 85             continue;
 86         }
 87         // 获取当前时间
 88         aeGetTime(&now_sec, &now_ms);
 89         // 找到已经到时的时间事件
 90         if (now_sec > te->when_sec ||
 91             (now_sec == te->when_sec && now_ms >= te->when_ms))
 92         {
 93             int retval;
 94 
 95             id = te->id;
 96             // 调用时间事件处理方法，实际上就是serverCron函数
 97             retval = te->timeProc(eventLoop, id, te->clientData);
 98             // 时间事件次数加1
 99             processed++;
100             // 如果不是定时事件，则继续设置它的到时时间
101             if (retval != AE_NOMORE) {
102                 aeAddMillisecondsToNow(retval,&te->when_sec,&te->when_ms);
103             } else {
104                 // 如果是定时时间，则retval为-1，则将其时间事件删除，惰性删除
105                 te->id = AE_DELETED_EVENT_ID;
106             }
107         }
108         // 更新后继节点指针
109         te = te->next;
110     }
111     //返回执行事件的次数
112     return processed;
113 }

View Code

　　如果时间事件不存在，则就调用finalizerProc指向的回调函数，删除当前的时间事件，为什么会出现这种情况呢，这是因为定时事件到时处理结束后，根据timeProc函数的返回值判断该定时事件是否会继续存在，如果返回值不是-1，则说明该定时事件需要继续存在，则继续设置它的带时间；如果返回值是-1，那么说明该定时事件不需要继续存在，则将该事件的id设置为AE_DELETED_EVENT_ID，然后在下依一次的定时事件的遍历的时候对其进行删除。如果存在，就调用timeProc指向的回调函数处理时间事件。Redis的时间事件分为两类

定时事件：让一段程序在指定的时间后执行一次。
周期性事件：让一段程序每隔指定的时间后执行一次。

如果当前的时间事件是周期性，那么就会在将时间周期添加到周期事件的到时时间中。如果是定时事件，则将该时间事件删除。

五、参考文章

https://blog.csdn.net/men_wen/article/details/71514524

https://blog.csdn.net/wynter_/article/details/53318353

https://blog.csdn.net/chosen0ne/article/details/42717571

https://www.cyhone.com/articles/analysis-of-redis-ae/

本文来自博客园，作者：Mr-xxx，转载请注明原文链接：https://www.cnblogs.com/MrLiuZF/p/15001415.html

查看全文

相关阅读:
浏览器(WebRTC) 如何获取音视频流
 在linux shell中，用$#表示传递进来的参数个数
 Vi很省力
 Factorial! You Must be Kidding!!!（ICPC冬令营集训题目）
Pig-Latin（ICPC冬令营集训题目）
C# .net
编辑器
 VC 和 VS 区别
 Visual Studio和Visual Studio Code差异篇
 IDE 与编辑器的对比

原文地址：https://www.cnblogs.com/MrLiuZF/p/15001415.html