Redis的事件模型實現基於linux的epoll，sun的export,FreeBSD和Mac osx的queue，還有select；我們簡單分析下Redis基於epoll實現的事件模型。main函式呼叫initServer實現服務初始化：

void initServer(void) {
    int j;
	
	//SIG_DFL：預設訊號處理程式
	//SIG_IGN：忽略訊號的處理程式
    signal(SIGHUP, SIG_IGN);
    signal(SIGPIPE, SIG_IGN);
	
	//設定訊號處理函式
    setupSignalHandlers();

    if (server.syslog_enabled) {
        openlog(server.syslog_ident, LOG_PID | LOG_NDELAY | LOG_NOWAIT,
            server.syslog_facility);
    }

    server.pid = getpid();
    server.current_client = NULL;
	//客戶端連結串列
    server.clients = listCreate();
	//非同步關閉的客戶端
    server.clients_to_close = listCreate();
	//從機
    server.slaves = listCreate();
	
    server.monitors = listCreate();
	
    server.clients_pending_write = listCreate();
	
    server.slaveseldb = -1; /* Force to emit the first SELECT command. */
	
    server.unblocked_clients = listCreate();
	
    server.ready_keys = listCreate();
	
    server.clients_waiting_acks = listCreate();
	
    server.get_ack_from_slaves = 0;
	
    server.clients_paused = 0;
	//
    server.system_memory_size = zmalloc_get_memory_size();

    createSharedObjects();
	//
    adjustOpenFilesLimit();

	//構建aeEventLoop物件
    server.el = aeCreateEventLoop(server.maxclients+CONFIG_FDSET_INCR);
	//
    server.db = zmalloc(sizeof(redisDb)*server.dbnum);

    /* Open the TCP listening socket for the user commands. */
    if (server.port != 0 &&
		//啟動TCP監聽服務 --> 並設定為非阻塞
        listenToPort(server.port,server.ipfd,&server.ipfd_count) == C_ERR)
        exit(1);

    /* Open the listening Unix domain socket. */

	//啟動本地監聽服務
    if (server.unixsocket != NULL) {
        unlink(server.unixsocket); /* don't care if this fails */
        server.sofd = anetUnixServer(server.neterr,server.unixsocket,
            server.unixsocketperm, server.tcp_backlog);
        if (server.sofd == ANET_ERR) {
            serverLog(LL_WARNING, "Opening Unix socket: %s", server.neterr);
            exit(1);
        }
        anetNonBlock(NULL,server.sofd);
    }

    /* Abort if there are no listening sockets at all. */

	//如果沒有監聽套接字，則退出 --->初始化預設為0 -- > 在listenToPort每成功建立一個監聽套接字，自加1
    if (server.ipfd_count == 0 && server.sofd < 0) {
        serverLog(LL_WARNING, "Configured to not listen anywhere, exiting.");
        exit(1);
    }

    /* Create the Redis databases, and initialize other internal state. */
    for (j = 0; j < server.dbnum; j++) {
        server.db[j].dict = dictCreate(&dbDictType,NULL);
        server.db[j].expires = dictCreate(&keyptrDictType,NULL);
        server.db[j].blocking_keys = dictCreate(&keylistDictType,NULL);
        server.db[j].ready_keys = dictCreate(&setDictType,NULL);
        server.db[j].watched_keys = dictCreate(&keylistDictType,NULL);
        server.db[j].eviction_pool = evictionPoolAlloc();
        server.db[j].id = j;
        server.db[j].avg_ttl = 0;
    }
    server.pubsub_channels = dictCreate(&keylistDictType,NULL);
    server.pubsub_patterns = listCreate();
    listSetFreeMethod(server.pubsub_patterns,freePubsubPattern);
    listSetMatchMethod(server.pubsub_patterns,listMatchPubsubPattern);
    server.cronloops = 0;
    server.rdb_child_pid = -1;
    server.aof_child_pid = -1;
    server.rdb_child_type = RDB_CHILD_TYPE_NONE;
    server.rdb_bgsave_scheduled = 0;
    aofRewriteBufferReset();
    server.aof_buf = sdsempty();
    server.lastsave = time(NULL); /* At startup we consider the DB saved. */
    server.lastbgsave_try = 0;    /* At startup we never tried to BGSAVE. */
    server.rdb_save_time_last = -1;
    server.rdb_save_time_start = -1;
    server.dirty = 0;
    resetServerStats();
    /* A few stats we don't want to reset: server startup time, and peak mem. */
    server.stat_starttime = time(NULL);
    server.stat_peak_memory = 0;
    server.resident_set_size = 0;
    server.lastbgsave_status = C_OK;
    server.aof_last_write_status = C_OK;
    server.aof_last_write_errno = 0;
    server.repl_good_slaves_count = 0;
    updateCachedTime();

    /* Create the serverCron() time event, that's our main way to process
     * background operations. */
    if(aeCreateTimeEvent(server.el, 1, serverCron, NULL, NULL) == AE_ERR) {
        serverPanic("Can't create the serverCron time event.");
        exit(1);
    }

    /* Create an event handler for accepting new connections in TCP and Unix
     * domain sockets. */

	//建立epoll事件監聽  --> 一個tcp套接字一個監聽檔案
    for (j = 0; j < server.ipfd_count; j++) {
        if (aeCreateFileEvent(server.el, server.ipfd[j], AE_READABLE,
            acceptTcpHandler,NULL) == AE_ERR)
            {
                serverPanic(
                    "Unrecoverable error creating server.ipfd file event.");
            }
    }
	//建立epoll事件監聽  --> 一個本地套接字一個監聽檔案
    if (server.sofd > 0 && aeCreateFileEvent(server.el,server.sofd,AE_READABLE,
        acceptUnixHandler,NULL) == AE_ERR) serverPanic("Unrecoverable error creating server.sofd file event.");

    /* Open the AOF file if needed. */
    if (server.aof_state == AOF_ON) {
        server.aof_fd = open(server.aof_filename,
                               O_WRONLY|O_APPEND|O_CREAT,0644);
        if (server.aof_fd == -1) {
            serverLog(LL_WARNING, "Can't open the append-only file: %s",
                strerror(errno));
            exit(1);
        }
    }

    /* 32 bit instances are limited to 4GB of address space, so if there is
     * no explicit limit in the user provided configuration we set a limit
     * at 3 GB using maxmemory with 'noeviction' policy'. This avoids
     * useless crashes of the Redis instance for out of memory. */
    if (server.arch_bits == 32 && server.maxmemory == 0) {
        serverLog(LL_WARNING,"Warning: 32 bit instance detected but no memory limit set. Setting 3 GB maxmemory limit with 'noeviction' policy now.");
        server.maxmemory = 3072LL*(1024*1024); /* 3 GB */
        server.maxmemory_policy = MAXMEMORY_NO_EVICTION;
    }

    if (server.cluster_enabled) clusterInit();
    replicationScriptCacheInit();
    scriptingInit(1);
    slowlogInit();
    latencyMonitorInit();
    bioInit();
}
 

