Redis 持久化与故障恢复之rdb

一、摘要

老生常谈一下吧，redis持久化分为rdb和aof两种模式，本篇先说一说rdb模式吧，共分为三部分：1：如何触发rdb持久化， 2：rdb持久化源码， 3：rdb文件解析。

ps:本文基于redis7分析。

二、如何触发rdb持久化

1：通过save关键字在redis.conf文件配置触发条件

# save <seconds> <changes> [<seconds> <changes> ...]

save 3600 1 600 100 60 3000

上述配置表示如果满足每隔3600s内有1个key发生变化，每隔600s内有100个key发生变化，每隔60s内有3000个key发生变化三个条件中的一个，就会触发rdb持久化。

ps:触发后执行过程与bgsave命令一样

2：在cli执行save或bgsave命令

save表示同步执行rdb持久化，会阻塞其它客户端命令的响应；

bgsve表示异步处理rdb持久化，不会阻塞。

3：bgsave执行流程如下图：

可以概括为：触发rdb持久化后，redis主进程会fork一个子进程出来，子进程会将内存数据dump到临时的rdb快照文件中，在完成rdb快照文件的生成之后，就替换（通过rename系统函数完成替换）之前的旧的快照文件dump.rdb，每次生成一个新的快照，都会覆盖之前的老快照。

三、rdb持久化源码

代码核心逻辑在rdb.c文件中，其中核心函数是rdbSaveBackground。

通过对rdb持久化触发方式的分析，可知有两种代码路径进入rdbSaveBackground函数。

1：redis.conf 的save配置

