redis源码分析--事件驱动模型

redis的服务器是一个事件驱动模型。驱动整个服务运转的关键技术就是IO多路复用，我认为，epoll(linux下的多路复用)是整个redis服务的"发动机"。

既然是事件驱动，那redis中的事件是什么呢？分为两类事件：文件事件(socket可读或可写)和时间事件(定时任务)，redis表示事件循环中的事件封装的结构体是struct aeEventLoop

ae.h

/* State of an event based program */
typedef struct aeEventLoop {
    int maxfd;   /* highest file descriptor currently registered */
    int setsize; /* max number of file descriptors tracked */
    long long timeEventNextId;
    time_t lastTime;     /* Used to detect system clock skew */
    aeFileEvent *events; /* Registered events */ /*文件事件数组，存储所有注册的文件事件*/
    aeFiredEvent *fired; /* Fired events */      /*存储被触发的文件事件*/
    aeTimeEvent *timeEventHead;                  /*事件事件链表的头结点，所有的定时任务存储在该链表中，这是一个无序的链表，因此处理超时事件的复杂度为O(n),这个数据结构可以优化*/
    int stop;     /*标识时间循环是否结束*/
    void *apidata; /* This is used for polling API specific data */
    aeBeforeSleepProc *beforesleep;       /*调用epool_wait阻塞程序之前调用*/
    aeBeforeSleepProc *aftersleep;        /*阻塞程序之后调用，epoll_wait之后还要处理定时事件，因此epoll_wait阻塞的时间需要关注*/
    int flags;
} aeEventLoop;

事件驱动程序的写法一般都是固定：一个死循环，等待事件的发生并处理，处理完开始下一次循环，redis的写法也是如此,在ae.c中：

ae.c

void aeMain(aeEventLoop *eventLoop) {
    eventLoop->stop = 0;
    while (!eventLoop->stop) {
        aeProcessEvents(eventLoop, AE_ALL_EVENTS|
                                   AE_CALL_BEFORE_SLEEP|
                                   AE_CALL_AFTER_SLEEP);
    }
}

//aeProcessEvents是事件处理的主函数，有两个参数。eventLoop是要处理的事件，AE_ALL_EVENTS表示处理所有事件，AE_CALL_BEFORE_SLEEP和AE_CALL_AFTER_SLEEP表示在调用epoll_wait阻塞之前和之后分别要执行beforesleep和aftersleep函数

下来我们分别看一下文件事件和事件事件的处理。

文件事件：

redis分为客户端程序和服务器程序，客户端通过TCP socket与服务器连接交互，因此，文件事件指的是socket可读可写事件。socket读写操作分为阻塞和非阻塞模式，redis采用的是非阻塞IO模式。

在非阻塞模式下，可以使用IO多路复用来同时处理多条网络连接。redis会根据不同的操作系统采用不同的多路复用机制，linux上使用的是epoll。

ae.c

int aeProcessEvents(aeEventLoop *eventLoop, int flags)
{
    //......
    aeTimeEvent *shortest = NULL;
    if (flags & AE_TIME_EVENTS && !(flags & AE_DONT_WAIT))
            shortest = aeSearchNearestTimer(eventLoop);
    //aeSearchNearestTimer函数的功能是遍历时间事件链表，返回最近超时的时间时间，这个参数在后边的epool_wait中要用
    //......
    //aeApiPoll函数是对IO多路复用机制的封装，在linux下调用的是epoll
    numevents = aeApiPoll(eventLoop, tvp);
    //......
    //对epool_wait返回的可读可写事件分别执行读写操作
    for (j = 0; j < numevents; j++) {
        if (!invert && fe->mask & mask & AE_READABLE) {
             fe->rfileProc(eventLoop,fd,fe->clientData,mask);
        }
        if (fe->mask & mask & AE_WRITABLE) {
            fe->wfileProc(eventLoop,fd,fe->clientData,mask);
        }    
    
    //处理时间时间
    processed += processTimeEvents(eventLoop);
}

 aeProcessEvents是redis事件循环的执行函数，该函数的执行流程可以总结为：

 （1）调用aeSearchNearestTimer函数遍历时间事件链表，找到最近要发生的超时事件

 （2）调用aeApiPoll 执行IO多路复用函数(linux下调用epoll)，阻塞等待文件事件的发生。这里需要注意的是，调用aeApiPoll时传的第二个参数是个时间时间结构体aeTimeEvent，这是从时间时间链表中找到的最早的超时时间。该参数有什么用呢? 答案是，调用epoll_wait时需要传入一个超时事件的参数，这个参数表示的意思是阻塞等待的最长时间，如果在该超时时间之内，还没有事件准备就绪的话，epoll_wait就会返回。这里把这个参数设置为最早的超时事件，目的是为了保证定时器的精度，即如果没有文件事件准备就绪的话，最早的超时事件也会被处理。

（3）回调处理文件事件

（4）调用processTimeEvents处理时间事件。

aeApiPoll函数是对IO多路复用的封装，具体实现在ae_epoll.c/ae_kqueue.c/ae_epoll.c中（根据操作系统选择）。linux下执行的是ae_epoll.c中的：

ae_epoll.c

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;

        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|AE_READABLE;
            if (e->events & EPOLLHUP) mask |= AE_WRITABLE|AE_READABLE;
            eventLoop->fired[j].fd = e->data.fd;
            eventLoop->fired[j].mask = mask;
        }
    }
    return numevents;
}

 这是非常标准的也是固定的epoll的写法。函数首先需要通过eventLoop->apidata字段获取epoll模型对应的aeApiState结构体对象，才能调用epoll_wait函数等待事件的发生;epoll_wait函数将已触发的事件存储到aeApiState对象的events字段，Redis再次遍历所有已触发事件，将其封装在eventLoop->fired数组，数组元素类型为结构体aeFiredEvent，只有两个字段，fd表示发生事件的socket文件描述符，mask表示发生的事件类型，如AE_READABLE可读事件和AE_WRITABLE可写事件。