我們主要關注三個地方

1 構建aeEventLoop物件

	//構建aeEventLoop物件
    server.el = aeCreateEventLoop(server.maxclients+CONFIG_FDSET_INCR);

2 建立套接字監聽，分為TCP監聽和本地套接字監聽

    /* Open the TCP listening socket for the user commands. */
    if (server.port != 0 &&
		//啟動TCP監聽服務 --> 並設定為非阻塞
        listenToPort(server.port,server.ipfd,&server.ipfd_count) == C_ERR)
        exit(1);

    /* Open the listening Unix domain socket. */

	//啟動本地監聽服務
    if (server.unixsocket != NULL) {
        unlink(server.unixsocket); /* don't care if this fails */
        server.sofd = anetUnixServer(server.neterr,server.unixsocket,
            server.unixsocketperm, server.tcp_backlog);
        if (server.sofd == ANET_ERR) {
            serverLog(LL_WARNING, "Opening Unix socket: %s", server.neterr);
            exit(1);
        }
        anetNonBlock(NULL,server.sofd);
    }
 

3 將建立監聽的套接字加入EventLoop事件監聽

	//建立epoll事件監聽  --> 一個tcp套接字一個監聽檔案
    for (j = 0; j < server.ipfd_count; j++) {
        if (aeCreateFileEvent(server.el, server.ipfd[j], AE_READABLE,
            acceptTcpHandler,NULL) == AE_ERR)
            {
                serverPanic(
                    "Unrecoverable error creating server.ipfd file event.");
            }
    }
	//建立epoll事件監聽  --> 一個本地套接字一個監聽檔案
    if (server.sofd > 0 && aeCreateFileEvent(server.el,server.sofd,AE_READABLE,
        acceptUnixHandler,NULL) == AE_ERR) serverPanic("Unrecoverable error creating server.sofd file event."); 
 

