（十八）--- db.c內存數據庫操作 · Redis源碼分析

? ? ? ?我們知道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)); } } ~~~ 用來判斷是否過期的時候用。