深入剖析Redis RDB持久化机制

jieforest

对是否有vm线程进行再次判断，因为如果是通过save命令过来的是没有判断过vm线程的。

创建并打开临时文件

写入文件签名“REDIS”和版本号“0002”

遍历所有db中的所有key

对每个key，先判断是否设置了expireTime, 如果设置了，则保存expireTime到rdb文件中。然后判断该key对应的value是否则内存中，如果是在内存中，则取出来写入到rdb文件中保存，如果被换出到虚拟内存了，则从虚拟内存读取然后写入到rdb文件中。

不同类型有有不同的存储格式，详细见rdb文件格式

最后写入rdb文件的结束符

关闭文件并重命名临时文件名到正式文件名

更新脏数据计数server.dirty为0和最近写rdb文件的时间server.lastsave为当前时间，这个只是在通过save命令触发的情况下有用。因为如果是通过fork一个子进程来写rdb文件的，更新无效，因为更新的是子进程的数据。

jieforest

如果是通过fork一个子进程来写rdb文件（即不是通过save命令触发的），在写rdb文件的过程中，可能又有一些数据被更改了，那此时的脏数据计数server.dirty怎么更新呢？ redis是怎样处理的呢?
我们来看看写rdb的子进程推出时得处理

Redis.c：605

1. if (server.bgsavechildpid != -1 || server.bgrewritechildpid != -1) {
   
2.         int statloc;
   
3.         pid_t pid;
   
4. 
   
5.         if ((pid = wait3(&statloc,WNOHANG,NULL)) != 0) {
   
6.             if (pid == server.bgsavechildpid) {
   
7.                 backgroundSaveDoneHandler(statloc);
   
8.             } else {
   
9.                 backgroundRewriteDoneHandler(statloc);
   
10.             }
    
11.             updateDictResizePolicy();
    
12.         }
    
13. }

复制代码

jieforest

如果捕捉到写rdb文件的子进程退出，则调用backgroundSaveDoneHandler进行处理

接着看看backgroundSaveDoneHandler函数

Rdb.c：1062

1. void backgroundSaveDoneHandler(int statloc) {
   
2.     int exitcode = WEXITSTATUS(statloc);
   
3.     int bysignal = WIFSIGNALED(statloc);
   
4. 
   
5.     if (!bysignal && exitcode == 0) {
   
6.         redisLog(REDIS_NOTICE,
   
7.             "Background saving terminated with success");
   
8.         server.dirty = server.dirty - server.dirty_before_bgsave;
   
9.         server.lastsave = time(NULL);
   
10.     } else if (!bysignal && exitcode != 0) {
    
11.         redisLog(REDIS_WARNING, "Background saving error");
    
12.     } else {
    
13.         redisLog(REDIS_WARNING,
    
14.             "Background saving terminated by signal %d", WTERMSIG(statloc));
    
15.         rdbRemoveTempFile(server.bgsavechildpid);
    
16.     }
    
17.     server.bgsavechildpid = -1;
    
18.     /* Possibly there are slaves waiting for a BGSAVE in order to be served
    
19.      * (the first stage of SYNC is a bulk transfer of dump.rdb) */
    
20.     updateSlavesWaitingBgsave(exitcode == 0 ? REDIS_OK : REDIS_ERR);
    
21. }

复制代码

jieforest

更新脏数据计数server.dirty为0和最近写rdb文件的时间server.lastsave为当前时间

唤醒因为正在保存快照而等待的slave，关于slave的具体内容，见replication

快照导入

当redis因为停电或者某些原因挂掉了，此时重启redis时，我们就需要从rdb文件中读取快照文件，把保存到rdb文件中的数据重新导入到内存中。

先来看看启动时对快照导入的处理

Redis.c：1717

1.     if (server.appendonly) {
   
2.         if (loadAppendOnlyFile(server.appendfilename) == REDIS_OK)
   
3.             redisLog(REDIS_NOTICE,"DB loaded from append only file: %ld seconds",time(NULL)-start);
   
4.     } else {
   
5.         if (rdbLoad(server.dbfilename) == REDIS_OK) {
   
6.             redisLog(REDIS_NOTICE,"DB loaded from disk: %ld seconds",
   
7.                 time(NULL)-start);
   
8.         } else if (errno != ENOENT) {
   
9.             redisLog(REDIS_WARNING,"Fatal error loading the DB. Exiting.");
   
10.             exit(1);
    
11.         }
    
12.     }

