Redis是如何工作的？我们常说的事件驱动、IO多路复用具体是怎么进行落地实现的？下面通过Redis源码进行学习和剖析一看究竟。

源码剖析

main()主方法入口

int main(int argc, char **argv) {

   /*

    *  此处省略一系列系统初始化工作

    */

   // 初始化事件处理器前置函数

   aeSetBeforeSleepProc(server.el,beforeSleep);

   

   // 运行事件处理器，一直到服务器关闭为止

   aeMain(server.el);

   // 服务器关闭，停止事件循环

   aeDeleteEventLoop(server.el);

   return 0;

}

在redis.c中可以找到主方法main()入口，在主方法中一系列的系统初始化工作之后的方法末端，有aeMain(server.el)方法，这里的server.el是aeEventLoop 对象引用，它是对服务器定义事件的封装。

aeEventLoop事件处理封装类

* State of an event based program

*

* 事件处理器的状态

*/

typedef struct aeEventLoop {

   // 目前已注册的最大描述符

   int maxfd;   /* highest file descriptor currently registered */

   // 目前已追踪的最大描述符

   int setsize; /* max number of file descriptors tracked */

   // 用于生成时间事件 id

   long long timeEventNextId;

   // 最后一次执行时间事件的时间

   time_t lastTime;     /* Used to detect system clock skew */

   // 已注册的文件事件

   aeFileEvent *events; /* Registered events */

   // 已就绪的文件事件

   aeFiredEvent *fired; /* Fired events */

   // 时间事件

   aeTimeEvent *timeEventHead;

   // 事件处理器的开关

   int stop;

   // 多路复用库的私有数据

   void *apidata; /* This is used for polling API specific data */

   // 在处理事件前要执行的函数

   aeBeforeSleepProc *beforesleep;

} aeEventLoop;

aeEventLoop封装了服务器事件属性，是一个事件载体

aeCreateEventLoop()事件初始化

/*

* 初始化事件处理器状态

*/

aeEventLoop *aeCreateEventLoop(int setsize) {

   aeEventLoop *eventLoop;

   int i;

   // 创建事件状态结构

   if ((eventLoop = zmalloc(sizeof(*eventLoop))) == NULL) goto err;

   // 初始化文件事件结构和已就绪文件事件结构数组

   eventLoop->events = zmalloc(sizeof(aeFileEvent)*setsize);

   eventLoop->fired = zmalloc(sizeof(aeFiredEvent)*setsize);

   if (eventLoop->events == NULL || eventLoop->fired == NULL) goto err;

   // 设置数组大小

   eventLoop->setsize = setsize;

   // 初始化执行最近一次执行时间

   eventLoop->lastTime = time(NULL);

   // 初始化时间事件结构

   eventLoop->timeEventHead = NULL;

   eventLoop->timeEventNextId = 0;

   eventLoop->stop = 0;

   eventLoop->maxfd = -1;

   eventLoop->beforesleep = NULL;

   if (aeApiCreate(eventLoop) == -1) goto err;

   /* Events with mask == AE_NONE are not set. So let's initialize the

    * vector with it. */

   // 初始化监听事件

   for (i = 0; i < setsize; i++)

       eventLoop->events[i].mask = AE_NONE;

   // 返回事件循环

   return eventLoop;

err:

   if (eventLoop) {

       zfree(eventLoop->events);

       zfree(eventLoop->fired);

       zfree(eventLoop);

   }

   return NULL;

}

Redis定义了三种文件事件状态

• 初始化文件事件（event）结构、已就绪文件事件（fired ）结构数组

• 分别使用连续数组来存储事件信息
  
  eventLoop->events = zmalloc(sizeof(aeFileEvent)*setsize);eventLoop->fired = zmalloc(sizeof(aeFiredEvent)*setsize);
• 数组长度为setsize，通过aeCreateEventLoop(server.maxclients+CONFIG_FDSET_INCR)创建

• maxclients代表用户配置的最大连接数，可在启动时由--maxclients指定，默认为10000。
• CONFIG_FDSET_INCR 大小为128，目的是给Redis预留一些安全空间。

