因为Redis是内存数据库，因此将数据存储在内存中，如果一旦服务器进程退出，服务器中的数据库状态就会消失不见，为了解决这个问题，Redis提供了两种持久化的机制：RDB和AOF。本篇主要剖析RDB持久化的过程。
RDB持久化是把当前进程数据生成时间点快照（point-in-time snapshot）保存到硬盘的过程，避免数据意外丢失。
RDB触发机制分为手动触发和自动触发。
手动触发的两条命令：
SAVE：阻塞当前Redis服务器，知道RDB过程完成为止。BGSAVE：Redis 进程执行fork()操作创建出一个子进程，在后台完成RDB持久化的过程。（主流）自动触发的配置：
c save 900 1 //服务器在900秒之内，对数据库执行了至少1次修改 save 300 10 //服务器在300秒之内，对数据库执行了至少10修改 save 60 1000 //服务器在60秒之内，对数据库执行了至少1000修改 // 满足以上三个条件中的任意一个，则自动触发 BGSAVE 操作 // 或者使用命令CONFIG SET 命令配置我们用图来表示 BGSAVE命令 的触发流程，如下图所示：
RDB命令源码如下：Redis 3.2 RDB源码注释
/* BGSAVE [SCHEDULE] */ // BGSAVE 命令实现 void bgsaveCommand(client *c) { int schedule = 0; //SCHEDULE控制BGSAVE的执行，避免和AOF重写进程冲突 /* The SCHEDULE option changes the behavior of BGSAVE when an AOF rewrite * is in progress. Instead of returning an error a BGSAVE gets scheduled. */ if (c->argc > 1) { // 设置schedule标志 if (c->argc == 2 && !strcasecmp(c->argv[1]->ptr,"schedule")) { schedule = 1; } else { addReply(c,shared.syntaxerr); return; } } // 如果正在执行RDB持久化操作，则退出 if (server.rdb_child_pid != -1) { addReplyError(c,"Background save already in progress"); // 如果正在执行AOF持久化操作，需要将BGSAVE提上日程表 } else if (server.aof_child_pid != -1) { // 如果schedule为真，设置rdb_bgsave_scheduled为1，表示将BGSAVE提上日程表 if (schedule) { server.rdb_bgsave_scheduled = 1; addReplyStatus(c,"Background saving scheduled"); } else { //没有设置schedule，则不能立即执行BGSAVE addReplyError(c, "An AOF log rewriting in progress: can't BGSAVE right now. " "Use BGSAVE SCHEDULE in order to schedule a BGSAVE whenver " "possible."); } // 执行BGSAVE } else if (rdbSaveBackground(server.rdb_filename) == C_OK) { addReplyStatus(c,"Background saving started"); } else { addReply(c,shared.err); } }我们后面会重点讲解rdbSaveBackground()函数的工作过程。
RDB的优点：
RDB是一个紧凑压缩的二进制文件，代表Redis在某个时间点上的数据快照。非常适用于备份，全景复制等场景。Redis 加载RDB恢复数据远远快于AOF的方式。RDB的缺点：
RDB没有办法做到实时持久化或秒级持久化。因为BGSAVE每次运行的又要进行fork()的调用创建子进程，这属于重量级操作，频繁执行成本过高，因为虽然Linux支持读时共享，写时拷贝(copy-on-write)的技术，但是仍然会有大量的父进程的空间内存页表，信号控制表，寄存器资源等等的复制。RDB文件使用特定的二进制格式保存，Redis版本演进的过程中，有多个RDB版本，这导致版本兼容的问题。阅读此部分，可以跳过源码，只看文字部分，因为所有过程的依据我都以源码的方式给出，因此篇幅会比较长，但是我都以文字解释，所以可以跳过源码，只读文字，理解RDB的过程。也可以上github查看所有代码的注释：Redis 3.2 源码注释
之前我们给出了 BGSAVE命令 的源码，因此我们就重点剖析 rdbSaveBackground()的工作过程，一层一层的剥开封装。
在RDB持久化之前需要设置一些标识，用来标识服务器当前的状态，定义在server.h/struct redisServer 结构体中，我们列出会用到的一部分，如果需要可以在这里查看。Redis 3.2 源码注释
struct redisServer { // 数据库数组，长度为16 redisDb *db; // 从节点列表和监视器列表 list *slaves, *qiank; /* List of slaves and MONITORs */ /* RDB / AOF loading information ××××××××××××××××××××××××××××××××××××××××××××××××××××××××××*/ // 正在载入状态 int loading; /* We are loading data from disk if true */ // 设置载入的总字节 off_t loading_total_bytes; // 已载入的字节数 off_t loading_loaded_bytes; // 载入的开始时间 time_t loading_start_time; // 在load时，用来设置读或写的最大字节数max_processing_chunk off_t loading_process_events_interval_bytes; // 服务器内存使用的 size_t stat_peak_memory; /* Max used memory record */ // 计算fork()的时间 long long stat_fork_time; /* Time needed to perform latest fork() */ // 计算fork的速率，GB/每秒 double stat_fork_rate; /* Fork rate in GB/sec. */ /* RDB persistence ××××××××××××××××××××××××××××××××××××××××××××××××××××××××××××××××××××××××××××*/ // 脏键，记录数据库被修改的次数 long long dirty; /* Changes to DB from the last save */ // 在BGSAVE之前要备份脏键dirty的值，如果BGSAVE失败会还原 long long dirty_before_bgsave; /* Used to restore dirty on failed BGSAVE */ // 执行BGSAVE的子进程的pid pid_t rdb_child_pid; /* PID of RDB saving child */ // 保存save参数的数组 struct saveparam *saveparams; /* Save points array for RDB */ // 数组长度 int saveparamslen; /* Number of saving points */ // RDB文件的名字，默认为dump.rdb char *rdb_filename; /* Name of RDB file */ // 是否采用LZF压缩算法压缩RDB文件，默认yes int rdb_compression; /* Use compression in RDB? */ // RDB文件是否使用校验和，默认yes int rdb_checksum; /* Use RDB checksum? */ // 上一次执行SAVE成功的时间 time_t lastsave; /* Unix time of last successful save */ // 最近一个尝试执行BGSAVE的时间 time_t lastbgsave_try; /* Unix time of last attempted bgsave */ // 最近执行BGSAVE的时间 time_t rdb_save_time_last; /* Time used by last RDB save run. */ // BGSAVE开始的时间 time_t rdb_save_time_start; /* Current RDB save start time. */ // 当rdb_bgsave_scheduled为真时，才能开始BGSAVE int rdb_bgsave_scheduled; /* BGSAVE when possible if true. */ // rdb执行的类型，是写入磁盘，还是写入从节点的socket int rdb_child_type; /* Type of save by active child. */ // BGSAVE执行完的状态 int lastbgsave_status; /* C_OK or C_ERR */ // 如果不能执行BGSAVE则不能写 int stop_writes_on_bgsave_err; /* Don't allow writes if can't BGSAVE */ // 无磁盘同步，管道的写端 int rdb_pipe_write_result_to_parent; /* RDB pipes used to return the state */ // 无磁盘同步，管道的读端 int rdb_pipe_read_result_from_child; /* of each slave in diskless SYNC. */ /* time cache ××××××××××××××××××××××××××××××××××××××××××××××××××××××××××××××××××××××××××××*/ // 保存秒单位的Unix时间戳的缓存 time_t unixtime; /* Unix time sampled every cron cycle. */ // 保存毫秒单位的Unix时间戳的缓存 long long mstime; /* Like 'unixtime' but with milliseconds resolution. */ /* Latency monitor ××××××××××××××××××××××××××××××××××××××××××××××××××××××××××××××××××××××××××××*/ // 延迟的阀值 long long latency_monitor_threshold; // 延迟与造成延迟的事件关联的字典 dict *latency_events; };然后我们直接给rdbSaveBackground()函数出源码：
在这里，就可以看见fork()函数的执行，在子进程中执行了rdbSave()函数，父进程则执行了一些设置状态的操作。
// 后台进行RDB持久化BGSAVE操作 int rdbSaveBackground(char *filename) { pid_t childpid; long long start; // 当前没有正在进行AOF和RDB操作，否则返回C_ERR if (server.aof_child_pid != -1 || server.rdb_child_pid != -1) return C_ERR; // 备份当前数据库的脏键值 server.dirty_before_bgsave = server.dirty; // 最近一个执行BGSAVE的时间 server.lastbgsave_try = time(NULL); // fork函数开始时间，记录fork函数的耗时 start = ustime(); // 创建子进程 if ((childpid = fork()) == 0) { int retval; // 子进程执行的代码 /* Child */ // 关闭监听的套接字 closeListeningSockets(0); // 设置进程标题，方便识别 redisSetProcTitle("redis-rdb-bgsave"); // 执行保存操作，将数据库的写到filename文件中 retval = rdbSave(filename); if (retval == C_OK) { // 得到子进程进程的脏私有虚拟页面大小，如果做RDB的同时父进程正在写入的数据，那么子进程就会拷贝一个份父进程的内存，而不是和父进程共享一份内存。 size_t private_dirty = zmalloc_get_private_dirty(); // 将子进程分配的内容写日志 if (private_dirty) { serverLog(LL_NOTICE, "RDB: %zu MB of memory used by copy-on-write", private_dirty/(1024*1024)); } } // 子进程退出，发送信号给父进程，发送0表示BGSAVE成功，1表示失败 exitFromChild((retval == C_OK) ? 0 : 1); } else { // 父进程执行的代码 /* Parent */ // 计算出fork的执行时间 server.stat_fork_time = ustime()-start; // 计算fork的速率，GB/每秒 server.stat_fork_rate = (double) zmalloc_used_memory() * 1000000 / server.stat_fork_time / (1024*1024*1024); /* GB per second. */ //如果fork执行时长，超过设置的阀值，则要将其加入到一个字典中，与传入"fork"关联，以便进行延迟诊断 latencyAddSampleIfNeeded("fork",server.stat_fork_time/1000); // 如果fork出错 if (childpid == -1) { server.lastbgsave_status = C_ERR; //设置BGSAVE错误 // 更新日志信息 serverLog(LL_WARNING,"Can't save in background: fork: %s", strerror(errno)); return C_ERR; } // 更新日志信息 serverLog(LL_NOTICE,"Background saving started by pid %d",childpid); server.rdb_save_time_start = time(NULL); //设置BGSAVE开始的时间 server.rdb_child_pid = childpid; //设置负责执行BGSAVE操作的子进程id server.rdb_child_type = RDB_CHILD_TYPE_DISK;//设置BGSAVE的类型，往磁盘中写入 //关闭哈希表的resize，因为resize过程中会有复制拷贝动作 updateDictResizePolicy(); return C_OK; } return C_OK; /* unreached */ }我们接着看rdbSave()函数的源码：
在该函数中，就可以看见RDB文件的初始操作，刚开始生成一个临时的RDB文件，只有在执行成功后，才会进行rename操作，然后以写权限打开文件，然后调用了rdbSaveRio()函数将数据库的内容写到临时的RDB文件，之后进行刷新缓冲区和同步操作，就关闭文件进行rename操作和更新服务器状态。
我在此说一下rio，rio是Redis抽象的IO层，它可以面向三种对象，分别是缓冲区，文件IO和socket IO，在这里是调用rioInitWithFile()初始化了一个文件IO对象rdb，实际上SAVE和LOAD命令分别对rdb对象的写和读操作的封装，因此，可以直接调用rdbSave*一类的函数进行写操作。具体的rio源码剖析：Redis 输入输出的抽象(rio)源码剖析和注释，Redis 在复制部分，还实现了无盘复制，生成的RDB文件不保存在磁盘中，而是直接写向一个网络的socket，所以，在初始化rio时，只需调用初始化socket IO的接口，而写和读操作的函数接口都不变。
/* Save the DB on disk. Return C_ERR on error, C_OK on success. */ // 将数据库保存在磁盘上，返回C_OK成功，否则返回C_ERR int rdbSave(char *filename) { char tmpfile[256]; char cwd[MAXPATHLEN]; /* Current working dir path for error messages. */ FILE *fp; rio rdb; int error = 0; // 创建临时文件 snprintf(tmpfile,256,"temp-%d.rdb", (int) getpid()); // 以写方式打开该文件 fp = fopen(tmpfile,"w"); // 打开失败，获取文件目录，写入日志 if (!fp) { char *cwdp = getcwd(cwd,MAXPATHLEN); // 写日志信息到logfile serverLog(LL_WARNING, "Failed opening the RDB file %s (in server root dir %s) " "for saving: %s", filename, cwdp ? cwdp : "unknown", strerror(errno)); return C_ERR; } // 初始化一个rio对象，该对象是一个文件对象IO rioInitWithFile(&rdb,fp); // 将数据库的内容写到rio中 if (rdbSaveRio(&rdb,&error) == C_ERR) { errno = error; goto werr; } /* Make sure data will not remain on the OS's output buffers */ // 冲洗缓冲区，确保所有的数据都写入磁盘 if (fflush(fp) == EOF) goto werr; // 将fp指向的文件同步到磁盘中 if (fsync(fileno(fp)) == -1) goto werr; // 关闭文件 if (fclose(fp) == EOF) goto werr; /* Use RENAME to make sure the DB file is changed atomically only * if the generate DB file is ok. */ // 原子性改变rdb文件的名字 if (rename(tmpfile,filename) == -1) { // 改变名字失败，则获得当前目录路径，发送日志信息，删除临时文件 char *cwdp = getcwd(cwd,MAXPATHLEN); serverLog(LL_WARNING, "Error moving temp DB file %s on the final " "destination %s (in server root dir %s): %s", tmpfile, filename, cwdp ? cwdp : "unknown", strerror(errno)); unlink(tmpfile); return C_ERR; } // 写日志文件 serverLog(LL_NOTICE,"DB saved on disk"); // 重置服务器的脏键 server.dirty = 0; // 更新上一次SAVE操作的时间 server.lastsave = time(NULL); // 更新SAVE操作的状态 server.lastbgsave_status = C_OK; return C_OK; // rdbSaveRio()函数的写错误处理，写日志，关闭文件，删除临时文件，发送C_ERR werr: serverLog(LL_WARNING,"Write error saving DB on disk: %s", strerror(errno)); fclose(fp); unlink(tmpfile); return C_ERR; }因此，我们接着往下挖，查看一下rdbSaveRio()函数干了什么。
在rdbSaveRio()函数中，我们已经清楚的看到往RDB文件中写了什么内容。
例如：Redis标识，RDB版本号，rdb文件的默认信息，还有就是写数据库中的内容，接下来写入一个EOF码，最后执行校验和。因此一个完成的RDB文件如图所示：
// 将一个RDB格式文件内容写入到rio中，成功返回C_OK，否则C_ERR和一部分或所有的出错信息 // 当函数返回C_ERR，并且error不是NULL，那么error被设置为一个错误码errno int rdbSaveRio(rio *rdb, int *error) { dictIterator *di = NULL; dictEntry *de; char magic[10]; int j; long long now = mstime(); uint64_t cksum; // 开启了校验和选项 if (server.rdb_checksum) // 设置校验和的函数 rdb->update_cksum = rioGenericUpdateChecksum; // 将Redis版本信息保存到magic中 snprintf(magic,sizeof(magic),"REDISd",RDB_VERSION); // 将magic写到rio中 if (rdbWriteRaw(rdb,magic,9) == -1) goto werr; // 将rdb文件的默认信息写到rio中 if (rdbSaveInfoAuxFields(rdb) == -1) goto werr; // 遍历所有服务器内的数据库 for (j = 0; j < server.dbnum; j++) { redisDb *db = server.db+j; //当前的数据库指针 dict *d = db->dict; //当数据库的键值对字典 // 跳过为空的数据库 if (dictSize(d) == 0) continue; // 创建一个字典类型的迭代器 di = dictGetSafeIterator(d); if (!di) return C_ERR; /* Write the SELECT DB opcode */ // 写入数据库的选择标识码 RDB_OPCODE_SELECTDB为254 if (rdbSaveType(rdb,RDB_OPCODE_SELECTDB) == -1) goto werr; // 写入数据库的id，占了一个字节的长度 if (rdbSaveLen(rdb,j) == -1) goto werr; /* Write the RESIZE DB opcode. We trim the size to UINT32_MAX, which * is currently the largest type we are able to represent in RDB sizes. * However this does not limit the actual size of the DB to load since * these sizes are just hints to resize the hash tables. */ // 写入调整数据库的操作码，我们将大小限制在UINT32_MAX以内，这并不代表数据库的实际大小，只是提示去重新调整哈希表的大小 uint32_t db_size, expires_size; // 如果字典的大小大于UINT32_MAX，则设置db_size为最大的UINT32_MAX db_size = (dictSize(db->dict) <= UINT32_MAX) ? dictSize(db->dict) : UINT32_MAX; // 设置有过期时间键的大小超过UINT32_MAX，则设置expires_size为最大的UINT32_MAX expires_size = (dictSize(db->expires) <= UINT32_MAX) ? dictSize(db->expires) : UINT32_MAX; // 写入调整哈希表大小的操作码，RDB_OPCODE_RESIZEDB = 251 if (rdbSaveType(rdb,RDB_OPCODE_RESIZEDB) == -1) goto werr; // 写入提示调整哈希表大小的两个值，如果 if (rdbSaveLen(rdb,db_size) == -1) goto werr; if (rdbSaveLen(rdb,expires_size) == -1) goto werr; /* Iterate this DB writing every entry */ // 遍历数据库所有的键值对 while((de = dictNext(di)) != NULL) { sds keystr = dictGetKey(de); //当前键 robj key, *o = dictGetVal(de); //当前键的值 long long expire; // 在栈中创建一个键对象并初始化 initStaticStringObject(key,keystr); // 当前键的过期时间 expire = getExpire(db,&key); // 将键的键对象，值对象，过期时间写到rio中 if (rdbSaveKeyValuePair(rdb,&key,o,expire,now) == -1) goto werr; } dictReleaseIterator(di); //释放迭代器 } di = NULL; /* So that we don't release it again on error. */ /* EOF opcode */ // 写入一个EOF码，RDB_OPCODE_EOF = 255 if (rdbSaveType(rdb,RDB_OPCODE_EOF) == -1) goto werr; /* CRC64 checksum. It will be zero if checksum computation is disabled, the * loading code skips the check in this case. */ // CRC64检验和，当校验和计算为0，没有开启是，在载入rdb文件时会跳过 cksum = rdb->cksum; memrev64ifbe(&cksum); if (rioWrite(rdb,&cksum,8) == 0) goto werr; return C_OK; // 写入错误 werr: if (error) *error = errno; //保存错误码 if (di) dictReleaseIterator(di); //如果没有释放迭代器，则释放 return C_ERR; }调用rdbSaveInfoAuxFields()函数写入一些默认的辅助信息，具体如下：
/* Save a few default AUX fields with information about the RDB generated. */ // 将一个rdb文件的默认信息写入到rio中 int rdbSaveInfoAuxFields(rio *rdb) { // 判断主机的总线宽度，是64位还是32位 int redis_bits = (sizeof(void*) == 8) ? 64 : 32; /* Add a few fields about the state when the RDB was created. */ // 添加rdb文件的状态信息：Redis版本，redis位数，当前时间和Redis当前使用的内存数 if (rdbSaveAuxFieldStrStr(rdb,"redis-ver",REDIS_VERSION) == -1) return -1; if (rdbSaveAuxFieldStrInt(rdb,"redis-bits",redis_bits) == -1) return -1; if (rdbSaveAuxFieldStrInt(rdb,"ctime",time(NULL)) == -1) return -1; if (rdbSaveAuxFieldStrInt(rdb,"used-mem",zmalloc_used_memory()) == -1) return -1; return 1; }因此，一个空数据库持久化生成的dump.rdb文件，使用od -cx dump.rdb命令查看一下
0000000 R E D I S 0 0 0 7 372 \t r e d i s 4552 4944 3053 3030 fa37 7209 6465 7369 0000020 - v e r 005 3 . 2 . 8 372 \n r e d i 762d 7265 3305 322e 382e 0afa 6572 6964 0000040 s - b i t s 300 @ 372 005 c t i m e 302 2d73 6962 7374 40c0 05fa 7463 6d69 c265 0000060 u 7 \f Y 372 \b u s e d - m e m 302 0 3775 590c 08fa 7375 6465 6d2d 6d65 30c2 0000100 211 \f \0 377 8 341 Y 220 225 346 L 245 0c89 ff00 e138 9059 e695 a54c 0000114我们将其统计整合一下：
REDIS0007 372\t //Redis版本号：REDIS0007 redis-ver 005 3.2.8 372\n //Redis的版本：redis-ver 3.2.8 redis-bits 300 @ 372 005 //主机系统位数：redis-bits ctime 302 246 242 \b Y 372 \b //RDB操作的时间 userd-mem 302 205 \f \0 //子进程使用的内存量 377 //八进制377 = 十六进制255 = EOF常量 8 341 Y 220 225 346 L 245 //校验和：8字节虽然大概的看懂了一些，但是仍然还有一些八进制数字看不懂，这就是我们所描述RDB文件的特点：紧凑压缩。这些都是一些压缩过的数据或操作码。接下来，还是通过源码，查看这些压缩的规则，Redis将各种类型编码封装成许多函数，不利于查看编码规则，因此，我们就给出rdbLoad()函数，这个函数是服务器启动时，将RDB文件中的内容载入到数据库中。
rdbLoad()函数源码如下：
// 将指定的RDB文件读到数据库中 int rdbLoad(char *filename) { uint32_t dbid; int type, rdbver; redisDb *db = server.db+0; char buf[1024]; long long expiretime, now = mstime(); //获取当前load操作的时间 FILE *fp; rio rdb; // 只读打开文件 if ((fp = fopen(filename,"r")) == NULL) return C_ERR; // 初始化一个文件流对象rio且设置对应文件指针 rioInitWithFile(&rdb,fp); // 设置计算校验和的函数 rdb.update_cksum = rdbLoadProgressCallback; // 设置载入读或写的最大字节数，2M rdb.max_processing_chunk = server.loading_process_events_interval_bytes; // 读出9个字节到buf，buf中保存着Redis版本"redis0007" if (rioRead(&rdb,buf,9) == 0) goto eoferr; buf[9] = '\0'; //"redis0007\0" //检查读出的版本号标识 if (memcmp(buf,"REDIS",5) != 0) { fclose(fp); serverLog(LL_WARNING,"Wrong signature trying to load DB from file"); errno = EINVAL; //读出的值非法 return C_ERR; } // 转换成整数检查版本大小 rdbver = atoi(buf+5); if (rdbver < 1 || rdbver > RDB_VERSION) { fclose(fp); serverLog(LL_WARNING,"Can't handle RDB format version %d",rdbver); errno = EINVAL; return C_ERR; } // 设置载入时server的状态信息 startLoading(fp); // 开始读取RDB文件到数据库中 while(1) { robj *key, *val; expiretime = -1; /* Read type. */ // 首先读出类型 if ((type = rdbLoadType(&rdb)) == -1) goto eoferr; /* Handle special types. */ // 处理特殊情况 // 如果首先是读出过期时间单位为秒 if (type == RDB_OPCODE_EXPIRETIME) { /* EXPIRETIME: load an expire associated with the next key * to load. Note that after loading an expire we need to * load the actual type, and continue. */ // 从rio中读出过期时间 if ((expiretime = rdbLoadTime(&rdb)) == -1) goto eoferr; /* We read the time so we need to read the object type again. */ // 从过期时间后读出一个键值对的类型 if ((type = rdbLoadType(&rdb)) == -1) goto eoferr; /* the EXPIRETIME opcode specifies time in seconds, so convert * into milliseconds. */ expiretime *= 1000; //转换成毫秒 //读出过期时间单位为毫秒 } else if (type == RDB_OPCODE_EXPIRETIME_MS) { /* EXPIRETIME_MS: milliseconds precision expire times introduced * with RDB v3. Like EXPIRETIME but no with more precision. */ // 从rio中读出过期时间 if ((expiretime = rdbLoadMillisecondTime(&rdb)) == -1) goto eoferr; /* We read the time so we need to read the object type again. */ // 从过期时间后读出一个键值对的类型 if ((type = rdbLoadType(&rdb)) == -1) goto eoferr; // 如果读到EOF，则直接跳出循环 } else if (type == RDB_OPCODE_EOF) { /* EOF: End of file, exit the main loop. */ break; // 读出的是切换数据库操作 } else if (type == RDB_OPCODE_SELECTDB) { /* SELECTDB: Select the specified database. */ // 读取出一个长度，保存的是数据库的ID if ((dbid = rdbLoadLen(&rdb,NULL)) == RDB_LENERR) goto eoferr; // 检查读出的ID是否合法 if (dbid >= (unsigned)server.dbnum) { serverLog(LL_WARNING, "FATAL: Data file was created with a Redis " "server configured to handle more than %d " "databases. Exiting\n", server.dbnum); exit(1); } // 切换数据库 db = server.db+dbid; // 跳过本层循环，在读一个type continue; /* Read type again. */ // 如果读出调整哈希表的操作 } else if (type == RDB_OPCODE_RESIZEDB) { /* RESIZEDB: Hint about the size of the keys in the currently * selected data base, in order to avoid useless rehashing. */ uint32_t db_size, expires_size; // 读出一个数据库键值对字典的大小 if ((db_size = rdbLoadLen(&rdb,NULL)) == RDB_LENERR) goto eoferr; // 读出一个数据库过期字典的大小 if ((expires_size = rdbLoadLen(&rdb,NULL)) == RDB_LENERR) goto eoferr; // 扩展两个字典 dictExpand(db->dict,db_size); dictExpand(db->expires,expires_size); // 重新读出一个type continue; /* Read type again. */ // 读出的是一个辅助字段 } else if (type == RDB_OPCODE_AUX) { /* AUX: generic string-string fields. Use to add state to RDB * which is backward compatible. Implementations of RDB loading * are requierd to skip AUX fields they don't understand. * * An AUX field is composed of two strings: key and value. */ robj *auxkey, *auxval; // 读出辅助字段的键对象和值对象 if ((auxkey = rdbLoadStringObject(&rdb)) == NULL) goto eoferr; if ((auxval = rdbLoadStringObject(&rdb)) == NULL) goto eoferr; // 键对象的第一个字符是% if (((char*)auxkey->ptr)[0] == '%') { /* All the fields with a name staring with '%' are considered * information fields and are logged at startup with a log * level of NOTICE. */ // 写日志信息 serverLog(LL_NOTICE,"RDB '%s': %s", (char*)auxkey->ptr, (char*)auxval->ptr); } else { /* We ignore fields we don't understand, as by AUX field * contract. */ serverLog(LL_DEBUG,"Unrecognized RDB AUX field: '%s'", (char*)auxkey->ptr); } decrRefCount(auxkey); decrRefCount(auxval); // 重新读出一个type continue; /* Read type again. */ } /* Read key */ // 读出一个key对象 if ((key = rdbLoadStringObject(&rdb)) == NULL) goto eoferr; /* Read value */ // 读出一个val对象 if ((val = rdbLoadObject(type,&rdb)) == NULL) goto eoferr; /* Check if the key already expired. This function is used when loading * an RDB file from disk, either at startup, or when an RDB was * received from the master. In the latter case, the master is * responsible for key expiry. If we would expire keys here, the * snapshot taken by the master may not be reflected on the slave. */ // 如果当前环境不是从节点，且该键设置了过期时间，已经过期 if (server.masterhost == NULL && expiretime != -1 && expiretime < now) { // 释放键值对 decrRefCount(key); decrRefCount(val); continue; } /* Add the new object in the hash table */ // 将没有过期的键值对添加到数据库键值对字典中 dbAdd(db,key,val); /* Set the expire time if needed */ // 如果需要，设置过期时间 if (expiretime != -1) setExpire(db,key,expiretime); decrRefCount(key); //释放临时对象 } // 此时已经读出完所有数据库的键值对，读到了EOF，但是EOF不是RDB文件的结束，还要进行校验和 /* Verify the checksum if RDB version is >= 5 */ // 当RDB版本大于5时，且开启了校验和的功能，那么进行校验和 if (rdbver >= 5 && server.rdb_checksum) { uint64_t cksum, expected = rdb.cksum; // 读出一个8字节的校验和，然后比较 if (rioRead(&rdb,&cksum,8) == 0) goto eoferr; memrev64ifbe(&cksum); if (cksum == 0) { serverLog(LL_WARNING,"RDB file was saved with checksum disabled: no check performed."); } else if (cksum != expected) { serverLog(LL_WARNING,"Wrong RDB checksum. Aborting now."); rdbExitReportCorruptRDB("RDB CRC error"); } } fclose(fp); //关闭RDB文件 stopLoading(); //设置载入完成的状态 return C_OK; // 错误退出 eoferr: /* unexpected end of file is handled here with a fatal exit */ serverLog(LL_WARNING,"Short read or OOM loading DB. Unrecoverable error, aborting now."); // 检查rdb错误发送信息且退出 rdbExitReportCorruptRDB("Unexpected EOF reading RDB file"); return C_ERR; /* Just to avoid warning */ }从这个函数中，我们可以看到许多RDB_TYPE_*类型的对象，他们定义在rdb.h中。
/* Dup object types to RDB object types. Only reason is readability (are we * dealing with RDB types or with in-memory object types?). */ #define RDB_TYPE_STRING 0 //字符串类型 #define RDB_TYPE_LIST 1 //列表类型 #define RDB_TYPE_SET 2 //集合类型 #define RDB_TYPE_ZSET 3 //有序集合类型 #define RDB_TYPE_HASH 4 //哈希类型 /* NOTE: WHEN ADDING NEW RDB TYPE, UPDATE rdbIsObjectType() BELOW */ /* Object types for encoded objects. */ #define RDB_TYPE_HASH_ZIPMAP 9 #define RDB_TYPE_LIST_ZIPLIST 10 //列表对象的ziplist编码类型 #define RDB_TYPE_SET_INTSET 11 //集合对象的intset编码类型 #define RDB_TYPE_ZSET_ZIPLIST 12 //有序集合的ziplist编码类型 #define RDB_TYPE_HASH_ZIPLIST 13 //哈希对象的ziplist编码类型 #define RDB_TYPE_LIST_QUICKLIST 14 //列表对象的quicklist编码类型 /* NOTE: WHEN ADDING NEW RDB TYPE, UPDATE rdbIsObjectType() BELOW */ /* Test if a type is an object type. */ // 测试t是否是一个对象的编码类型 #define rdbIsObjectType(t) ((t >= 0 && t <= 4) || (t >= 9 && t <= 14)) /* Special RDB opcodes (saved/loaded with rdbSaveType/rdbLoadType). */ #define RDB_OPCODE_AUX 250 //辅助标识 #define RDB_OPCODE_RESIZEDB 251 //提示调整哈希表大小的操作码 #define RDB_OPCODE_EXPIRETIME_MS 252 //过期时间毫秒 #define RDB_OPCODE_EXPIRETIME 253 //过期时间秒 #define RDB_OPCODE_SELECTDB 254 //选择数据库的操作 #define RDB_OPCODE_EOF 255 //EOF码因此，看到这，我们就可以剖析dump.rdb文件了。
0000000 R E D I S 0 0 0 7 372 \t r e d i s 4552 4944 3053 3030 fa37 7209 6465 7369 0000020 - v e r 005 3 . 2 . 8 372 \n r e d i 762d 7265 3305 322e 382e 0afa 6572 6964 0000040 s - b i t s 300 @ 372 005 c t i m e 302 2d73 6962 7374 40c0 05fa 7463 6d69 c265 0000060 u 7 \f Y 372 \b u s e d - m e m 302 0 3775 590c 08fa 7375 6465 6d2d 6d65 30c2 0000100 211 \f \0 377 8 341 Y 220 225 346 L 245 0c89 ff00 e138 9059 e695 a54c 0000114八进制372 对应着十进制的RDB_OPCODE_AUX，然后在到rdbLoad()函数中，找到type == RDB_OPCODE_AUX的情况，要分别读出一个键对象和一个值对象；
读对象时，先读1个字节的长度，因此八进制'\t'对应十进制的9，所以在读键对象的长度为9字节，正如所分析的，redis-ver长度为9字节。 然后读出一值对象，先读1字节的长度，因此八进制的005对应十进制的5，所以在读出值对象的长度为5字节，正如所分析的，3.2.8长度为5字节。判断完type == RDB_OPCODE_AUX的情况，然后根据代码，要跳出当前循环，于是，在读出1个字节的type，此时type =还是372，于是还是分别读出一个键对象和一个值对象；
读对象时，先读1个字节的长度，因此八进制'\n'对应十进制的10，所以在读键对象的长度为10字节，正如所分析的，redis-bits长度为10字节。然后读出一值对象，先读1字节的长度，因此八进制的300对应十进制的192，此时，这显然不对，是因为RDB是经过压缩过得文件，接下来，我们介绍压缩的规则： /* When a length of a string object stored on disk has the first two bits * set, the remaining two bits specify a special encoding for the object * accordingly to the following defines: */ #define RDB_ENC_INT8 0 /* 8位有符号整数 8 bit signed integer */ #define RDB_ENC_INT16 1 /* 16位有符号整数 16 bit signed integer */ #define RDB_ENC_INT32 2 /* 32位有符号整数 32 bit signed integer */ #define RDB_ENC_LZF 3 /* LZF压缩过的字符串 string compressed with FASTLZ */ #define RDB_6BITLEN 0 //6位长 #define RDB_14BITLEN 1 //14位长 #define RDB_32BITLEN 2 //32位长 #define RDB_ENCVAL 3 //编码值 #define RDB_LENERR UINT_MAX //错误值一个字符串压缩可能有如上4种，它的读法，可以看rdbLoadLen()函数的源码：可以从这个函数中看出，不同编码类型，保存值的长度所占的字节数。
我们读一值对象，先读1字节的长度，因此八进制的300对应二进制的1100 0000，它的最高两位是11，十进制是3，对应RDB_ENCVAL类型，并且返回0。 // 返回一个从rio读出的len值，如果该len值不是整数，而是被编码后的值，那么将isencoded设置为1 uint32_t rdbLoadLen(rio *rdb, int *isencoded) { unsigned char buf[2]; uint32_t len; int type; // 默认为没有编码 if (isencoded) *isencoded = 0; // 将rio中的值读到buf中 if (rioRead(rdb,buf,1) == 0) return RDB_LENERR; // (buf[0]&0xC0)>>6 = (1100 000 & buf[0]) >> 6 = buf[0]的最高两位 type = (buf[0]&0xC0)>>6; // 一个编码过的值，返回解码值，设置编码标志 if (type == RDB_ENCVAL) { /* Read a 6 bit encoding type. */ if (isencoded) *isencoded = 1; return buf[0]&0x3F; //取出剩下六位表示的长度值 // 一个6位长的值 } else if (type == RDB_6BITLEN) { /* Read a 6 bit len. */ return buf[0]&0x3F; //取出剩下六位表示的长度值 // 一个14位长的值 } else if (type == RDB_14BITLEN) { /* Read a 14 bit len. */ // 从buf+1读出1个字节的值 if (rioRead(rdb,buf+1,1) == 0) return RDB_LENERR; return ((buf[0]&0x3F)<<8)|buf[1]; //取出除最高两位的长度值 // 一个32位长的值 } else if (type == RDB_32BITLEN) { /* Read a 32 bit len. */ // 读出4个字节的值 if (rioRead(rdb,&len,4) == 0) return RDB_LENERR; return ntohl(len); //转换为主机序的值 } else { rdbExitReportCorruptRDB( "Unknown length encoding %d in rdbLoadLen()",type); return -1; /* Never reached. */ } } 然后回到创建字符串对象的函数rdbGenericLoadStringObject()，rdbLoadLen()函数的返回值是0，对应RDB_ENC_INT8，然后又调用了rdbLoadIntegerObject()函数。 // 根据flags，将从rio读出一个字符串对象进行编码 void *rdbGenericLoadStringObject(rio *rdb, int flags) { int encode = flags & RDB_LOAD_ENC; //编码 int plain = flags & RDB_LOAD_PLAIN; //原生的值 int isencoded; uint32_t len; // 从rio中读出一个字符串对象，编码类型保存在isencoded中，所需的字节为len len = rdbLoadLen(rdb,&isencoded); // 如果读出的对象被编码(isencoded被设置为1)，则根据不同的长度值len映射到不同的整数编码 if (isencoded) { switch(len) { case RDB_ENC_INT8: case RDB_ENC_INT16: case RDB_ENC_INT32: // 以上三种类型的整数编码，根据flags返回不同类型值 return rdbLoadIntegerObject(rdb,len,flags); case RDB_ENC_LZF: // 如果是压缩后的字符串，进行构建压缩字符串编码对象 return rdbLoadLzfStringObject(rdb,flags); default: rdbExitReportCorruptRDB("Unknown RDB string encoding type %d",len); } } // 如果len值错误，则返回NULL if (len == RDB_LENERR) return NULL; // 如果不是原生值 if (!plain) { // 根据encode编码类型创建不同的字符串对象 robj *o = encode ? createStringObject(NULL,len) : createRawStringObject(NULL,len); // 设置o对象的值，从rio中读出来，如果失败，释放对象返回NULL if (len && rioRead(rdb,o->ptr,len) == 0) { decrRefCount(o); return NULL; } return o; // 如果设置了原生值 } else { // 分配空间 void *buf = zmalloc(len); // 从rio中读出来 if (len && rioRead(rdb,buf,len) == 0) { zfree(buf); return NULL; } return buf; //返回 } } 当传入的编码是RDB_ENC_INT8时。它又从后面读取了1字节。后面的八进制值\n，对应十进制为64，因此redis-bits所对应的值为64，也就是64位的Redis服务器。
// 将rio中的整数值根据不同的编码读出来，并根据flags构建成一个不同类型的值并返回 void *rdbLoadIntegerObject(rio *rdb, int enctype, int flags) { int plain = flags & RDB_LOAD_PLAIN; //无格式 int encode = flags & RDB_LOAD_ENC; //字符串对象 unsigned char enc[4]; long long val; // 根据不同的整数编码类型，从rio中读出整数值到enc中 if (enctype == RDB_ENC_INT8) { if (rioRead(rdb,enc,1) == 0) return NULL; val = (signed char)enc[0]; } else if (enctype == RDB_ENC_INT16) { uint16_t v; if (rioRead(rdb,enc,2) == 0) return NULL; v = enc[0]|(enc[1]<<8); val = (int16_t)v; } else if (enctype == RDB_ENC_INT32) { uint32_t v; if (rioRead(rdb,enc,4) == 0) return NULL; v = enc[0]|(enc[1]<<8)|(enc[2]<<16)|(enc[3]<<24); val = (int32_t)v; } else { val = 0; /* anti-warning */ rdbExitReportCorruptRDB("Unknown RDB integer encoding type %d",enctype); } // 如果是整数，转换为字符串类型返回 if (plain) { char buf[LONG_STR_SIZE], *p; int len = ll2string(buf,sizeof(buf),val); p = zmalloc(len); memcpy(p,buf,len); return p; // 如果是编码过的整数值，则转换为字符串对象，返回 } else if (encode) { return createStringObjectFromLongLong(val); } else { // 返回一个字符串对象 return createObject(OBJ_STRING,sdsfromlonglong(val)); } }此时，也就介绍完了所有规则，后面的分析和之前的如出一辙，因此，不在继续分析了。SAVE和LOAD是相反的过程，因此可以反过来理解。
我将RDB持久化所有的源码放在了github上，欢迎阅读：Redis 3.2 源码注释