在分析如何構建aeEventLoop物件之前，我們先看下aeEventLoop的定義：

/* State of an event based program */
typedef struct aeEventLoop {
    int maxfd;   				//監聽的最大檔案號  
    int setsize; 				//可以註冊的事件的上限，預設為1024*10
    long long timeEventNextId;	//定時器事件的ID編號管理（分配ID號所用）  
    time_t lastTime;     		/* Used to detect system clock skew */
    aeFileEvent *events; 		//註冊的檔案事件，這些是需要程序關注的檔案  
    aeFiredEvent *fired; 		//poll結果，待處理的檔案事件的檔案號和事件型別  
    aeTimeEvent *timeEventHead;	//定時器時間連結串列 
    int stop;		 			//時間輪詢是否結束？  
    void *apidata; 				//檔案事件的輪詢資料和結果資料：poll； 三種輪詢方式：epoll（linux），select（windows），kqueue  
    aeBeforeSleepProc *beforesleep;
} aeEventLoop; 

接著分析aeCreateEventLoop函式：

//底層epoll多路複用初始化，然後存放在aeEventLoop中 void * 型別的apidata，隱藏了底層的實現。

/*****************************************************************************
 * 函 數 名  : aeCreateEventLoop
 * 函式功能  : 構建aeEventLoop物件eventLoop  
 * 輸入引數  : int setsize  訊息佇列大小
 * 輸出引數  : 無
 * 返 回 值  : aeEventLoop
 * 呼叫關係  : 
 * 記    錄
 * 1.日    期: 2017年10月21日
 *   作    者: zyz
 *   修改內容: 新生成函式
*****************************************************************************/
aeEventLoop *aeCreateEventLoop(int setsize) {
	//setsize指的是放到eventloop中最大描述符大小，也就是該事件迴圈中可能有多少個fd。
    aeEventLoop *eventLoop;
    int i;

    if ((eventLoop = zmalloc(sizeof(*eventLoop))) == NULL) goto err;
    eventLoop->events = zmalloc(sizeof(aeFileEvent)*setsize);
    eventLoop->fired = zmalloc(sizeof(aeFiredEvent)*setsize);
	//eventLoop->events和event_loop->fired都是一個set size大小的
	//陣列，表明了event loop是用陣列下標來當fd。
    if (eventLoop->events == NULL || eventLoop->fired == NULL) goto err;
    eventLoop->setsize = setsize;
    eventLoop->lastTime = time(NULL);
    eventLoop->timeEventHead = NULL;
    eventLoop->timeEventNextId = 0;
	//eventLoop->timeEventHead儲存第一個時間事件，值得注意的是時間事件
	//連結串列不是有序列表，因此會後面的檢索會是O(n)
	
    eventLoop->stop = 0;			 	//指示事件迴圈是否繼續進行。
    eventLoop->maxfd = -1;			 	//事件連結串列中目前最大的fd，用於減小檢索範圍
    eventLoop->beforesleep = NULL;		//每輪迴圈前的回撥函式
    // aeApiCreate封裝了不同多路複用的實現，如linux的epoll，sun的export,FreeBSD和Mac osx的queue，還有select。
    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;// AE_NONE來指示該fd沒有放入事件迴圈庫
    return eventLoop;

err:
    if (eventLoop) {
        zfree(eventLoop->events);
        zfree(eventLoop->fired);
        zfree(eventLoop);
    }
    return NULL;
} 