• 初始化监听事件
  将事件状态全部初始化成未设置状态
  
  
  for (i = 0; i < setsize; i++)eventLoop->events[i].mask = AE_NONE;

aeMain() 主循环函数

/*

* 事件处理器的主循环

*/

void aeMain(aeEventLoop *eventLoop) {

   eventLoop->stop = 0;

   while (!eventLoop->stop) {

       // 如果有需要在事件处理前执行的函数，那么运行它

       if (eventLoop->beforesleep != NULL)

           eventLoop->beforesleep(eventLoop);

       // 开始处理事件，这里是处理所有Redis定义的事件类型，即文件事件、时间事件

       aeProcessEvents(eventLoop, AE_ALL_EVENTS);

   }

}

aeMain()事件处理主方法位于ae.c文件中，这里是通过一个主循环轮询处理aeProcessEvents()方法进行所有系统事件处理，也就是常说的Redis是单线程的原因。

Redis定义了两种事件类型

aeProcessEvents()事件处理函数

• 时间事件

• 允许阻塞模式&时间事件：根据时间事件判断文件事件阻塞时长或一直阻塞
• 不允许阻塞模式：非阻塞模式进行
• aeSearchNearestTimer()链表存储时间事件，此处查找时间复杂度是O(n)

• 文件事件

• aeApiPoll() 获取文件事件。Redis根据OS的不同会进行select、epoll、kqueue、evport四种事件函数库的选择，在初始化阶段根据OS的不同进行判断选择最为合适的事件函数库作为事件获取方式，这也是多路复用的重要组成之一。 选择优先级为evport > epoll > kqueue > select。
• rfileProc() 处理文件读状态事件
• wfileProc()处理文件写状态事件

• 事件调度

• 文件事件的阻塞时间是通过时间事件协调进行控制的。这样既可以不长时间进行阻塞在文件事件的等待处理上，又可以有一定的阻塞减少线程空轮询耗费线程资源
• 时间事件是周期性、较为稳定出现的，而文件事件是随机性、不稳定出现的。这里是通过稳定的时间事件来协调不稳定的文件事件阻塞等待或非阻塞轮询
• 对文件事件、时间事件的处理都是同步、有序、原子方式进行的，两种事件类型在单线程的执行下共同协作完成的，因此对两种事件的执行和处理需要额外关注，一旦有一方出现阻塞或长时间执行则会影响整个Redis服务性能

/* Process every pending time event, then every pending file event

* (that may be registered by time event callbacks just processed).

*

* 处理所有已到达的时间事件，以及所有已就绪的文件事件。

*

* Without special flags the function sleeps until some file event

* fires, or when the next time event occurs (if any).

*

* 如果不传入特殊 flags 的话，那么函数睡眠直到文件事件就绪，

* 或者下个时间事件到达（如果有的话）。

*

* If flags is 0, the function does nothing and returns.

* 如果 flags 为 0 ，那么函数不作动作，直接返回。

*

* if flags has AE_ALL_EVENTS set, all the kind of events are processed.

* 如果 flags 包含 AE_ALL_EVENTS ，所有类型的事件都会被处理。

*

* if flags has AE_FILE_EVENTS set, file events are processed.

* 如果 flags 包含 AE_FILE_EVENTS ，那么处理文件事件。

*

* if flags has AE_TIME_EVENTS set, time events are processed.

* 如果 flags 包含 AE_TIME_EVENTS ，那么处理时间事件。

*

* if flags has AE_DONT_WAIT set the function returns ASAP until all

* the events that's possible to process without to wait are processed.

* 如果 flags 包含 AE_DONT_WAIT ，

* 那么函数在处理完所有不许阻塞的事件之后，即刻返回。

*

* The function returns the number of events processed.

* 函数的返回值为已处理事件的数量

*/

int aeProcessEvents(aeEventLoop *eventLoop, int flags)

