redis4.0的时间事件与expire

redis4.0的时间事件与expire

简介

时间事件和文件事件有着相似的接口，他们都在aeProcessEvents中被调用。不同的是文件事件底层委托给 select,epoll等多路复用接口。而时间事件通过每个tick检查时间事件的触发时间是否已经到期。redis4.0版本中只注册了一个时间事件serverCron，它在initServer中注册，在每次aeProcessEvents函数末尾被调用。上文已经提到aeMain函数是redis的事件主循环，它会不断地调用aeProcessEvents。

expire指令在server->expires字典dict中插入sds内部数据类型的key值和到期时间，并触发键空间事件。在serverCron中的databaseCron函数中调用activeExpireCycle(ACTIVE_EXPIRE_CYCLE_SLOW)随机抽取expires中的键值，如果过期，则在server->dict中删除对应的键值。

正文

时间事件注册

首先我们观察一下时间事件的结构体，虽然结构体中有许多成员，但可以说实际用到的就when_sec ，when_ms，timeProc3个成员还有timeProc的返回值。我们可以观察到aeTimeProc会返回一个int类型的值，如果不为-1,会作为下次调用的间隔时间。

/* Time event structure */

typedef struct aeTimeEvent {

    long long id; /* time event identifier. */

    long when_sec; /* seconds */

    long when_ms; /* milliseconds */

    aeTimeProc *timeProc;

    aeEventFinalizerProc *finalizerProc;

    void *clientData;

    struct aeTimeEvent *prev;

    struct aeTimeEvent *next;

} aeTimeEvent;

typedef int aeTimeProc(struct aeEventLoop *eventLoop, long long id, void *clientData);

注册的函数位于initServer中，aeCreateTimeEvent函数会生成一个aeTimeEvent对象，并将其赋值给eventLoop->timeEventHead。

    if (aeCreateTimeEvent(server.el, 1, serverCron, NULL, NULL) == AE_ERR) {

        serverPanic("Can't create event loop timers.");

        exit(1);

    }

时间事件触发

真正处理时间事件的函数是processTimeEvents，但我们回到aeProcessEvents中学习redis中的一个小技巧。aeSearchNearestTimer会找到距离最近的时间事件。如果有（正常情况下肯定会有一个serverCron函数），那么会将距离下一次时间事件的间隔事件写入tvp参数，在aeApiPoll参数中会传入tvp，如果一直没有文件事件触发，那么aeApiPoll函数会等待恰当的时间返回，函数返回后刚好可以处理时间事件。

int aeProcessEvents(aeEventLoop *eventLoop, int flags)

{

    int processed = 0, numevents;

    int j;

    aeTimeEvent *shortest = NULL;

    struct timeval tv, *tvp;

    shortest = aeSearchNearestTimer(eventLoop);

    if (shortest) {

        long now_sec, now_ms;

        aeGetTime(&now_sec, &now_ms);

        tvp = &tv;

        /* How many milliseconds we need to wait for the next

             * time event to fire? */

        long long ms =

            (shortest->when_sec - now_sec)*1000 +

            shortest->when_ms - now_ms;

        if (ms > 0) {

            tvp->tv_sec = ms/1000;

            tvp->tv_usec = (ms % 1000)*1000;

        } else {

            tvp->tv_sec = 0;

            tvp->tv_usec = 0;

        }

    }

    numevents = aeApiPoll(eventLoop, tvp);

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

        //process file events

    }

    /* Check time events */

    if (flags & AE_TIME_EVENTS)

        processed += processTimeEvents(eventLoop);

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

}

在processTimeEvents函数中，会遍历之前注册的函数，如果时间条件满足，则会调用对应的函数。如果函数返回的值不是-1，意味着函数将会利用返回值作为下一次调用函数的间隔时间。ServerCron的频率定义在server.hz，表示一秒钟调用几次serverCron函数，默认是10次/秒。

expire命令

在了解expire命令之前，我们先回顾一下前文的内容，在文件事件处理过程中，redis会将querybuf中的内容转化为client->argc和client->argv，方式是通过createStringObject转化为对应的字符串类型的对象，因此，argv中redisObject的编码类型只可能是embstr或者是raw。

#define OBJ_ENCODING_EMBSTR_SIZE_LIMIT 44

robj *createStringObject(const char *ptr, size_t len) {

    if (len <= OBJ_ENCODING_EMBSTR_SIZE_LIMIT)

        return createEmbeddedStringObject(ptr,len);

    else

        return createRawStringObject(ptr,len);

}

如果传递的time to live参数是负数，那么exipre指令会被转化为del指令，直接删除对应的键值。

否则在server->expire内部数据类型dict中添加对应的到期时间。