很簡單，分配一個aeEventLoop 物件，然後一折物件為引數，呼叫底層實現函式aeApiCreate，我們看下epoll型別的實現函式：

/*****************************************************************************
 * 函 數 名  : aeApiCreate
 * 函式功能  : 封裝epoll事件驅動構建函式
 * 輸入引數  : aeEventLoop *eventLoop  事件迴圈描述結構物件
 * 輸出引數  : 無
 * 返 回 值  : static
 * 呼叫關係  : 
 * 記    錄
 * 1.日    期: 2017年10月23日
 *   作    者: zyz
 *   修改內容: 新生成函式
*****************************************************************************/
static int aeApiCreate(aeEventLoop *eventLoop) {
	//包含一個epoll專用檔案描述符和一個epoll_event物件指標
    aeApiState *state = zmalloc(sizeof(aeApiState));

    if (!state) return -1;
	//建立setsize個epoll_event
    state->events = zmalloc(sizeof(struct epoll_event)*eventLoop->setsize);
    if (!state->events) {
        zfree(state);
        return -1;
    }
	//該函式生成一個epoll專用的檔案描述符。它其實是在核心申請一空間，用來存放你想關注的socket fd上是否發生以及發生了什麼事件
    state->epfd = epoll_create(1024); /* 1024 is just a hint for the kernel */
    if (state->epfd == -1) {
        zfree(state->events);
        zfree(state);
        return -1;
    }
    eventLoop->apidata = state;
    return 0;
} 

非常簡單粗暴，構建一個aeApiState物件，這個物件包含一個檔案描述符和一個epoll_event 物件指標。

typedef struct aeApiState {
    int epfd;
    struct epoll_event *events;
} aeApiState;

然後呼叫epoll_create建立事件模型，將得到檔案描述符賦值給aeApiState物件的epfd；分配指定size的poll_event物件記憶體，將地址賦值給aeApiState物件的events；然後aeApiState物件地址賦值給aeEventLoop物件的apidata；

第二件事是建立套接字監聽，不是這裡關注的重點，我們只需知道得到一個存放套接字描述符的陣列。這些套接字就是被監聽的物件。

第三件事是將這些套接字封裝後加入epoll監聽服務。我們看下函式aeCreateFileEvent：

/*****************************************************************************
 * 函 數 名  : aeCreateFileEvent
 * 函式功能  : 建立事件監聽物件
 * 輸入引數  : aeEventLoop *eventLoop  事件監聽服務描述符物件指標
               int fd                  被監聽檔案描述符
               int mask                監聽事件掩碼
               aeFileProc *proc        事件響應函式
               void *clientData        引數
 * 輸出引數  : 無
 * 返 回 值  : 
 * 呼叫關係  : 
 * 記    錄
 * 1.日    期: 2018年03月06日
 *   作    者: zyz
 *   修改內容: 新生成函式
*****************************************************************************/
int aeCreateFileEvent(aeEventLoop *eventLoop, int fd, int mask,
        aeFileProc *proc, void *clientData)
{
    if (fd >= eventLoop->setsize) {
        errno = ERANGE;
        return AE_ERR;
    }
    aeFileEvent *fe = &eventLoop->events[fd];
	//呼叫底層實現，將被監聽檔案描述符加入監聽服務
    if (aeApiAddEvent(eventLoop, fd, mask) == -1)
        return AE_ERR;
    fe->mask |= mask;
    if (mask & AE_READABLE) fe->rfileProc = proc;
    if (mask & AE_WRITABLE) fe->wfileProc = proc;
    fe->clientData = clientData;
    if (fd > eventLoop->maxfd)
        eventLoop->maxfd = fd;
    return AE_OK;
}

在繼續分析之前我們先看下一個結構體：

/* File event structure */
/* 檔案事件結構體 */  
typedef struct aeFileEvent {
	//只為讀事件或者寫事件中的1種	
    int mask; /* one of AE_(READABLE|WRITABLE) */
	//讀方法  
    aeFileProc *rfileProc;
	//寫方法 
    aeFileProc *wfileProc;
	 //客戶端資料  
    void *clientData;
} aeFileEvent;