{

   int processed = 0, numevents;

   /* Nothing to do? return ASAP */

   if (!(flags & AE_TIME_EVENTS) && !(flags & AE_FILE_EVENTS)) return 0;

   /* Note that we want call select() even if there are no

    * file events to process as long as we want to process time

    * events, in order to sleep until the next time event is ready

    * to fire. */

   if (eventLoop->maxfd != -1 ||

       ((flags & AE_TIME_EVENTS) && !(flags & AE_DONT_WAIT))) {

       int j;

       aeTimeEvent *shortest = NULL;

       struct timeval tv, *tvp;

       // 获取最近的时间事件

       if (flags & AE_TIME_EVENTS && !(flags & AE_DONT_WAIT))

        //这里使用链表存储时间事件，因此需要全部遍历出来对比找到时间最近的事件，时间复杂度是O(n)

           shortest = aeSearchNearestTimer(eventLoop);

       if (shortest) {

           // 如果时间事件存在的话

           // 那么根据最近可执行时间事件和现在时间的时间差来决定文件事件的阻塞时间

           long now_sec, now_ms;

           /* Calculate the time missing for the nearest

            * timer to fire. */

           // 计算距今最近的时间事件还要多久才能达到

           // 并将该时间距保存在 tv 结构中

           aeGetTime(&now_sec, &now_ms);

           tvp = &tv;

           tvp->tv_sec = shortest->when_sec - now_sec;

           if (shortest->when_ms < now_ms) {

               tvp->tv_usec = ((shortest->when_ms+1000) - now_ms)*1000;

               tvp->tv_sec --;

           } else {

               tvp->tv_usec = (shortest->when_ms - now_ms)*1000;

           }

           // 时间差小于 0 ，说明事件已经可以执行了，将秒和毫秒设为 0 （不阻塞）

           if (tvp->tv_sec < 0) tvp->tv_sec = 0;

           if (tvp->tv_usec < 0) tvp->tv_usec = 0;

       } else {

           

           // 执行到这一步，说明没有时间事件

           // 那么根据 AE_DONT_WAIT 是否设置来决定是否阻塞，以及阻塞的时间长度

           /* If we have to check for events but need to return

            * ASAP because of AE_DONT_WAIT we need to set the timeout

            * to zero */

           if (flags & AE_DONT_WAIT) {

               // 设置文件事件不阻塞

               tv.tv_sec = tv.tv_usec = 0;

               tvp = &tv;

           } else {

               /* Otherwise we can block */

               // 文件事件可以阻塞直到有事件到达为止

               tvp = NULL; /* wait forever */

           }

       }

       // 处理文件事件，阻塞时间由 tvp 决定

       numevents = aeApiPoll(eventLoop, tvp);

       for (j = 0; j < numevents; j++) {

           // 从已就绪数组中获取事件

           aeFileEvent *fe = &eventLoop->events[eventLoop->fired[j].fd];

           int mask = eventLoop->fired[j].mask;

           int fd = eventLoop->fired[j].fd;

           int rfired = 0;

          /* note the fe->mask & mask & ... code: maybe an already processed

            * event removed an element that fired and we still didn't

            * processed, so we check if the event is still valid. */

           // 读事件

           if (fe->mask & mask & AE_READABLE) {

               // rfired 确保读/写事件只能执行其中一个

               rfired = 1;

               fe->rfileProc(eventLoop,fd,fe->clientData,mask);

           }

           // 写事件

           if (fe->mask & mask & AE_WRITABLE) {

               if (!rfired || fe->wfileProc != fe->rfileProc)

                   fe->wfileProc(eventLoop,fd,fe->clientData,mask);

           }

           processed++;

       }

   }

   /* Check time events */

   // 执行时间事件

   if (flags & AE_TIME_EVENTS)

       processed += processTimeEvents(eventLoop);

   return processed; /* return the number of processed file/time events */

}

aeApiPoll()事件获取函数

这里接触到的是epoll、select，我们例举这两个简单说明下调用逻辑，关于epoll、select的OS底层实现以后开章节再深度剖析，这里是看Redis的多路复用模型即可。

epoll实现

如下是ae_epoll.c中aeApiPoll实现

/*

* 获取可执行事件

*/