void expireGenericCommand(client *c, long long basetime, int unit) {

    robj *key = c->argv[1], *param = c->argv[2];

    long long when; /* unix time in milliseconds when the key will expire. */

    if (getLongLongFromObjectOrReply(c, param, &when, NULL) != C_OK)

        return;

    if (unit == UNIT_SECONDS) when *= 1000;

    when += basetime;

    /* No key, return zero. */

    if (lookupKeyWrite(c->db,key) == NULL) {

        addReply(c,shared.czero);

        return;

    }

    if (when <= mstime() && !server.loading && !server.masterhost) {

        robj *aux;

        int deleted = dbSyncDelete(c->db,key);

        serverAssertWithInfo(c,key,deleted);

        server.dirty++;

        aux =  shared.del;

        rewriteClientCommandVector(c,2,aux,key);

        signalModifiedKey(c->db,key);

        notifyKeyspaceEvent(NOTIFY_GENERIC,"del",key,c->db->id);

        addReply(c, shared.cone);

        return;

    } else {

        setExpire(c,c->db,key,when);

        addReply(c,shared.cone);

        signalModifiedKey(c->db,key);

        notifyKeyspaceEvent(NOTIFY_GENERIC,"expire",key,c->db->id);

        server.dirty++;

        return;

    }

}

删除过期键值

删除过期键值的方式有3种：定时删除，定期删除，被动删除。redis结合使用了定期删除和被动删除。

被动删除

在客户端向服务端发送get ，expire等请求时，会调用expireIfNeeded(c->db,c->argv[j]);函数删除过期的键值。令人好奇的是del请求也会调用expireIfNeeded，也就是有可能调用2次dbSyncDelete函数。

主动删除/定期删除

在前文提到的时间事件serverCron函数中，如果不是从库并且开启了active_expire_enabled（默认开启），则会调用activeExpireCycle函数主动清理过期的键值。

默认情况下，CRON_DBS_PER_CALL的值为16，也是dbnum的值，意味着activeExpireCycle一次会处理16个数据库。而且如果上次调用超时，也会按照一次处理dbnum的数据库处理。

并且对每个数据库至少会进行一轮处理，一轮处理中抽取20个样本，如果样本过期，则删除该键。而且如果样本中过期的键超过25%并且没有超时，则会继续迭代，再进行一轮处理。

timelimit的单位是微秒，如果对当前db处理的过程中超时，那么处理之后的db只进行一轮处理。

所以定期删除并不会将所有的过期键删除，在服务器正常运行的情况下，过期键会维持在25%以内。

void activeExpireCycle(int type) {

	//静态全局变量

    static unsigned int current_db = 0; /* Last DB tested. */

    static int timelimit_exit = 0;      /* Time limit hit in previous call? */

    int j, iteration = 0;

    int dbs_per_call = CRON_DBS_PER_CALL;

    long long start = ustime(), timelimit;

    //一次处理多少db

    if (dbs_per_call > server.dbnum || timelimit_exit)

        dbs_per_call = server.dbnum;

	//时间限制，如果总的时间超过限制，则只处理一轮当前的db

    timelimit = 1000000*ACTIVE_EXPIRE_CYCLE_SLOW_TIME_PERC/server.hz/100;

    timelimit_exit = 0;

    if (timelimit <= 0) timelimit = 1;

    if (type == ACTIVE_EXPIRE_CYCLE_FAST)

        timelimit = ACTIVE_EXPIRE_CYCLE_FAST_DURATION; /* in microseconds. */

    for (j = 0; j < dbs_per_call && timelimit_exit == 0; j++) {

        int expired;

        redisDb *db = server.db+(current_db % server.dbnum);

        current_db++;

        do {

            unsigned long num, slots;

            long long now, ttl_sum;

            int ttl_samples;

            iteration++;

            if ((num = dictSize(db->expires)) == 0) {

                db->avg_ttl = 0;

                break;

            }

            slots = dictSlots(db->expires);

            now = mstime();

            expired = 0;

            if (num > ACTIVE_EXPIRE_CYCLE_LOOKUPS_PER_LOOP)

                num = ACTIVE_EXPIRE_CYCLE_LOOKUPS_PER_LOOP;

            while (num--) {

                dictEntry *de;

                if ((de = dictGetRandomKey(db->expires)) == NULL) break;

                if (activeExpireCycleTryExpire(db,de,now)) expired++;

            }

            total_ += expired;

            if ((iteration & 0xf) == 0) { /* check once every 16 iterations. */

                elapsed = ustime()-start;

                if (elapsed > timelimit) {

                    timelimit_exit = 1;

                    server.stat_expired_time_cap_reached_count++;

                    break;

                }

            }

        } while (expired > ACTIVE_EXPIRE_CYCLE_LOOKUPS_PER_LOOP/4);

    }

}

参考文献

redis 文档

《Redis设计与实现》

巴特西

自顶向下redis4.0（4）时间事件与expire