關于Redis網絡模型的源碼詳析
前言
Redis的網絡模型是基于I/O多路復用程序來實現(xiàn)的。源碼中包含四種多路復用函數庫epoll、select、evport、kqueue。在程序編譯時會根據系統(tǒng)自動選擇這四種庫其中之一。下面以epoll為例，來分析Redis的I/O模塊的源碼。
epoll系統(tǒng)調用方法
Redis網絡事件處理模塊的代碼都是圍繞epoll那三個系統(tǒng)方法來寫的。先把這三個方法弄清楚，后面就不難了。
epfd = epoll_create(1024);
創(chuàng)建epoll實例
參數：表示該 epoll 實例最多可監(jiān)聽的 socket fd（文件描述符）數量。
返回： epoll 專用的文件描述符。
int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event)
管理epoll中的事件，對事件進行注冊、修改和刪除。
參數：
epfd：epoll實例的文件描述符；
op：取值三種：EPOLL_CTL_ADD 注冊、EPOLL_CTL_MOD 修 改、EPOLL_CTL_DEL 刪除；
fd：socket的文件描述符；
epoll_event *event：事件
event代表一個事件，類似于Java NIO中的channel“通道”。epoll_event 的結構如下：
typedef union epoll_data {
void *ptr;
int fd; /* socket文件描述符 */
__uint32_t u32;
__uint64_t u64;
} epoll_data_t;
struct epoll_event {
__uint32_t events; /* Epoll events 就是各種待監(jiān)聽操作的操作碼求與的結果，例如EPOLLIN（fd可讀）、EPOLLOUT（fd可寫） */
epoll_data_t data; /* User data variable */
};
int epoll_wait(int epfd, struct epoll_event * events, intmaxevents, int timeout);
等待事件是否就緒，類似于Java NIO中 select 方法。如果事件就緒，將就緒的event存入events數組中。
參數
epfd：epoll實例的文件描述符；
events：已就緒的事件數組；
intmaxevents：每次能處理的事件數；
timeout：阻塞時間，等待產生就緒事件的超時值。
源碼分析
事件
Redis事件系統(tǒng)中將事件分為兩種類型：
- 文件事件；網絡套接字對應的事件；
- 時間事件：Redis中一些定時操作事件，例如 serverCron 函數。
下面從事件的注冊、觸發(fā)兩個流程對源碼進行分析
綁定事件
建立 eventLoop
在 initServer方法（由 redis.c 的 main 函數調用） 中,在建立 RedisDb 對象的同時，會初始化一個“eventLoop”對象，我稱之為事件處理器對象。結構體的關鍵成員變量如下所示：
struct aeEventLoop｛ aeFileEvent *events;//已注冊的文件事件數組 aeFiredEvent *fired;//已就緒的文件事件數組 aeTimeEvent *timeEventHead;//時間事件數組 ... ｝
初始化 eventLoop 在 ae.c 的“aeCreateEventLoop”方法中執(zhí)行。該方法中除了初始化 eventLoop 還調用如下方法初始化了一個 epoll 實例。
/*
* ae_epoll.c
* 創(chuàng)建一個新的 epoll 實例，并將它賦值給 eventLoop
*/
static int aeApiCreate(aeEventLoop *eventLoop) {
aeApiState *state = zmalloc(sizeof(aeApiState));
if (!state) return -1;
// 初始化事件槽空間
state->events = zmalloc(sizeof(struct epoll_event)*eventLoop->setsize);
if (!state->events) {
zfree(state);
return -1;
}
// 創(chuàng)建 epoll 實例
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
eventLoop->apidata = state;
return 0;
}
也正是在此處調用了系統(tǒng)方法“epoll_create”。這里的state是一個aeApiState結構，如下所示：
/*
* 事件狀態(tài)
*/
typedef struct aeApiState {
// epoll 實例描述符
int epfd;
// 事件槽
struct epoll_event *events;
} aeApiState;
這個 state 由 eventLoop->apidata 來記錄。
綁定ip端口與句柄
通過 listenToPort 方法開啟TCP端口，每個IP端口會對應一個文件描述符 ipfd（因為服務器可能會有多個ip地址）
// 打開 TCP 監(jiān)聽端口，用于等待客戶端的命令請求 if (server.port != 0 && listenToPort(server.port,server.ipfd,&server.ipfd_count) == REDIS_ERR) exit(1);
注意：*eventLoop 和 ipfd 分別被 server.el 和 server.ipfd[] 引用。server 是結構體 RedisServer 的實例，是Redis的全局變量。
注冊事件
如下所示代碼，為每一個文件描述符綁定一個事件函數
// initServer方法：
for (j = 0; j < server.ipfd_count; j++) {
if (aeCreateFileEvent(server.el, server.ipfd[j], AE_READABLE,
acceptTcpHandler,NULL) == AE_ERR)
{
redisPanic(
"Unrecoverable error creating server.ipfd file event.");
}
}
// ae.c 中的 aeCreateFileEvent 方法
/*
* 根據 mask 參數的值，監(jiān)聽 fd 文件的狀態(tài)，
* 當 fd 可用時，執(zhí)行 proc 函數
*/
int aeCreateFileEvent(aeEventLoop *eventLoop, int fd, int mask,
aeFileProc *proc, void *clientData)
{
if (fd >= eventLoop->setsize) {
errno = ERANGE;
return AE_ERR;
}
if (fd >= eventLoop->setsize) return AE_ERR;
// 取出文件事件結構
aeFileEvent *fe = &eventLoop->events[fd];
// 監(jiān)聽指定 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;
// 如果有需要，更新事件處理器的最大 fd
if (fd > eventLoop->maxfd)
eventLoop->maxfd = fd;
return AE_OK;
}
aeCreateFileEvent 函數中有一個方法調用：aeApiAddEvent，代碼如下
/*
* ae_epoll.c
* 關聯(lián)給定事件到 fd
*/
static int aeApiAddEvent(aeEventLoop *eventLoop, int fd, int mask) {
aeApiState *state = eventLoop->apidata;
struct epoll_event ee;
/* If the fd was already monitored for some event, we need a MOD
* operation. Otherwise we need an ADD operation.
*
* 如果 fd 沒有關聯(lián)任何事件，那么這是一個 ADD 操作。
*
* 如果已經關聯(lián)了某個/某些事件，那么這是一個 MOD 操作。
*/
int op = eventLoop->events[fd].mask == AE_NONE ?
EPOLL_CTL_ADD : EPOLL_CTL_MOD;
// 注冊事件到 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.u64 = 0; /* avoid valgrind warning */
ee.data.fd = fd;
if (epoll_ctl(state->epfd,op,fd,&ee) == -1) return -1;
return 0;
}
這里實際上就是調用系統(tǒng)方法“epoll_ctl”，將事件（文件描述符）注冊進 epoll 中。首先要封裝一個 epoll_event 結構，即 ee ，通過“epoll_ctl”將其注冊進 epoll 中。
除此之外，aeCreateFileEvent 還完成了下面兩個重要操作：
- 將事件函數“acceptTcpHandler”存入了eventLoop中，即由eventLoop->events[fd]->rfileProc 來引用（也可能是wfileProc，分別代表讀事件和寫事件）；
- 將當操作碼添加進 eventLoop->events[fd]->mask 中（mask 類似于JavaNIO中的ops操作碼，代表事件類型）。
事件監(jiān)聽與執(zhí)行
redis.c 的main函數會調用 ae.c 中的 main 方法，如下所示：
/*
* 事件處理器的主循環(huán)
*/
void aeMain(aeEventLoop *eventLoop) {
eventLoop->stop = 0;
while (!eventLoop->stop) {
// 如果有需要在事件處理前執(zhí)行的函數，那么運行它
if (eventLoop->beforesleep != NULL)
eventLoop->beforesleep(eventLoop);
// 開始處理事件
aeProcessEvents(eventLoop, AE_ALL_EVENTS);
}
}
上述代碼會調用 aeProcessEvents 方法用于處理事件，方法如下所示
/* Process every pending time event, then every pending file event
* (that may be registered by time event callbacks just 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;
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))
shortest = aeSearchNearestTimer(eventLoop);
if (shortest) {
// 如果時間事件存在的話
// 那么根據最近可執(zhí)行時間事件和現(xiàn)在時間的時間差來決定文件事件的阻塞時間
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 ，說明事件已經可以執(zhí)行了，將秒和毫秒設為 0 （不阻塞）
if (tvp->tv_sec < 0) tvp->tv_sec = 0;
if (tvp->tv_usec < 0) tvp->tv_usec = 0;
} else {
// 執(zhí)行到這一步，說明沒有時間事件
// 那么根據 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 確保讀/寫事件只能執(zhí)行其中一個
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 */
// 執(zhí)行時間事件
if (flags & AE_TIME_EVENTS)
processed += processTimeEvents(eventLoop);
return processed;
}
該函數中代碼大致分為三個主要步驟
- 根據時間事件與當前時間的關系，決定阻塞時間 tvp;
- 調用aeApiPoll方法，將就緒事件都寫入eventLoop->fired[]中，返回就緒事件數目；
- 遍歷eventLoop->fired[]，遍歷每一個就緒事件，執(zhí)行之前綁定好的方法rfileProc 或者wfileProc。
ae_epoll.c 中的 aeApiPoll 方法如下所示：
/*
* 獲取可執(zhí)行事件
*/
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;
}
執(zhí)行epoll_wait后，就緒的事件會被寫入 eventLoop->apidata->events 事件槽。后面的循環(huán)就是將事件槽中的事件寫入到 eventLoop->fired[] 中。具體描述：每一個事件都是一個 epoll_event 結構，用e來指代，則e.data.fd代表文件描述符，e->events表示其操作碼，將操作碼轉化為mask，最后將fd 和 mask 都寫入eventLoop->fired[j]中。
之后，在外層的 aeProcessEvents 方法中會執(zhí)行函數指針 rfileProc 或者 wfileProc 指向的方法，例如前文提到已注冊的“acceptTcpHandler”。
總結
Redis的網絡模塊其實是一個簡易的Reactor模式。本文順著“服務端注冊事件——>接受客戶端連接——>監(jiān)聽事件是否就緒——>執(zhí)行事件”這樣的路線，來分析Redis源碼，描述了Redis接受客戶端connect的過程。實際上NIO的思想都基本類似。
到此這篇關于Redis網絡模型的源碼詳析的文章就介紹到這了,更多相關Redis網絡模型源碼內容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持腳本之家！
相關文章
Redis，英文全稱是Remote Dictionary Server(遠程字典服務)，是一個開源的使用ANSI C語言編寫、支持網絡、可基于內存亦可持久化的日志型、Key-Value數據庫,需要的朋友可以參考下2023-07-07
springboot +redis 實現(xiàn)點贊、瀏覽、收藏、評論等數量的增減操作
這篇文章主要介紹了springboot +redis 實現(xiàn)點贊、瀏覽、收藏、評論等數量的增減操作，本文通過實例代碼給大家介紹的非常詳細，對大家的學習或工作具有一定的參考借鑒價值,需要的朋友可以參考下2020-09-09
這篇文章主要為大家介紹了Redisson 主從一致性問題詳解，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進步，早日升職加薪2022-08-08
Redis是通過Key-Value的形式來組織數據的,而Key的類型都是String,而Value的類型可以有很多,在Redis中最通用的數據類型大致有這幾種：String、List、Set、Hash、Sorted Set,下面通過本文介紹Redis數據類型和內部編碼,感興趣的朋友一起看看吧2024-04-04
本文主要介紹了Redis的六種底層數據結構，文中通過示例代碼介紹的非常詳細，具有一定的參考價值，感興趣的小伙伴們可以參考一下2022-01-01
這篇文章主要介紹了redis2.8配置文件中文翻譯版,本文翻譯了配置文件中的參數說明,非常詳細,需要的朋友可以參考下2015-06-06
這篇文章主要介紹了Redis安裝及基本數據類型,小編覺得挺不錯的，現(xiàn)在分享給大家，也給大家做個參考。一起跟隨小編過來看看吧2018-05-05
這篇文章主要介紹了Redis 事務與過期時間詳細介紹的相關資料,需要的朋友可以參考下2017-05-05
Redis是一款內存高速緩存數據庫，可以滿足我們對海量數據的讀寫需求，本文重點給大家介紹Windows下Redis?x64的安裝與使用教程，感興趣的朋友一起看看吧2022-03-03
關于SpringBoot 使用 Redis 分布式鎖解決并發(fā)問題
針對上面問題，一般的解決方案是使用分布式鎖來解決，本文通過場景分析給大家介紹關于SpringBoot 使用 Redis 分布式鎖解決并發(fā)問題，感興趣的朋友一起看看吧2021-11-11