写在文章开头
为避免服务器宕机着情况导致redis
内存数据库数据丢失，redis
默认出通过rdb
保证可靠性，本文将从源码的角度带读者了解rdb
读写时机和写入流程。
Hi，我是 sharkChili ，是个不断在硬核技术上作死的技术人，是 CSDN的博客专家 ，也是开源项目 Java Guide 的维护者之一，熟悉 Java 也会一点 Go ，偶尔也会在 C源码 边缘徘徊。写过很多有意思的技术博客，也还在研究并输出技术的路上，希望我的文章对你有帮助，非常欢迎你关注我的公众号： 写代码的SharkChili 。
同时也非常欢迎你star我的开源项目mini-redis:https://github.com/shark-ctrl/mini-redis
因为近期收到很多读者的私信，所以也专门创建了一个交流群，感兴趣的读者可以通过上方的公众号获取笔者的联系方式完成好友添加，点击备注 “加群” 即可和笔者和笔者的朋友们进行深入交流。
详解RDB持久化
save指令触发rdb
redis
支持通过命令的方式持久化内存数据库数据，当我们键入save
的时候，redis
解析到这个指令之后，主线程直接调用saveCommand
方法生成rdb文件落到磁盘中。
我们可以在rdb.c
文件中看到该方法的实现，可以看到为了避免脏写等问题，saveCommand
会检查当前是否有rdb
子进程执行，如果没有在子进程执行rdb
持久化则直接调用rdbSave
方法生成dump.rdb
文件落盘：
//调用save指令其内部调用rdbSave完成rdb文件生成
void saveCommand(redisClient *c) {
//检查是否子进程执行rdb，若有则直接返回
if (server.rdb_child_pid != -1) {
addReplyError(c,"Background save already in progress");
return;
}
//调用rdbSave
if (rdbSave(server.rdb_filename) == REDIS_OK) {
addReply(c,shared.ok);
} else {
addReply(c,shared.err);
}
}复制
步入rdbSave
即可看到生成临时rdb
写入数据，然后数据刷盘，最后完成文件名原子修改的操作：
int rdbSave(char *filename) {
char tmpfile[256];
FILE *fp;
rio rdb;
int error;
//生成一个tmp文件
snprintf(tmpfile,256,"temp-%d.rdb", (int) getpid());
fp = fopen(tmpfile,"w");
if (!fp) {
redisLog(REDIS_WARNING, "Failed opening .rdb for saving: %s",
strerror(errno));
return REDIS_ERR;
}
//调用rdbSaveRio完成数据写入
rioInitWithFile(&rdb,fp);
if (rdbSaveRio(&rdb,&error) == REDIS_ERR) {
errno = error;
goto werr;
}
//直接刷盘到磁盘，避免留在系统输出缓冲区
/* Make sure data will not remain on the OS's output buffers */
if (fflush(fp) == EOF) goto werr;
if (fsync(fileno(fp)) == -1) goto werr;
if (fclose(fp) == EOF) goto werr;
//完成写入后文件重命名为dump.rdb
if (rename(tmpfile,filename) == -1) {
redisLog(REDIS_WARNING,"Error moving temp DB file on the final destination: %s", strerror(errno));
unlink(tmpfile);
return REDIS_ERR;
}
//......
return REDIS_OK;
//......
}复制
bgsave指令触发rdb
同时redis
也支持后台持久化，如果用户需要考虑redis
性能问题，可以直接通过bgsave
指令创建rdb
子进程完成数据库数据持久化。
我们同样可以在rdb.c
文件中看到bgsave指令调用的方法bgsaveCommand
，可以看到如果没有子进程进行rdb
或者aof
，该指令会调用rdbSaveBackground
完成异步数据持久化：
//调用rdbSaveBackground创建一个子进程生成rdb文件，不影响主线程
void bgsaveCommand(redisClient *c) {
//如果有子进程执行rdb或者aof，则直接返回错误提醒
if (server.rdb_child_pid != -1) {
addReplyError(c,"Background save already in progress");
} else if (server.aof_child_pid != -1) {
addReplyError(c,"Can't BGSAVE while AOF log rewriting is in progress");
} else if (rdbSaveBackground(server.rdb_filename) == REDIS_OK) {//调用rdbSaveBackground进行数据持久化
addReplyStatus(c,"Background saving started");
} else {
addReply(c,shared.err);
}
}复制
步入rdbSaveBackground
可以看到，其内部还会检查一次是否有文件进行rdb
，如果明确没有之后直接fork
一个子进程出来调用上文所说的rdbSave
完成数据持久化到dump.rdb
中：
int rdbSaveBackground(char *filename) {
pid_t childpid;
long long start;
if (server.rdb_child_pid != -1) return REDIS_ERR;
//......
start = ustime();
if ((childpid = fork()) == 0) {//创建子进程
int retval;
//......
retval = rdbSave(filename);//生成rdb文件
exitFromChild((retval == REDIS_OK) ? 0 : 1);//退出子进程
} else {
//......
}
return REDIS_OK; /* unreached */
}复制
rdb被动触发
redis
被动触发由时间事件轮询处理，我们可以在redis.conf
配置rdb被动触发持久化的时机，默认配置如下当60s
生成10000
或者300s
生成10
次改变亦或者900s
生成1次改变，我们就会执行一次被动rdb
持久化：
save 900 1
save 300 10
save 60 10000复制
对应的我们可以在redis.c
的serverCron
函数在看到这段逻辑，它会遍历出我们配置的保存间隔配置saveparam
，通过比对这3条配置的上次保存时间计算出时间间隔，以及当前redis
变化书dirty看看是否符合要求，若如何要求则进行后台rdb持久化：
int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
//......
/* Check if a background saving or AOF rewrite in progress terminated. */
if (server.rdb_child_pid != -1 || server.aof_child_pid != -1) {
//......
}
} else {
//遍历3个配置的params，如果改变数和事件间隔配置要求则直接进行后台被动rdb持久化
for (j = 0; j < server.saveparamslen; j++) {
struct saveparam *sp = server.saveparams+j;
if (server.dirty >= sp->changes && //查看变化数是否大于当前配置的changes
server.unixtime-server.lastsave > sp->seconds && //查看时间间隔是否大于配置
(server.unixtime-server.lastbgsave_try >
REDIS_BGSAVE_RETRY_DELAY ||
server.lastbgsave_status == REDIS_OK))
{
//......
//执行异步持久化
rdbSaveBackground(server.rdb_filename);
break;
}
}
//......
}
}
//......
return 1000/server.hz;
}复制
其他被动落盘时机
其实有些时候我们执行的某些执行也会进行rdb
持久化，例如flushall
刷盘指令,其调用函数flushallCommand
就会时间串行执行rdb
持久化：
//调用flush指令时会调用rdbSave进行数据持久化
void flushallCommand(redisClient *c) {
//......
if (server.saveparamslen > 0) {
//串行执行rdb持久化
int saved_dirty = server.dirty;
rdbSave(server.rdb_filename);
//......
}
server.dirty++;
}复制
当我们关闭redis
服务器的时候也会执行rdb
串行持久化：
//服务器进程关闭时调用rdbSave生成rdb文件
int prepareForShutdown(int flags) {
//......
if (server.rdb_child_pid != -1) {
//......
}
if (server.aof_state != REDIS_AOF_OFF) {
//......
}
if ((server.saveparamslen > 0 && !nosave) || save) {
if (rdbSave(server.rdb_filename) != REDIS_OK) {
//......
return REDIS_ERR;
}
}
//......
return REDIS_OK;
}复制
rdb写入文件数据详解
无论是rdbsave
还是rdbbgsave
对应的方法，其内部都会调用rdbSaveRio
，它进行文件写入时对应写入数据大体顺序是:
写入 REDIS
大写。补0填充长度。 写入当前 redis
版本号，以笔者源码为例则是6。遍历数据库写入 REDIS_RDB_OPCODE_SELECTDB
表示开始存储数据库数据，这个值默认为254
，redis
会转为八进制376
写入。遍历当前数据库键值对 key
长度和key
，value
长度和value
写入，后续数据库都是如此往复。所有数据库写完后补上 REDIS_RDB_OPCODE_EOF
和checksum
用于后续rdb
数据恢复的校验。
为保证读者更直观的了解redis持久化写入的内容，我们可以删除本地rdb
文件，然后执行如下执行生成一个全新的rdb
文件：
# 保存键值对
set key value
# 切换到1库
select 1
# 保存键值对到1库
set key-1 value
# 调用save进行数据持久化
save复制
正常情况下我们打开rdb
文件会得到一堆类型乱码的内容，我们无法知晓写入的信息，我们可以直接键入od
生成rdb文件16
进制数据及其对应的ASCII
字符：
od -A x -t x1c -v dump.rdb复制
最终我们就可以得到如下文件，可以看到数据格式和笔者上文所说基本一致：
# 大写REDIS 补0 254的8进制 当前数据库索引 键值对`key`长度和`key`，`value`长度和`value`
#000000 52 45 44 49 53 30 30 30 36 fe 00 00 03 6b 65 79
R E D I S 0 0 0 6 376 \0 \0 003 k e y
000010 05 76 61 6c 75 65 fe 01 00 05 6b 65 79 2d 31 05
005 v a l u e
# 254的8进制 当前数据库索引1 键值对key长度和key，value长度和value
376 001 \0 005 k e y - 1 005
000020 76 61 6c 75 65 ff 76 eb e4 80 bd df 66 11
v a l u e
# EOF 255八进制 剩下8位是对应的checksum
377 v 353 344 200 275 337 f 021
00002e复制
对应的我们给出这段源码，对应的写入流程如上文笔者所述：
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;
snprintf(magic,sizeof(magic),"REDIS%04d",REDIS_RDB_VERSION);//对应redis 3个0 然后版本号，当前版本为6
if (rdbWriteRaw(rdb,magic,9) == -1) goto werr;//上述魔数写入rdb文件
//遍历数据库
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 REDIS_ERR;
/* Write the SELECT DB opcode */
if (rdbSaveType(rdb,REDIS_RDB_OPCODE_SELECTDB) == -1) goto werr;//写入254，也就是内容中的376
if (rdbSaveLen(rdb,j) == -1) goto werr;//写入当前库索引
//遍历当前键值对写入
while((de = dictNext(di)) != NULL) {
sds keystr = dictGetKey(de);
robj key, *o = dictGetVal(de);
long long expire;
initStaticStringObject(key,keystr);
expire = getExpire(db,&key);
if (rdbSaveKeyValuePair(rdb,&key,o,expire,now) == -1) goto werr;//写入键值对
}
dictReleaseIterator(di);
}
//......
/* EOF opcode */
if (rdbSaveType(rdb,REDIS_RDB_OPCODE_EOF) == -1) goto werr;//写入结束符254 八进制为377
cksum = rdb->cksum;
memrev64ifbe(&cksum);
if (rioWrite(rdb,&cksum,8) == 0) goto werr;//写入8位数校验和,其底层调用rioGenericUpdateChecksum，按照cksum到数组中获取就对应的值并
return REDIS_OK;
//......
}复制
对应的我们步入rdbSaveKeyValuePair
即可看到redis
获取key
长度和key，以及value
长度和value
并写入rdb
文件的核心流程:
int rdbSaveKeyValuePair(rio *rdb, robj *key, robj *val,
long long expiretime, long long now)
{
//......
/* Save type, key, value */
if (rdbSaveObjectType(rdb,val) == -1) return -1;//写入类型以字符串形式就是0
if (rdbSaveStringObject(rdb,key) == -1) return -1;//写入key长度和key
if (rdbSaveObject(rdb,val) == -1) return -1;//写入value长度和value
return 1;
}复制
小结
自此我们将redis
持久化策略rdb都分析完成了，希望对你有帮助。
我是 sharkchili ，CSDN Java 领域博客专家，mini-redis的作者，我想写一些有意思的东西，希望对你有帮助，如果你想实时收到我写的硬核的文章也欢迎你关注我的公众号： 写代码的SharkChili 。
同时也非常欢迎你star我的开源项目mini-redis:https://github.com/shark-ctrl/mini-redis
因为近期收到很多读者的私信，所以也专门创建了一个交流群，感兴趣的读者可以通过上方的公众号获取笔者的联系方式完成好友添加，点击备注 “加群” 即可和笔者和笔者的朋友们进行深入交流。
参考
《Redis 源码剖析与实战》