static int aeApiPoll(aeEventLoop *eventLoop, struct timeval *tvp) {

   aeApiState *state = eventLoop->apidata;

   int retval, numevents = 0;

   // 等待时间

   retval = epoll_wait(state->epfd,state->events,eventLoop->setsize,

           tvp ? (tvp->tv_sec*1000 + tvp->tv_usec/1000) : -1);

   // 有至少一个事件就绪？

   if (retval > 0) {

       int j;

       // 为已就绪事件设置相应的模式

       // 并加入到 eventLoop 的 fired 数组中

       numevents = retval;

       for (j = 0; j < numevents; j++) {

           int mask = 0;

           struct epoll_event *e = state->events+j;

           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;

           eventLoop->fired[j].fd = e->data.fd;

           eventLoop->fired[j].mask = mask;

       }

   }

   

   // 返回已就绪事件个数

   return numevents;

}

函数方法 int epoll_wait(int epfd, struct epoll_event * events, int maxevents, int timeout);

• epfd 是epoll的描述符。
• events 是分配好的epoll_event结构体数组，epoll将会把发生的事件复制到events数组
• maxevents 表示本次可以返回的最大事件数目，通常maxevents参数与预分配的events数组的大小是相等的。
• timeout 表示在没有检测到事件发生时最多等待的时间（单位为毫秒）

• timeout = null，即不传入时间结构，就是将epoll置于阻塞状态，一定等到监视文件描述符集合中某个文件描述符发生变化为止
• timeout = 0，就变成一个纯粹的非阻塞函数，不管文件描述符是否有变化，都立刻返回继续执行，文件无变化返回0，有变化返回一个正值
• timeout > 0，这就是等待的超时时间，即epoll在timeout时间内阻塞，超时时间之内有事件到来就返回，超时返回0

select实现

如下是ae_select.c中aeApiPoll实现

static int aeApiPoll(aeEventLoop *eventLoop, struct timeval *tvp) {

   aeApiState *state = eventLoop->apidata;

   int retval, j, numevents = 0;

   memcpy(&state->_rfds,&state->rfds,sizeof(fd_set));

   memcpy(&state->_wfds,&state->wfds,sizeof(fd_set));

   retval = select(eventLoop->maxfd+1,

               &state->_rfds,&state->_wfds,NULL,tvp);

   if (retval > 0) {

       for (j = 0; j <= eventLoop->maxfd; j++) {

           int mask = 0;

           aeFileEvent *fe = &eventLoop->events[j];

           if (fe->mask == AE_NONE) continue;

           if (fe->mask & AE_READABLE && FD_ISSET(j,&state->_rfds))

               mask |= AE_READABLE;

           if (fe->mask & AE_WRITABLE && FD_ISSET(j,&state->_wfds))

               mask |= AE_WRITABLE;

           eventLoop->fired[numevents].fd = j;

           eventLoop->fired[numevents].mask = mask;

           numevents++;

       }

   }

   return numevents;

}

函数方法readset, fd_set int select(int maxfdp1, fd_set writeset, struct timeval *timeout);

• maxfdp1  是一个整数值，是指集合中所有文件描述符的范围，即所有文件描述符的最大值加1，不能错。Redis在select()函数中传入了maxfd+1。
• readset 是指向fd_set结构的指针，这个集合中应该包括文件描述符，我们是要监视这些文件描述符的读变化的，即我们关心是否可以从这些文件中读取数据了，如果这个集合中有一个文件可读，select就会返回一个大于0的值，表示有文件可读；如果没有可读的文件，则根据timeout参数再判断是否超时，若超出timeout的时间，select返回0，若发生错误返回负值。可以传入NULL值，表示不关心任何文件的读变化
• writeset 类似readset，这里是写变化
• timeout

• timeout = null，即不传入时间结构，就是将select置于阻塞状态，一定等到监视文件描述符集合中某个文件描述符发生变化为止
• timeout = 0，就变成一个纯粹的非阻塞函数，不管文件描述符是否有变化，都立刻返回继续执行，文件无变化返回0，有变化返回一个正值
• timeout > 0，这就是等待的超时时间，即 select在timeout时间内阻塞，超时时间之内有事件到来就返回，超时返回0

对比

事件驱动模型

参考

《Redis设计与实现》

https://github.com/huangz1990/redis-3.0-annotated

https://zhuanlan.zhihu.com/p/92739237

https://www.cnblogs.com/xuewangkai/p/11158576.html epoll

https://blog.csdn.net/lingfengtengfei/article/details/12392449 select