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

rwewqe

       我们知道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));  
        }  
    }  

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