//server.c
int main(int argc, char **argv) {
```initServer();
``
}
void initServer(void) {
```/* Create the timer callback, this is our way to process many background* operations incrementally, like clients timeout, eviction of unaccessed* expired keys and so forth. *///将定时任务添加到reactor的时间事件中去，1s一次if (aeCreateTimeEvent(server.el, 1, serverCron, NULL, NULL) == AE_ERR) {serverPanic("Can't create event loop timers.");exit(1);}
```
}
int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
```
/* Check if a background saving or AOF rewrite in progress terminated. */if (hasActiveChildProcess() || ldbPendingChildren()){//如果有正在处理的rdb持久化或aof持久化，则不执行，仅仅检查run_with_period(1000) receiveChildInfo();checkChildrenDone();} else {/* If there is not a background saving/rewrite in progress check if* we have to save/rewrite now. */for (j = 0; j < server.saveparamslen; j++) {//这里的server.saveparams就是save关键字的配置，满足其中一个就执行rdbSaveBackgroundstruct saveparam *sp = server.saveparams+j;/* Save if we reached the given amount of changes,* the given amount of seconds, and if the latest bgsave was* successful or if, in case of an error, at least* CONFIG_BGSAVE_RETRY_DELAY seconds already elapsed. */if (server.dirty >= sp->changes &&server.unixtime-server.lastsave > sp->seconds &&(server.unixtime-server.lastbgsave_try >CONFIG_BGSAVE_RETRY_DELAY ||server.lastbgsave_status == C_OK)){serverLog(LL_NOTICE,"%d changes in %d seconds. Saving...",sp->changes, (int)sp->seconds);rdbSaveInfo rsi, *rsiptr;rsiptr = rdbPopulateSaveInfo(&rsi);rdbSaveBackground(SLAVE_REQ_NONE,server.rdb_filename,rsiptr,RDBFLAGS_NONE);break;}}
```
}

2：bgsave 和save

redis收到bgsave的命令，会执行bgsaveCommand函数;

redis收到save的命令，会执行saveCommand函数。

void saveCommand(client *c) {if (server.child_type == CHILD_TYPE_RDB) {addReplyError(c,"Background save already in progress");return;}server.stat_rdb_saves++;rdbSaveInfo rsi, *rsiptr;rsiptr = rdbPopulateSaveInfo(&rsi);//执行保存内存数据到rdb文件if (rdbSave(SLAVE_REQ_NONE,server.rdb_filename,rsiptr,RDBFLAGS_NONE) == C_OK) {addReply(c,shared.ok);} else {addReplyErrorObject(c,shared.err);}
}/* BGSAVE [SCHEDULE] */
void bgsaveCommand(client *c) {```if (server.child_type == CHILD_TYPE_RDB) {addReplyError(c,"Background save already in progress");} else if (hasActiveChildProcess() || server.in_exec) {//有活跃的子进程就不会执行rdbSaveBackgroundif (schedule || server.in_exec) {server.rdb_bgsave_scheduled = 1;addReplyStatus(c,"Background saving scheduled");} else {addReplyError(c,"Another child process is active (AOF?): can't BGSAVE right now. ""Use BGSAVE SCHEDULE in order to schedule a BGSAVE whenever ""possible.");}} else if (rdbSaveBackground(SLAVE_REQ_NONE,server.rdb_filename,rsiptr,RDBFLAGS_NONE) == C_OK) {addReplyStatus(c,"Background saving started");} else {addReplyErrorObject(c,shared.err);}
}
int rdbSaveBackground(int req, char *filename, rdbSaveInfo *rsi, int rdbflags) {
```if ((childpid = redisFork(CHILD_TYPE_RDB)) == 0) {int retval;/* Child */redisSetProcTitle("redis-rdb-bgsave");redisSetCpuAffinity(server.bgsave_cpulist);//执行保存内存数据到rdb文件retval = rdbSave(req, filename,rsi,rdbflags);if (retval == C_OK) {//如果重新生成rdb文件成功，则通知主进程sendChildCowInfo(CHILD_INFO_TYPE_RDB_COW_SIZE, "RDB");}exitFromChild((retval == C_OK) ? 0 : 1);}
```
}

3：概括代码函数调用过程

4：rdbsave

/* Save the DB on disk. Return C_ERR on error, C_OK on success. */
int rdbSave(int req, char *filename, rdbSaveInfo *rsi, int rdbflags) {
```//创建临时rdb文件snprintf(tmpfile,256,"temp-%d.rdb", (int) getpid());fp = fopen(tmpfile,"w");//将内存数据放入临时rdb文件if (rdbSaveRio(req,&rdb,&error,rdbflags,rsi) == C_ERR) {errno = error;err_op = "rdbSaveRio";goto werr;}/* Make sure data will not remain on the OS's output buffers */if (fflush(fp)) { err_op = "fflush"; goto werr; }if (fsync(fileno(fp))) { err_op = "fsync"; goto werr; }if (fclose(fp)) { fp = NULL; err_op = "fclose"; goto werr; }//重命名临时文件为正式rdb文件if (rename(tmpfile,filename) == -1) {...}```
}
int rdbSaveRio(int req, rio *rdb, int *error, int rdbflags, rdbSaveInfo *rsi) {
```//定义rdb文件中check_sum部分的生成函数if (server.rdb_checksum)rdb->update_cksum = rioGenericUpdateChecksum;//定义rdb文件开头部分，REDIS+db_versionsnprintf(magic,sizeof(magic),"REDIS%04d",RDB_VERSION);//保存每个db的数据/* save all databases, skip this if we're in functions-only mode */if (!(req & SLAVE_REQ_RDB_EXCLUDE_DATA)) {for (j = 0; j < server.dbnum; j++) {if (rdbSaveDb(rdb, j, rdbflags, &key_counter) == -1) goto werr;}}//生成check_sum，并追加到rdb文件最后cksum = rdb->cksum;memrev64ifbe(&cksum);if (rioWrite(rdb,&cksum,8) == 0) goto werr;
```
}
//保存每个db的数据
ssize_t rdbSaveDb(rio *rdb, int dbid, int rdbflags, long *key_counter) {
```/* Write the SELECT DB opcode *///写入rdb文件SELECTDB标志位，表示这里开始要进入某个db了if ((res = rdbSaveType(rdb,RDB_OPCODE_SELECTDB)) < 0) goto werr;written += res;//写入rdb文件当前db的索引号，表示这里开始的数据是某个db的数据if ((res = rdbSaveLen(rdb, dbid)) < 0) goto werr;written += res;//写key val /* Iterate this DB writing every entry */while((de = dictNext(di)) != NULL) {sds keystr = dictGetKey(de);robj key, *o = dictGetVal(de);long long expire;size_t rdb_bytes_before_key = rdb->processed_bytes;expire = getExpire(db,&key);if ((res = rdbSaveKeyValuePair(rdb, &key, o, expire, dbid)) < 0) goto werr;}
```
}
int rdbSaveKeyValuePair(rio *rdb, robj *key, robj *val, long long expiretime, int dbid) {/* Save the expire time */if (expiretime != -1) {//有超时时间的话，保存RDB_OPCODE_EXPIRETIME_MS标志位if (rdbSaveType(rdb,RDB_OPCODE_EXPIRETIME_MS) == -1) return -1;//保存过期时间戳，毫秒if (rdbSaveMillisecondTime(rdb,expiretime) == -1) return -1;}//依次保存数据类型，key val三个内容/* Save type, key, value */if (rdbSaveObjectType(rdb,val) == -1) return -1;if (rdbSaveStringObject(rdb,key) == -1) return -1;if (rdbSaveObject(rdb,val,key,dbid) == -1) return -1; }

ps:注意下rdbSaveRio之后的函数调用，其中可以一窥rdb文件结构。

四、rdb文件结构

1：一个完整RDB文件所包含的各个部分如下图，代码见rdbSaveRio函数

1）RDB文件的最开头是REDIS部分，这个部分的长度为5字节，保存着“REDIS”五个字符。通过这五个字符，程序可以在载入文件时，快速检查所载入的文件是否RDB文件。

2）db_version长度为4字节，它的值是一个字符串表示的整数，这个整数记录了RDB文件的版本号，比如redis7的该值就是11，见rdbSaveRio函数的RDB_VERSION。

3）db_content部分包含着零个或多个数据库，以及各个数据库中的键值对数据。

4）EOF常量的长度为1字节，这个常量标志着RDB文件正文内容的结束，当读入程序遇到这个值的时候，它知道所有数据库的所有键值对都已经载入完毕了。

5）check_sum是一个8字节长的无符号整数，保存着一个校验和，这个校验和是程序通过对REDIS、db_version、databases、EOF四个部分的内容进行计算得出的。服务器在载入RDB文件时，会将载入数据所计算出的校验和与check_sum所记录的校验和进行对比，以此来检查RDB文件是否有出错或者损坏的情况出现。

2：db_content内部结构如下图，代码见rdbSaveDb函数

1）SELECTDB常量的长度为1字节，当读入程序遇到这个值的时候，它知道接下来要读入的将是一个数据库索引号。

2）db_number保存着一个数据库索引号，根据号码的大小不同，这个部分的长度可以是1字节、2字节或者5字节。当程序读入db_number部分之后，服务器会调用SELECT命令，根据读入的数据库号码进行数据库切换，使得之后读入的键值对可以载入到正确的数据库中。

3）key_value_pairs部分保存了数据库中的所有键值对数据，如果键值对带有过期时间，那么过期时间也会和键值对保存在一起。根据键值对的数量、类型、内容以及是否有过期时间等条件的不同， key_value_pairs部分的长度也会有所不同。

3：key_value_pairs结构如下图，代码见rdbSaveKeyValuePair函数

[1] :无过期时间的结构

[2]：有过期时间的结构

1）EXPIRETIME_MS常量的长度为1字节，它告知读入程序，接下来要读入的将是一个以毫秒为单位的过期时间。

2）ms是一个8字节长的带符号整数，记录着一个以毫秒为单位的 UNIX时间戳，这个时间戳就是键值对的过期时间，这样在加载rdb文件时如果ms过期就不加载该值了。

3）TYPE记录了value的类型，长度为1字节，值可以是以下常量的其中一个：

REDIS_RDB_TYPE_STRING
REDIS_RDB_TYPE_LIST
REDIS_RDB_TYPE_SET
REDIS_RDB_TYPE_HASH_ZIPLIST
REDIS_RDB_TYPE_ZSET_ZIPLIST
REDIS_RDB_TYPE_SET_INTSET
REDIS_RDB_TYPE_HASH
REDIS_RDB_TYPE_LIST_ZIPLIST
REDIS_RDB_TYPE_ZSET

这样redis在加载value时就可以根据TYPE去解析了

4） key 总是一个字符串对象，它的编码方式和 REDIS_RDB_TYPE_STRING 类型的 value 一样。根据内容长度的不同， key 的长度也会有所不同。

5）value就是具体的数据内容了，根据 TYPE 类型的不同，以及保存内容长度的不同，保存 value 的结构和长度也会有所不同。