Redis源码分析（十八）--- db.c内存数据库操作

我们知道Redis数据库作为一个内存数据库，与memcached比较类似，基本的操作都是存储在内存缓冲区中，等到缓冲区中数据满后，在持久化到磁盘中。今天，我主要研究了对于redis中对于内存数据库的操作。与普通的数据操作比较，并没有什么特别多的其他的一些操作。下面是我分类出的一些API:

/*-----------------------------------------------------------------------------
 * C-level DB API
 *----------------------------------------------------------------------------*/
robj *lookupKey(redisDb *db, robj *key) /* 从db中获取key代表的值 */
robj *lookupKeyRead(redisDb *db, robj *key) /* 寻找某个key的值，与lookupKey方法的区别是多了过期检查 */
robj *lookupKeyWrite(redisDb *db, robj *key) /* 与lookupKeyRead一样，只是少了命中刷的统计 */
robj *lookupKeyReadOrReply(redisClient *c, robj *key, robj *reply) /* 有回复的读擦操作 */
robj *lookupKeyWriteOrReply(redisClient *c, robj *key, robj *reply) /* 有回复的写操作 */
void dbAdd(redisDb *db, robj *key, robj *val) /* 往内存数据库中添加值，如果key已经存在，则操作无效 */
void dbOverwrite(redisDb *db, robj *key, robj *val) /* db  key value覆盖操作，如果不存在此key,操作失效 */
void setKey(redisDb *db, robj *key, robj *val) /* 高级设置操作，如果不存在的直接添加，存在的就覆盖 */
int dbExists(redisDb *db, robj *key) /* db是否存在此key */
robj *dbRandomKey(redisDb *db) /* 随机返回没有过期的key */
int dbDelete(redisDb *db, robj *key) /* db删除操作 */
robj *dbUnshareStringValue(redisDb *db, robj *key, robj *o) /* 解除key的共享，之后就可以进行修改操作 */
long long emptyDb(void(callback)(void*)) /* 将server中的所有数据库清空，回调函数作为参数传入 */
int selectDb(redisClient *c, int id) /* 客户端选择服务端的某个db */
void signalModifiedKey(redisDb *db, robj *key) /* 每当key被修改时，就会调用此方法，touchWatchedKey(db,key)方法，就把此key对应的客户端锁住了 */
void signalFlushedDb(int dbid) /* 把dbid中的key都touch一遍 */
void flushdbCommand(redisClient *c) /* 刷新client所在的db命令 */
void flushallCommand(redisClient *c) /* 刷新所有的server中的数据库 */
void delCommand(redisClient *c) /* 根据Client的命令参数删除数据库 */
void existsCommand(redisClient *c) /* 某个key是否存在命令 */
void selectCommand(redisClient *c) /* Client客户端选择数据库命令 */
void randomkeyCommand(redisClient *c) /* 获取随机key指令 */
void keysCommand(redisClient *c) /* 向客户端回复key obj命令 */
void scanCallback(void *privdata, const dictEntry *de) /* type scan扫描出key,val */
int parseScanCursorOrReply(redisClient *c, robj *o, unsigned long *cursor) /* 判断scan Cursor是否有效 */
void scanGenericCommand(redisClient *c, robj *o, unsigned long cursor) /* 3: Filter elements.（过滤元素）4: Reply to the client.(回复客户端) */
void scanCommand(redisClient *c) /* 扫描命令 */
void dbsizeCommand(redisClient *c) /* 客户端所用的db的字典总数 */
void lastsaveCommand(redisClient *c) /* 服务端最后一次保存的操作 */
void typeCommand(redisClient *c) /* 客户端查询的key的type类型 */
void shutdownCommand(redisClient *c) /* shutdown终止命令，服务端要做最后的保存操作 */
void renameGenericCommand(redisClient *c, int nx) /*为key重命名操作 */
void renameCommand(redisClient *c) /* 重命名可能会覆盖原值命令 */
void renamenxCommand(redisClient *c) /* 重命名时不覆盖原来的值 */
void moveCommand(redisClient *c) /* 将源db中的key移到目标db上 */
int removeExpire(redisDb *db, robj *key) /* 移除过期的key */
void setExpire(redisDb *db, robj *key, long long when) /* 设置过期的key,操作为将主要的dict中key移入expire的dict中，并对此key设置时间 */
long long getExpire(redisDb *db, robj *key) /*  获取key的过期时间*/
void propagateExpire(redisDb *db, robj *key)
int expireIfNeeded(redisDb *db, robj *key) /* 判断此key是否过期，2个条件，1是否存在expire的key里没有就不过期
2.在expire里面了，判断when时间有没有超过当前时间，没有超过也不算过期 */
void expireGenericCommand(redisClient *c, long long basetime, int unit)
void expireCommand(redisClient *c)
void expireatCommand(redisClient *c)
void pexpireCommand(redisClient *c)
void pexpireatCommand(redisClient *c)
void ttlGenericCommand(redisClient *c, int output_ms) /* 返回key的ttl生存时间 ，下面的一些方法是时间单位的不同,默认为秒*/
void ttlCommand(redisClient *c)
void pttlCommand(redisClient *c)
void persistCommand(redisClient *c) /* key是否存在的命令 */
int *getKeysUsingCommandTable(struct redisCommand *cmd,robj **argv, int argc, int *numkeys)
int *getKeysFromCommand(struct redisCommand *cmd,robj **argv, int argc, int *numkeys, int flags)
void getKeysFreeResult(int *result)
int *noPreloadGetKeys(struct redisCommand *cmd,robj **argv, int argc, int *numkeys, int flags)
int *renameGetKeys(struct redisCommand *cmd,robj **argv, int argc, int *numkeys, int flags)
int *zunionInterGetKeys(struct redisCommand *cmd,robj **argv, int argc, int *numkeys, int flags)