epoll_wait返回可读可写的文件描述符后，程序会回调注册的读写函数，循环处理每一个文件描述符上的读写事件。文件事件的结构体为：

/* File event structure */
typedef struct aeFileEvent {
    int mask; /* one of AE_(READABLE|WRITABLE|BARRIER) */
    aeFileProc *rfileProc;
    aeFileProc *wfileProc;
    void *clientData;
} aeFileEvent;

mask∶存储监控的文件事件类型，如AE_READABLE可读事件和AE_WRITABLE可写事件;
rfileProc∶为函数指针，指向读事件处理函数;
wfileProc∶同样为函数指针，指向写事件处理函数;
clientData∶指向对应的客户端对象。

Redis服务器启动时需要创建socket并监听，（initServer函数）等待客户端连接;客户端与服务器建立socket连接之后，服务器会等待客户端的命令请求;服务器处理完客户端的命令请求之后，命令回复会暂时缓存在client结构体的buf缓冲区，待客户端文件描述符的可写事件发生时，才会真正往客户端发送回复命令。这些都需要创建对应的文件事件。

server.c

void initServer(void){
    //........
    aeCreateFileEvent(server.el,server.ipfd[j],AE_READABLE,
acceptTcpHandler,NULL);
        
    aeCreateFileEvent(server.el, fd,AE_READABLE,
readQueryFromClient,c);
 
    aeCreateFileEvent(server.el,c->fd, ae_flags,
sendReplyToClient,Cc);
    //........
}

时间事件：

时间事件的处理在文件事件之后，执行的函数为processTimeEvents。事件时间的处理就是处理定时任务。多个定时任务被串成链表，链表头结点timeEventHead存储在aeEventLoop结构体中，之前已经提到过。时间事件aeTimeEvent的结构体定义为：

/* Time event structure */
typedef struct aeTimeEvent {
    long long id; /* time event identifier. 事件时间唯一ID*/
    long when_sec; /* seconds 超时事件触发的秒数*/
    long when_ms; /* milliseconds 超时事件触发的毫秒数*/
    aeTimeProc *timeProc  /*处理超时事件的函数指针*/;
    aeEventFinalizerProc *finalizerProc  /*函数指针，删除超时事件节点之前调用*/;
    void *clientData;  /*指向对应的客户端对象*/
    struct aeTimeEvent *prev;  //双向链表前向节点，指向前一个超时事件
    struct aeTimeEvent *next;  //双向链表后继节点，指向后一个超时事件
    int refcount; /* refcount to prevent timer events from being
  		   * freed in recursive time event calls. 引用计数，防止重复释放*/
} aeTimeEvent;

时间事件执行函数processTimeEvents的处理逻辑比较简单，只是遍历时间事件链表，判断当前时间事件是否已经到期，如果到期则执行时间事件处理函数timeProc∶

static int processTimeEvents(aeEventLoop *eventLoop) {
    int processed = 0;
    aeTimeEvent *te;
    long long maxId;
    time_t now = time(NULL);
    /*
        如果将系统时钟移到未来，然后将其设置回正确的值，则时间事件可能会以随机方式延迟。这通常意味着预定的操作不会很快执行。
在这里，我们尝试检测系统时钟偏差，并强制所有时间事件在发生这种情况时尽快处理：其思想是，更早地处理事件比无限期地延迟事件危险性小，实践表明确实如此。
    */
    /*
    eventLoop->lastTime记录的是系统事件偏差，如果当前时间比这个时间早，就让这个超时事件提前执行
    */
    if (now < eventLoop->lastTime) {
        te = eventLoop->timeEventHead;
        while(te) {
            te->when_sec = 0;
            te = te->next;
        }
    }
    eventLoop->lastTime = now;

    te = eventLoop->timeEventHead;
    maxId = eventLoop->timeEventNextId-1;
    /*
    遍历链表，处理超时事件
    */
    while(te) {
        long now_sec, now_ms;
        long long id;
        
        /* Remove events scheduled for deletion. 处理过的事件或无效事件删除掉*/
        if (te->id == AE_DELETED_EVENT_ID) {
            aeTimeEvent *next = te->next;
            /* If a reference exists for this timer event,
             * don't free it. This is currently incremented
             * for recursive timerProc calls */
            if (te->refcount) {
                te = next;
                continue;
            }
            if (te->prev)
                te->prev->next = te->next;
            else
                eventLoop->timeEventHead = te->next;
            if (te->next)
                te->next->prev = te->prev;
            if (te->finalizerProc)
                te->finalizerProc(eventLoop, te->clientData);
            zfree(te);
            te = next;
            continue;
        }

        if (te->id > maxId) {
            te = te->next;
            continue;
        }
        aeGetTime(&now_sec, &now_ms);
        //获取当前时间，如果当前时间大于或等于超时事件，说明该超时事件已到，调用回调函数去处理
        if (now_sec > te->when_sec ||
            (now_sec == te->when_sec && now_ms >= te->when_ms))
        {
            int retval;

            id = te->id;
            te->refcount++;
            retval = te->timeProc(eventLoop, id, te->clientData);
            te->refcount--;
            processed++;
            //是否需要循环执行，是则重新加入超时事件链表，等待下一次执行
            if (retval != AE_NOMORE) {
                aeAddMillisecondsToNow(retval,&te->when_sec,&te->when_ms);
            } else {
                te->id = AE_DELETED_EVENT_ID;
            }
        }
        te = te->next;
    }
    return processed;
}