复制代码

jieforest

如果保存了AOF文件，则使用AOF文件来恢复数据，AOF的具体内容见AOF

如果没有AOF，则使用rdb文件恢复数据，调用rdbLoad函数

接着看看rdbLoad函数

Rdb.c：929

1. int rdbLoad(char *filename) {
   
2.     ...
   
3.     fp = fopen(filename,"r");
   
4.     if (!fp) {
   
5.         errno = ENOENT;
   
6.         return REDIS_ERR;
   
7.     }
   
8.     if (fread(buf,9,1,fp) == 0) goto eoferr;
   
9.     buf[9] = '\0';
   
10.     if (memcmp(buf,"REDIS",5) != 0) {
    
11.         fclose(fp);
    
12.         redisLog(REDIS_WARNING,"Wrong signature trying to load DB from file");
    
13.         errno = EINVAL;
    
14.         return REDIS_ERR;
    
15.     }
    
16.     rdbver = atoi(buf+5);
    
17.     if (rdbver < 1 || rdbver > 2) {
    
18.         fclose(fp);
    
19.         redisLog(REDIS_WARNING,"Can't handle RDB format version %d",rdbver);
    
20.         errno = EINVAL;
    
21.         return REDIS_ERR;
    
22.     }
    
23. 
    
24.     startLoading(fp);
    
25.     while(1) {
    
26.         robj *key, *val;
    
27.         int force_swapout;
    
28. 
    
29.         expiretime = -1;
    
30. 
    
31.         /* Serve the clients from time to time */
    
32.         if (!(loops++ % 1000)) {
    
33.             loadingProgress(ftello(fp));
    
34.             aeProcessEvents(server.el, AE_FILE_EVENTS|AE_DONT_WAIT);
    
35.         }
    
36. 
    
37.         /* Read type. */
    
38.         if ((type = rdbLoadType(fp)) == -1) goto eoferr;
    
39.         if (type == REDIS_EXPIRETIME) {
    
40.             if ((expiretime = rdbLoadTime(fp)) == -1) goto eoferr;
    
41.             /* We read the time so we need to read the object type again */
    
42.             if ((type = rdbLoadType(fp)) == -1) goto eoferr;
    
43.         }
    
44.         if (type == REDIS_EOF) break;
    
45.         /* Handle SELECT DB opcode as a special case */
    
46.         if (type == REDIS_SELECTDB) {
    
47.             if ((dbid = rdbLoadLen(fp,NULL)) == REDIS_RDB_LENERR)
    
48.                 goto eoferr;
    
49.             if (dbid >= (unsigned)server.dbnum) {
    
50.                 redisLog(REDIS_WARNING,"FATAL: Data file was created with a Redis server configured to handle more than %d databases. Exiting\n", server.dbnum);
    
51.                 exit(1);
    
52.             }
    
53.             db = server.db+dbid;
    
54.             continue;
    
55.         }
    
56.         /* Read key */
    
57.         if ((key = rdbLoadStringObject(fp)) == NULL) goto eoferr;
    
58.         /* Read value */
    
59.         if ((val = rdbLoadObject(type,fp)) == NULL) goto eoferr;
    
60.         /* Check if the key already expired. This function is used when loading
    
61.          * an RDB file from disk, either at startup, or when an RDB was
    
62.          * received from the master. In the latter case, the master is
    
63.          * responsible for key expiry. If we would expire keys here, the
    
64.          * snapshot taken by the master may not be reflected on the slave. */
    
65.         if (server.masterhost == NULL && expiretime != -1 && expiretime < now) {
    
66.             decrRefCount(key);
    
67.             decrRefCount(val);
    
68.             continue;
    
69.         }
    
70.         /* Add the new object in the hash table */
    
71.         dbAdd(db,key,val);
    
72. 
    
73.         /* Set the expire time if needed */
    
74.         if (expiretime != -1) setExpire(db,key,expiretime);
    
75. 
    
76.         /* Handle swapping while loading big datasets when VM is on */
    
77. 
    
78.         /* If we detecter we are hopeless about fitting something in memory
    
79.          * we just swap every new key on disk. Directly…
    
80.          * Note that's important to check for this condition before resorting
    
81.          * to random sampling, otherwise we may try to swap already
    
82.          * swapped keys. */
    
83.         if (swap_all_values) {
    
84.             dictEntry *de = dictFind(db->dict,key->ptr);
    
85. 
    
86.             /* de may be NULL since the key already expired */
    
87.             if (de) {
    
88.                 vmpointer *vp;
    
89.                 val = dictGetEntryVal(de);
    
90. 
    
91.                 if (val->refcount == 1 &&
    
92.                     (vp = vmSwapObjectBlocking(val)) != NULL)
    
93.                     dictGetEntryVal(de) = vp;
    
94.             }
    
95.             decrRefCount(key);
    
96.             continue;
    
97.         }
    
98.         decrRefCount(key);
    
99. 
    
100.         /* Flush data on disk once 32 MB of additional RAM are used… */
     
101.         force_swapout = 0;
     
102.         if ((zmalloc_used_memory() - server.vm_max_memory) > 1024*1024*32)
     
103.             force_swapout = 1;
     
104. 
     
105.         /* If we have still some hope of having some value fitting memory
     
106.          * then we try random sampling. */
     
107.         if (!swap_all_values && server.vm_enabled && force_swapout) {
     
108.             while (zmalloc_used_memory() > server.vm_max_memory) {
     
109.                 if (vmSwapOneObjectBlocking() == REDIS_ERR) break;
     
110.             }
     
111.             if (zmalloc_used_memory() > server.vm_max_memory)
     
112.                 swap_all_values = 1; /* We are already using too much mem */
     
113.         }
     
114.     }
     
115.     fclose(fp);
     
116.     stopLoading();
     
117.     return REDIS_OK;
     
118. 
     
119. eoferr: /* unexpected end of file is handled here with a fatal exit */
     
120.     redisLog(REDIS_WARNING,"Short read or OOM loading DB. Unrecoverable error, aborting now.");
     
121.     exit(1);
     
122.     return REDIS_ERR; /* Just to avoid warning */
     
123. }