在API的后半部分API都是一些函数封装的一些命令操作。开放给系统调用。在上面的API中，比较典型的就是，read,write等API，

/* 从db中获取key代表的值 */
robj *lookupKey(redisDb *db, robj *key) {
	//从db的dict字典中查找
    dictEntry *de = dictFind(db->dict,key->ptr);
    if (de) {
        robj *val = dictGetVal(de);

        /* Update the access time for the ageing algorithm.
         * Don't do it if we have a saving child, as this will trigger
         * a copy on write madness. */
        if (server.rdb_child_pid == -1 && server.aof_child_pid == -1)
            val->lru = server.lruclock;
        return val;
    } else {
        return NULL;
    }
}

但是真正调用的时候，不会直接调用此方法，会加一些限制，会过滤掉过期的key,还有缓冲区命中数的统计：

/* 寻找某个key的值，与lookupKey方法的区别是多了过期检查 */
robj *lookupKeyRead(redisDb *db, robj *key) {
    robj *val;

    expireIfNeeded(db,key);
    val = lookupKey(db,key);
    if (val == NULL)
    	//命中数减一
        server.stat_keyspace_misses++;
    else
    	//命中数递增1
        server.stat_keyspace_hits++;
    return val;
}

可以有效调整缓冲区。下面给出一个修改内存数据库的操作：

/* High level Set operation. This function can be used in order to set
 * a key, whatever it was existing or not, to a new object.
 *
 * 1) The ref count of the value object is incremented.
 * 2) clients WATCHing for the destination key notified.
 * 3) The expire time of the key is reset (the key is made persistent). */
/* 高级设置操作，如果不存在的直接添加，存在的就覆盖 */
void setKey(redisDb *db, robj *key, robj *val) {
    if (lookupKeyWrite(db,key) == NULL) {
        dbAdd(db,key,val);
    } else {
        dbOverwrite(db,key,val);
    }
    //对此key增加引用计数
    incrRefCount(val);
    removeExpire(db,key);
    signalModifiedKey(db,key);
}

我们看到其实在每次更改数据库操作的时候，都会出现signalModifiedKey(db,key)这个方法，大致意思就是提示要改变key所对应的值了，里面执行的操作到底是什么呢，这个方法的实现就在db.c中:

/*-----------------------------------------------------------------------------
 * Hooks for key space changes.
 *
 * Every time a key in the database is modified the function
 * signalModifiedKey() is called.
 *
 * Every time a DB is flushed the function signalFlushDb() is called.
 *----------------------------------------------------------------------------*/
