快捷導(dǎo)航

Redis定時(shí)任務(wù)原理的實(shí)現(xiàn)

更新時(shí)間：2022年03月07日 09:37:04 作者：心城以北

本文主要是基于?redis?6.2?源碼進(jìn)行分析定時(shí)事件的數(shù)據(jù)結(jié)構(gòu)和常見操作，文中通過示例代碼介紹的非常詳細(xì)，具有一定的參考價(jià)值，感興趣的小伙伴們可以參考一下

數(shù)據(jù)結(jié)構(gòu)

在 redis 中通過 aeTimeEvent 結(jié)構(gòu)來創(chuàng)建定時(shí)任務(wù)事件，代碼如下：

/* Time event structure */
typedef struct aeTimeEvent {
? ? // 標(biāo)識(shí)符
? ? long long id; /* time event identifier. */
? ? // 定時(shí)納秒數(shù)
? ? monotime when;
? ? // 定時(shí)回調(diào)函數(shù)
? ? aeTimeProc *timeProc;
? ? // 注銷定時(shí)器時(shí)候的回調(diào)函數(shù)
? ? 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;

常見操作

1. 創(chuàng)建定時(shí)事件

redis 中最重要的定時(shí)函數(shù)且是周期執(zhí)行的函數(shù)，使用的是 serverCron 函數(shù)。在 redis 中由于定時(shí)任務(wù)比較少，因此并沒有嚴(yán)格的按照過期時(shí)間來排序的，而是按照 id自增 + 頭插法來保證基本有序。

if (aeCreateTimeEvent(server.el, 1, serverCron, NULL, NULL) == AE_ERR) {
? serverPanic("Can't create event loop timers.");
? exit(1);
}

//創(chuàng)建定時(shí)器對(duì)象
long long aeCreateTimeEvent(aeEventLoop *eventLoop, long long milliseconds,
? ? ? ? aeTimeProc *proc, void *clientData,
? ? ? ? aeEventFinalizerProc *finalizerProc)
{
? ? long long id = eventLoop->timeEventNextId++;
? ? aeTimeEvent *te;

? ? te = zmalloc(sizeof(*te));
? ? if (te == NULL) return AE_ERR;
? ? te->id = id;
? ? te->when = getMonotonicUs() + milliseconds * 1000;
? ? te->timeProc = proc;
? ? te->finalizerProc = finalizerProc;
? ? te->clientData = clientData;
? ? te->prev = NULL;
? ? // 頭插法?
? ? te->next = eventLoop->timeEventHead;
? ? te->refcount = 0;
? ? if (te->next)
? ? ? ? te->next->prev = te;
? ? eventLoop->timeEventHead = te;
? ? return id;
}

2. 觸發(fā)定時(shí)事件

redis 中是采用 IO 復(fù)用來進(jìn)行定時(shí)任務(wù)的。

查找距離現(xiàn)在最近的定時(shí)事件，見 usUntilEarliestTimer

?
/* How many microseconds until the first timer should fire.
?* If there are no timers, -1 is returned.
?*
?* Note that's O(N) since time events are unsorted.
?* Possible optimizations (not needed by Redis so far, but...):
?* 1) Insert the event in order, so that the nearest is just the head.
?* ? ?Much better but still insertion or deletion of timers is O(N).
?* 2) Use a skiplist to have this operation as O(1) and insertion as O(log(N)).
?*/
static int64_t usUntilEarliestTimer(aeEventLoop *eventLoop) {
? ? aeTimeEvent *te = eventLoop->timeEventHead;
? ? if (te == NULL) return -1;

? ? aeTimeEvent *earliest = NULL;
? ? while (te) {
? ? ? ? if (!earliest || te->when < earliest->when)
? ? ? ? ? ? earliest = te;
? ? ? ? te = te->next;
? ? }

? ? monotime now = getMonotonicUs();
? ? return (now >= earliest->when) ? 0 : earliest->when - now;
}

?

這里時(shí)間復(fù)雜度可能比較高，實(shí)際中需要結(jié)合具體場(chǎng)景使用。

更新剩余過期時(shí)間，想想為啥呢？因?yàn)槲覀兦懊嫣岬竭^，IO 復(fù)用有可能因?yàn)?IO 事件返回，所以需要更新。

if (flags & AE_TIME_EVENTS && !(flags & AE_DONT_WAIT))
? usUntilTimer = usUntilEarliestTimer(eventLoop);

if (usUntilTimer >= 0) {
? tv.tv_sec = usUntilTimer / 1000000;
? tv.tv_usec = usUntilTimer % 1000000;
? tvp = &tv;
} else {
? if (flags & AE_DONT_WAIT) {
? ? // 不等待
? ? tv.tv_sec = tv.tv_usec = 0;
? ? tvp = &tv;
? } else {
? ? /* Otherwise we can block */
? ? tvp = NULL; /* wait forever */
? }
}

3. 執(zhí)行定時(shí)事件

一次性的執(zhí)行完直接刪除，周期性的執(zhí)行完在重新添加到鏈表。

/* Process time events */
static int processTimeEvents(aeEventLoop *eventLoop) {
? int processed = 0;
? aeTimeEvent *te;
? long long maxId;

? te = eventLoop->timeEventHead;
? maxId = eventLoop->timeEventNextId-1;
? monotime now = getMonotonicUs();
??
? // 刪除定時(shí)器
? while(te) {
? ? long long id;
?? ??? ?
? ? // 下一輪中對(duì)事件進(jìn)行刪除
? ? /* 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);
? ? ? ? now = getMonotonicUs();
? ? ? }
? ? ? zfree(te);
? ? ? te = next;
? ? ? continue;
? ? }
? ??
? ? if (te->id > maxId) {
? ? ? te = te->next;
? ? ? continue;
? ? }

? ? if (te->when <= now) {
? ? ? int retval;

? ? ? id = te->id;
? ? ? te->refcount++;
? ? ? // timeProc 函數(shù)返回值 retVal 為時(shí)間事件執(zhí)行的間隔
? ? ? retval = te->timeProc(eventLoop, id, te->clientData);
? ? ? te->refcount--;
? ? ? processed++;
? ? ? now = getMonotonicUs();
? ? ? if (retval != AE_NOMORE) {
? ? ? ? te->when = now + retval * 1000;
? ? ? } else {
? ? ? ? // 如果超時(shí)了，那么標(biāo)記為刪除
? ? ? ? te->id = AE_DELETED_EVENT_ID;
? ? ? }
? ? }
? ? // 執(zhí)行下一個(gè)
? ? te = te->next;
? }
? return processed;
}