（十八）— db.c内存数据库操作
最后更新于：2022-04-01 20:20:51
我们知道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));
}
}
~~~
用来判断是否过期的时候用。
';