在呼叫aeCreateEventLoop函式時，分配了指定大小的記憶體，用於在抽象層代表可以監聽指定數量的檔案描述符。在函式中就根據指定檔案描述符取出指定的aeFileEvent 物件，然後以該物件為引數，呼叫底層實現函式aeApiAddEvent：

//EPOLL_CTL_ADD：	註冊新的fd到epfd中；
//EPOLL_CTL_MOD：	修改已經註冊的fd的監聽事件；
//EPOLL_CTL_DEL：	從epfd中刪除一個fd；

/*****************************************************************************
 * 函 數 名  : aeApiAddEvent
 * 函式功能  : epoll事件註冊函式封裝
 * 輸入引數  : aeEventLoop *eventLoop  事件驅動模型結構物件
               int fd                  被監聽事件的檔案描述符
               int mask                事件掩碼
 * 輸出引數  : 無
 * 返 回 值  : static
 * 呼叫關係  : 
 * 記    錄
 * 1.日    期: 2017年10月23日
 *   作    者: zyz
 *   修改內容: 新生成函式
*****************************************************************************/
static int aeApiAddEvent(aeEventLoop *eventLoop, int fd, int mask) {
    aeApiState *state = eventLoop->apidata;
    struct epoll_event ee = {0}; /* avoid valgrind warning */
    /* If the fd was already monitored for some event, we need a MOD
     * operation. Otherwise we need an ADD operation. */
    int op = eventLoop->events[fd].mask == AE_NONE ?
            EPOLL_CTL_ADD : EPOLL_CTL_MOD;
	
	//events可以是以下幾個巨集的集合：
	//EPOLLIN：         觸發該事件，表示對應的檔案描述符上有可讀資料。(包括對端SOCKET正常關閉)；
	//EPOLLOUT：        觸發該事件，表示對應的檔案描述符上可以寫資料；
	//EPOLLPRI：        表示對應的檔案描述符有緊急的資料可讀（這裡應該表示有帶外資料到來）；
	//EPOLLERR：        表示對應的檔案描述符發生錯誤；
	//EPOLLHUP：        表示對應的檔案描述符被結束通話；
	//EPOLLET：         將EPOLL設為邊緣觸發(Edge Triggered)模式，這是相對於水平觸發(Level Triggered)來說的。
	//EPOLLONESHOT：    只監聽一次事件，當監聽完這次事件之後，如果還需要繼續監聽這個socket的話，需要再次把這個socket加入到EPOLL佇列裡。
    ee.events = 0;
    mask |= eventLoop->events[fd].mask; /* Merge old events */
    if (mask & AE_READABLE) ee.events |= EPOLLIN;
    if (mask & AE_WRITABLE) ee.events |= EPOLLOUT;
    ee.data.fd = fd;
    if (epoll_ctl(state->epfd,op,fd,&ee) == -1) return -1;
    return 0;
}

基於epoll的實現方式，實際呼叫的是epoll_ctl，將被監聽檔案描述符加入監聽服務。在這裡事件掩碼為AE_READABLE，對應到epoll為EPOLLIN，即只對輸入事件感興趣。

到這裡，事件監聽服務已經構建完成，下面就是事件處理的部分了

事件處理部分在函式中實現，在main函式中呼叫

/*****************************************************************************
 * 函 數 名  : aeMain
 * 函式功能  : 事件迴圈處理
 * 輸入引數  : aeEventLoop *eventLoop  事件監聽服務描述符物件指標
 * 輸出引數  : 無
 * 返 回 值  : 
 * 呼叫關係  : 
 * 記    錄
 * 1.日    期: 2018年03月06日
 *   作    者:  
 *   修改內容: 新生成函式
*****************************************************************************/
void aeMain(aeEventLoop *eventLoop) {
    eventLoop->stop = 0;
    while (!eventLoop->stop) {						//以死迴圈的方式進入輪詢，直到stop被置為非0
        if (eventLoop->beforesleep != NULL)
            eventLoop->beforesleep(eventLoop);		//前期處理，回撥函式
		//處理訊息
        aeProcessEvents(eventLoop, AE_ALL_EVENTS);	//事件處理（見下面）  
    }
}