/* 每当key被修改时，就会调用此方法，touchWatchedKey(db,key)方法，就把此key对应的客户端锁住了 */
void signalModifiedKey(redisDb *db, robj *key) {
    touchWatchedKey(db,key);
}

调用的就是touch -key方法了，就是把监听此key的Client列表进行设置，只能让一个客户端操作执行成功，客户端的其他操作无效，达到同步。当内存数据渐渐满的时候，会定期的刷新到磁盘中:

/* 刷新所有的server中的数据库 */
void flushallCommand(redisClient *c) {
    signalFlushedDb(-1);
    server.dirty += emptyDb(NULL);
    addReply(c,shared.ok);
    if (server.rdb_child_pid != -1) {
        kill(server.rdb_child_pid,SIGUSR1);
        rdbRemoveTempFile(server.rdb_child_pid);
    }
    if (server.saveparamslen > 0) {
        /* Normally rdbSave() will reset dirty, but we don't want this here
         * as otherwise FLUSHALL will not be replicated nor put into the AOF. */
        int saved_dirty = server.dirty;
        //在这里重新保存rdb了
        rdbSave(server.rdb_filename);
        server.dirty = saved_dirty;
    }
    server.dirty++;
}

rdbSave操作时重点。在db.c还提到了一个概念，expire过期的概念，也就是说，存在key过期的概念，在内存数据库中，频繁的操作比如会引起许多过期的键值对的存在，所以在db中，维护了一个db->expires的东西，所有过期的可以都存在于db->expires里面，定期会进行移除操作，所以在最早的那个函数中，往内存数据库取值的时候，要判断是否过期

/* 判断此key是否过期，2个条件，1是否存在expire的key里没有就不过期
	2.在expire里面了，判断when时间有没有超过当前时间，没有超过也不算过期 */
int expireIfNeeded(redisDb *db, robj *key) {
    mstime_t when = getExpire(db,key);
    mstime_t now;

    if (when < 0) return 0; /* No expire for this key */

    /* Don't expire anything while loading. It will be done later. */
    if (server.loading) return 0;

    /* If we are in the context of a Lua script, we claim that time is
     * blocked to when the Lua script started. This way a key can expire
     * only the first time it is accessed and not in the middle of the
     * script execution, making propagation to slaves / AOF consistent.
     * See issue #1525 on Github for more information. */
    now = server.lua_caller ? server.lua_time_start : mstime();

    /* If we are running in the context of a slave, return ASAP:
     * the slave key expiration is controlled by the master that will
     * send us synthesized DEL operations for expired keys.
     *
     * Still we try to return the right information to the caller,
     * that is, 0 if we think the key should be still valid, 1 if
     * we think the key is expired at this time. */
    if (server.masterhost != NULL) return now > when;

    /* Return when this key has not expired */
    if (now <= when) return 0;

    /* Delete the key */
    server.stat_expiredkeys++;
    propagateExpire(db,key);
    notifyKeyspaceEvent(REDIS_NOTIFY_EXPIRED,
        "expired",key,db->id);
    return dbDelete(db,key);
}

每个expire的key有个ttl的概念，就是"Time To Live"生存时间:

/* 返回key的ttl生存时间 */
void ttlGenericCommand(redisClient *c, int output_ms) {
    long long expire, ttl = -1;

    /* If the key does not exist at all, return -2 */
    if (lookupKeyRead(c->db,c->argv[1]) == NULL) {
        addReplyLongLong(c,-2);
        return;
    }
    /* The key exists. Return -1 if it has no expire, or the actual
     * TTL value otherwise. */
    expire = getExpire(c->db,c->argv[1]);
    if (expire != -1) {
    	//如果已被移入过期的key，计算过期时间里当前时间还差多远，ttl就是当前的生存时间单位为ms
        ttl = expire-mstime();
        if (ttl < 0) ttl = 0;
    }
    if (ttl == -1) {
        addReplyLongLong(c,-1);
    } else {
        addReplyLongLong(c,output_ms ? ttl : ((ttl+500)/1000));
    }
}

用来判断是否过期的时候用。