复制代码

jieforest

打开rdb文件

读取rdb文件的签名和版本号

开始进入 类型 | 值 | 类型 | 值 的循环读取，可参考rdb文件格式

作者还做了导入的进度条，是有人反馈说rdb文件很大时导入时要很久，但又不知道进度，所以作者就加了导入的进度条，改善用户体验

读取类型

如果类型是过期时间类型REDIS_EXPIRETIME，则读取过期时间

如果类型是文件结束类型REDIS_EOF，则跳出 类型 | 值 | 类型 | 值 的循环读取

如果类型是选择db类型REDIS_SELECTDB，则读取db索引并把当前db转成该db，然后继续 类型 | 值 | 类型 | 值 的循环读取。

如果不是以上类型，则表明该类型是数据类型，读取作为key的字符串，即读取字符串类型的值，然后接着读取作为value的字符串。不同类型的编码不一样，根据写入时得规则解释读取到的值即可

读取到key和value后，判断下该key是否过期，如果过期则丢弃，不再导入，然后继续 类型 | 值 | 类型 | 值 的循环读取。

如果读取成功，则导入到内存，如果有过期时间则设置过期时间

如果配置了虚拟内存并且内存的使用比虚拟内存配置的大32M时，开始随机的取一些数据换出到虚拟内存中。

从上边我们也可以看到，如果没有配置虚拟内存，rdb文件导入时会尽可能地占用操作系统的内存，甚至可能全部用完。

jieforest

总结

落地存储是数据设计的一大重点也是难点。原理很简单，定义某种协议，然后按照某种协议写入读出。Redis为了节省空间和读写时的I/O操作，做了很多很细致的工作来压缩数据。另外redis的丰富的数据类型也加大了落地的实现难度。作者也曾经在他的博客说过，redis的丰富的数据类型导致了很多经典的优化办法无法在redis上实现。

jieforest

over.