渐进式解析 Redis 源码 - RDB 持久化

阅读量

相信能看到这的各位大神 肯定都知道 RDB（Redis DataBase）/ AOF（Append Only File） 两种 Redis 持久化机制，大多数也熟知两种方式的 优劣之处；但是 它们的 原理、存储方式、协议 等等方面 就不是所有人都清楚的了，所以我们接下来就分作两期 分别对 RDB 和 AOF 机制进行详细说明…

RDB（Redis DataBase）顾名思义 Redis 数据存储，将 Redis 服务器里数据以 文件的形式 存入 硬盘 加以备份

  • 优势：RDB 为一个紧致的二进制文件，保存了截止到某个点的所有数据集，极其适合备份，可以方便根据业务进行不同时机的备份，由于只有一个文件，无论是在灾备还是在恢复上都十分便利。备份的时候可以选择 fork 出子进程进行持久化操作，这样就可以极大的避免影响服务进程的 IO 操作
  • 劣势：由于 RDB 每次备份可能由于 数据集 比较大而 在子进程中 运行一段时间，所以通常都是 定时执行，但是，如果业务对数据的高可用性要求很高，甚至不能容忍几分钟少量的数据丢失，那么 RDB 恐怕不是一个好的选择，毕竟服务器一旦在定时执行之前宕机，那么这段时间的数据将丢失掉

RDB文件创建和载入

手动创建

手动操作创建 RDB文件的 有 SAVE 和 BGSAVE 两个命令

  • SAVE 命令会阻塞 Redis 服务进程，直到 RDB 文件创建完毕，在服务器阻塞期间，服务器不能处理任何命令请求
  • BGSAVE 命令不会阻塞 Redis 服务进程，不过会 fork 一个子进程，然后由子进程负责创建 RDB 文件，服务器进程继续处理命令请求；但是在执行 BGSAVE 命令过程中：
    • SAVE 命令会被服务器拒绝，服务器禁止 SAVE 和 BGSAVE 命令同时执行，是因为 服务进程 和 子进程 同时执行 rdbsave，产生竞争
    • BGSAVE 命令也会被拒绝
    • BGREWRITEAOF 命令会被延迟到 BGSAVE 执行完以后才能执行
自动创建

由于 BGSAVE 命令是不阻塞服务进程的情况下执行的，所以 Redis 允许用户通过设置服务器配置 save 选项，让服务器每隔一段时间自动执行一次 BGSAVE 命令，假设我们在 conf 文件里面配置了如下一段策略（当然，如果没有在 conf 文件里面进行设置，服务器也会默认设置下方这三条策略）：

1
2
3
save 900 1
save 300 10
save 60 10000

以上三个条件只要有任意一个符合，则会被执行，分别代表如下含义：

  • 服务器在 900秒 以内，对数据库进行了至少 1次 修改
  • 服务器在 300秒 以内，对服务器进行了至少 10次 修改
  • 服务器在 60秒 以内，对服务器进行了至少 10000次 修改

我们测试一下第三种情况，比如我们用 脚本在短时间内执行了 10000次 set 命令，那么 服务器会有如下日志显示：

1
2
3
4
5
17823:M 19 Dec 17:49:13.020 * 10000 changes in 60 seconds. Saving...
17823:M 19 Dec 17:49:13.021 * Background saving started by pid 17877
17877:C 19 Dec 17:49:13.025 * DB saved on disk
17877:C 19 Dec 17:49:13.025 * RDB: 4 MB of memory used by copy-on-write
17823:M 19 Dec 17:49:13.121 * Background saving terminated with success

那么自动执行是如何做到的呢，我们继续来看

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
// redis 服务端全局配置参数
struct redisServer {
    // ......
    // 数据保存记录，dirty 用来存储上次保存前所有数据变动的长度，用于以后判断在执行命令的过程中是否有了db中数据的变化，用得到的结果来判断要不要执行持久化操作
    long long dirty;
    // 上次持久化操作时 dirty 数据
    long long dirty_before_bgsave;
    // 保存 RDB 策略信息数组，数组每个元素都是一个 saveparam 结构，保存一组 save 选项设置
    struct saveparam *saveparams;
    // 策略个数
    int saveparamslen;
    // ......
    // 上一次保存时间
    time_t lastsave;
    // ......
};
// RDB 保存策略 结构
struct saveparam {
    // 时间要求
    time_t seconds;
    // 修改次数要求
    int changes;
};

redisServer 里面除了 saveparam 数组外，需要注意的是 还维护着 dirty 计数器 以及 lastsave 属性

  • dirty 计数器 记录距离上次成功执行 SAVE命令 或者 BGSAVE命令 之后，服务器对数据库状态（服务器中的所有数据库）进行了多少次修改（包括 写入、删除、更新 等操作）
  • lastsave 属性 为一个 UNIX时间戳，记录了服务器上一次成功执行 SAVE命令 或者 BGSAVE命令 的时间

Redis 服务器周期操作函数 serverCron() 默认每隔 100毫秒 就会执行一次，该函数用于正在运行的服务器进行维护，它们的其中一项工作就是检查 save 选项所设置的条件是否有一项被满足，如果满足的话就行执行 BGSAVE 命令

载入

与 RDB 文件的创建不同，RDB 文件的载入工作是在服务器启动的时候自动执行的，所以 Redis 并没有专门用于载入 RDB 文件的命令，只要 Redis 服务器在启动的时候检测到 RDB 文件存在，就会自动载入 RDB 文件，启动的时候会有一段这样的日志显示

1
2
3
4
5
6
[wettper@web-lovers redis-src]$ src/redis-server 
16825:C 19 Dec 17:27:46.122 # oO0OoO0OoO0Oo Redis is starting oO0OoO0OoO0Oo
16825:C 19 Dec 17:27:46.122 # Redis version=4.0.11, bits=64, commit=00000000, modified=0, pid=16825, just started
......
16825:M 19 Dec 17:27:46.125 * DB loaded from disk: 0.000 seconds
......

另外呢，因为 AOF 文件的更新频率通常比 RDB 文件要高，所以

  • 如果服务器开启了 AOF 持久化功能，那么服务器会优先使用 AOF 文件来还原数据库状态
  • 只有在 AOF 关闭的状态下，服务器才会使用 RDB 来还原数据库状态

RDB 文件结构

上面我们介绍了 Redis 服务器 RDB 文件的创建 以及 载入策略，下面我们对 RDB 文件的结构进行详细介绍
RDB 文件结构
通过上面的示例图，我们可以看到 从整体上看 RDB 文件主要包含如下几个部分

  • REDIS 标识 REDIS 字符串用于标识是 Redis 的 RDB文件
  • RDB 版本 RDB 文件的版本号，比如我们现在使用的是 0008 版本
  • 辅助信息
    • REDIS_version REDIS 版本号
    • REDIS_bits REDIS 服务器 操作系统位数
    • ctime 系统当前时间
    • used-mem REDIS 已使用的内存数
    • aof-preamble 是否为启用的 RDB+AOF 混合持久化模式，这个我们放在 AOF 结束的时候进行说明
  • databases 数据库主体 数据库主体部分由 每一个非空的数据库 相邻组成，比如 database 0 ~ database 15
    • selectdb 标识当前进行切换数据库操作
    • db_number 当前数据库 id
    • db_size 当前数据库键值对个数
    • expire_size 当前数据库过期键个数
    • 键值对数据 这个比较复杂，我们下面进行讲解
  • EOF 结束符 1 字节，标志 RDB 文件正文内容结束，键值对加载完毕，默认 RDB_OPCODE_EOF = 255
  • 校验和 8字节的 CRC64 表示的文件校验和，用来验证 文件数据内容的有效性，防篡改以及数据缺失
长度编码

在讲解下面键值对数据之前，由于 C 中对指针指向的内存是无法计算长度的，于是 我们通常都是将该段内存的大小标识出来，在 Redis 中有很多长度信息需要保存，包括下面的键值对里面涉及的所有标识长度的部分，所以 我们先普及一个 长度编码（length encoding） 的概念

那么大家就会问了，直接使用 int 存储这个长度不就完了，怎么还特地整出来一个 长度编码呢？这是因为我们平时使用到的长度可能并不大，一个 int 4个字节是不是太浪费了。所以 Redis 就采用了一个 变长的编码方式，将不同大小的数字编码成不同的长度

长度编码 的使用中，我们一般首先读取 1个字节，其中 前两位 作为编码类型的判断

  • 00 代表剩下的 6位 就直接表示具体长度
  • 01 那么就再读取后面的 1个字节 8位，加上紧挨着后续的 6位，一共 14位 来表示具体长度
  • 10 那么紧挨着后续的 6位 就废弃了，直接再读取后面的 4个直接，总共 32位 来表示具体长度
  • 11 代表一个特殊编码，紧挨着后续的 6位 用于标识 特殊编码 的种类，特殊编码主要用于 将数字存成字符串，或者编码后的字符串
    • 000000 代表 再往后 8位 表示该整数
    • 000001 代表 再往后 16位 表示该整数
    • 000010 代表 再往后 32位 表示该整数
    • 000011 代表 后续为 LZF 压缩字符串

那么这样

  • 0 – 63 的数字 只需要 1个字节 存储
  • 64 – 16383 的数字 只需要 2个字节 存储
  • 16383 - 2^32 -1 的数字 只需要用 5个字节（1个字节 的标识加 4个字节 的值）存储
键值对数据

键值对数据（key_value_pair）涉及内容比较多，我们单独进行解析

其中 expirestime_ms（过期信息标识） 和 ms（过期时间） 根据 键值对的实际设置情况，可能有，也可能没有

type 部分存储 键值对类型，key 部分使用字符串的形式存储 键的内容，我们着重讲解一下每种数据类型的具体情况

字符串类型

字符串类型的存储分为两种形式 OBJ_ENCODING_INT 和 OBJ_ENCODING_RAW

  • OBJ_ENCODING_INT 保存对象是长度不超过 32 位的整数，超过以后转换为字符串保存
  • OBJ_ENCODING_RAW
    • 长度小于等于 20 字节 使用 普通字符串直接 存储
    • 长度大于 20 字节 使用 LZF 压缩字符串 存储
      1
      2
      3
      4
      5
      6
      7
      8
      9
      10
      11
      12
      13
      // OBJ_ENCODING_INT 编码结构
      // 8位数字
      11|000000 xxxxxxxx|
      // 16位数字
      11|000001 xxxxxxxx xxxxxxxx|
      // 32位数字
      11|000010 xxxxxxxx xxxxxxxx xxxxxxxx xxxxxxxx|
      // OBJ_ENCODING_RAW
      // 普通字符串
      00|xxxxxx|字符串内容
      // LZF 压缩字符串，其中下面 长度也是符合 长度编码原理的
      11|000011|压缩后的长度|原长度|压缩后的数据
列表类型

列表类型的结构比较简单了，不过里面还是有可能有 元素比较长，类似 字符串类型一样 使用了 LZF 压缩字符串

1
2
3
4
// 普通字符串元素
list_len |len 1| value 1 | len 2 | value 2 | ... | len N | value N
// 有压缩字符串元素
list_len | len 1 | value 1 | 压缩后的长度 | 原长度 | value 2 | ... | len N | value N

集合类型

集合类型比较类似列表类型，跟之前结构图上一致

有序集合类型

有序集合类型类似列表类型，跟之前结构图上一致，不过区别的是 score 存储的时候还是按照 数字方式存储

哈希类型

哈希类型的结构也比较简单，跟之前结构图上一致

另外，这里还有一个技巧，就是使用 od 命令进行查看 rdb 文件，再结合上面的结构讲解，还是能看出来一些东西

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
[wettper@web-lovers redis-src]$ od -c dump.rdb
0000000   R   E   D   I   S   0   0   0   8 372  \t   r   e   d   i   s
0000020   -   v   e   r 006   4   .   0   .   1   1 372  \n   r   e   d
0000040   i   s   -   b   i   t   s 300   @ 372 005   c   t   i   m   e
0000060 302   * 300   %   \ 372  \b   u   s   e   d   -   m   e   m 302
0000100 230 365  \f  \0 372  \f   a   o   f   -   p   r   e   a   m   b
0000120   l   e 300  \0 376  \0 373 003  \0  \r 004   f   o   o   3 303
0000140   @   @   @   w 004   w  \0  \0  \0   Z     003  \0 006     002
0000160 037   f   o   o   3   _   1  \b 032   a   b   c   d   e   f   g
0000200   h   i   g   k   l   m   n   o   p   q   r   s   t   u   v   w
0000220   x 002   y   z 034 200   #  \0   2 340 032   #  \0   3 340 022
0000240   # 001   z 377  \0 004   f   o   o   2 032   a   b   c   d   e
0000260   f   g   h   i   g   k   l   m   n   o   p   q   r   s   t   u
0000300   v   w   x   y   z  \0 004   f   o   o   1 032   a   b   c   d
0000320   e   f   g   h   i   g   k   l   m   n   o   p   q   r   s   t
0000340   u   v   w   x   y   z 377   2   { 036 202   2   i 312 316
0000357

源码解析

由于有些方法里的代码量比较大，我们这里按照 典型的代码片段进行解析，同志们可以根据文章提示的代码位置 和 代码里面的关键词 在源码中搜素，可能数据结构一些元素 看不太懂什么意思，没关系，先混个脸熟，后面看完回头再看过来就明白了

自动保存 RDB 文件是通过系统的 自动函数serverCron() 的逻辑中进行触发

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
/* rdb.c: line 955 */
int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
    // ......
    
    // 判断后台是否正在进行 rdb 或者 aof 操作
    if (server.rdb_child_pid != -1 || server.aof_child_pid != -1 || ldbPendingChildren())
    {
        // ......
    } else {
    // 到这儿就能确定 当前木有进行 rdb 或者 aof 操作
        
        // 遍历每一个 rdb 保存条件
        for (j = 0; j < server.saveparamslen; j++) {
            struct saveparam *sp = server.saveparams+j;
            
            // 如果 数据保存记录 大于规定的修改次数 且距离 上一次保存的时间大于规定时间或者上次BGSAVE命令执行成功，才执行 BGSAVE 操作
            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(server.rdb_filename,rsiptr);
                break;
            }
         }
    // ......
}

手动 SAVE 命令，函数分别为 saveCommand() 和 bgsaveCommand()，不过具体的执行逻辑为 rdbSave()函数 和 rdbSaveBackground()函数，而且 bgsave 命令在 fork 子进程以后，依然还是使用 rdbSave() 函数进行操作

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
/* rdb.c: line 1003 */
// 手动执行 save 命令处理逻辑
int rdbSave(char *filename, rdbSaveInfo *rsi) {
    char tmpfile[256];
    // 当前工作目录
    char cwd[MAXPATHLEN];
    FILE *fp;
    rio rdb;
    int error = 0;
    // 创建以当前进程ID命名的临时文件
    snprintf(tmpfile,256,"temp-%d.rdb", (int) getpid());
    // 打开文件句柄
    fp = fopen(tmpfile,"w");
    // 如果打开文件句柄失败
    if (!fp) {
        // 获取当前工作目录
        char *cwdp = getcwd(cwd,MAXPATHLEN);
        // 记录日志
        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;
    }
    // 初始化 I/O，用于后续操作
    rioInitWithFile(&rdb,fp);
    // 执行写入操作的主逻辑
    if (rdbSaveRio(&rdb,&error,RDB_SAVE_NONE,rsi) == C_ERR) {
        errno = error;
        goto werr;
    }
    // 确定写入缓冲区里没有剩下内容
    if (fflush(fp) == EOF) goto werr;
    if (fsync(fileno(fp)) == -1) goto werr;
    if (fclose(fp) == EOF) goto werr;
    // 重新命名 rdb 文件，把之前临时的名称修改为正式的 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);
    // 最后一次 bgsave 的状态置位 成功
    server.lastbgsave_status = C_OK;
    return C_OK;
werr:
    // 日志报错
    serverLog(LL_WARNING,"Write error saving DB on disk: %s", strerror(errno));
    // 关闭句柄
    fclose(fp);
    // 删除临时文件
    unlink(tmpfile);
    return C_ERR;
}
// 自动执行 save 命令处理逻辑
int rdbSaveBackground(char *filename, rdbSaveInfo *rsi) {
    pid_t childpid;
    long long start;
    // 检查后台是否正在执行 aof 或者 rdb 操作
    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);
    // 打开子进程通讯管道
    openChildInfoPipe();
    start = ustime();
    // 打开子进程
    if ((childpid = fork()) == 0) {
    // 子进程操作
        int retval;
        // 关闭子进程继承的 socket 监听
        closeListeningSockets(0);
        // 子进程 title 修改
        redisSetProcTitle("redis-rdb-bgsave");
        // 执行rdb 写入操作
        retval = rdbSave(filename,rsi);
        // 执行完毕以后
        if (retval == C_OK) {
            // 当前操作的内存统计
            size_t private_dirty = zmalloc_get_private_dirty(-1);
            if (private_dirty) {
                serverLog(LL_NOTICE,
                    "RDB: %zu MB of memory used by copy-on-write",
                    private_dirty/(1024*1024));
            }
            server.child_info_data.cow_size = private_dirty;
            sendChildInfo(CHILD_INFO_TYPE_RDB);
        }
        // 退出子进程
        exitFromChild((retval == C_OK) ? 0 : 1);
    } else {
    // 父进程操作
    
        // fork 统计时间
        server.stat_fork_time = ustime()-start;
        // 每秒操作情况
        server.stat_fork_rate = (double) zmalloc_used_memory() * 1000000 / server.stat_fork_time / (1024*1024*1024);
        // 记录将延迟事件和延迟时间关联到延迟诊断的字典中
        latencyAddSampleIfNeeded("fork",server.stat_fork_time/1000);
        // 如果创建子进程失败
        if (childpid == -1) {
            // 关闭进程通讯管道
            closeChildInfoPipe();
            // 上次 bgsave 保存状态置位 失败
            server.lastbgsave_status = C_ERR;
            // 记录日志
            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);
        // rdb 保存开始时间
        server.rdb_save_time_start = time(NULL);
        // bgsave 子进程
        server.rdb_child_pid = childpid;
        // 子进程方式
        server.rdb_child_type = RDB_CHILD_TYPE_DISK;
        // 关闭哈希表的 resize，因为 resize 过程中会有复制拷贝动作
        updateDictResizePolicy();
        return C_OK;
    }
    return C_OK;
}

根据上面的逻辑，rdb 保存的底层逻辑还需要继续看 rdbSaveRio() 函数

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
/* rdb.c: line 882 */
int rdbSaveRio(rio *rdb, int *error, int flags, rdbSaveInfo *rsi) {
    dictIterator *di = NULL;
    dictEntry *de;
    char magic[10];
    int j;
    uint64_t cksum;
    size_t processed = 0;
    // 设置文件校验和数据
    if (server.rdb_checksum)
        rdb->update_cksum = rioGenericUpdateChecksum;
    // 组合头标识内容：REDIS + RDB文件版本
    snprintf(magic,sizeof(magic),"REDIS%04d",RDB_VERSION);
    // 写入 REDIS文件标识和RDB版本号
    if (rdbWriteRaw(rdb,magic,9) == -1) goto werr;
    // 写入辅助信息（REDIS版本、服务器操作系统位数、当前时间 等等数据）
    if (rdbSaveInfoAuxFields(rdb,flags,rsi) == -1) goto werr;
    // 遍历每一个数据库，逐个数据库数据保存
    for (j = 0; j < server.dbnum; j++) {
        // 获取数据库指针地址，因为 server 存储了第一个数据库的指针地址，而且数据库指针地址是连续的
        redisDb *db = server.db+j;
        // 数据库字典
        dict *d = db->dict;
        // 跳过空数据库
        if (dictSize(d) == 0) continue;
        // 初始化数据库迭代器，方便遍历数据库所有数据
        di = dictGetSafeIterator(d);
        if (!di) return C_ERR;
        // 写入数据库部分的开始标识
        if (rdbSaveType(rdb,RDB_OPCODE_SELECTDB) == -1) goto werr;
        // 写入当前数据库号
        if (rdbSaveLen(rdb,j) == -1) goto werr;
        // 获取数据库字典的大小和过期键字典的大小
        // 为了编码方便这些大小最大为UINT32_MAX，但是并不影响数据库和过期键的实际大小，因为此大小只是在加载rdb数据的时候申请哈希表的初始大小
        uint32_t db_size, expires_size;
        // 数据库字典大小，也就是键值对个数，为了编码方便这些大小最大为UINT32_MAX，但是并不影响表示数据库实际大小，因为此大小只是在加载 rdb 数据的时候申请哈希表的初始大小
        db_size = (dictSize(db->dict) <= UINT32_MAX) ?
                                dictSize(db->dict) :
                                UINT32_MAX;
        // 数据库过期键字典大小，也就是过期键值对个数，为了编码方便这些大小最大为UINT32_MAX，但是并不影响表示过期键实际大小，因为此大小只是在加载 rdb 数据的时候申请哈希表的初始大小
        expires_size = (dictSize(db->expires) <= UINT32_MAX) ?
                                dictSize(db->expires) :
                                UINT32_MAX;
        // 写入当前待写入数据的类型，此处为 RDB_OPCODE_RESIZEDB，用来载入的时候标识 后续的两种 大小数据，方便申请 哈希表初始大小
        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;
        // 遍历键值对
        while((de = dictNext(di)) != NULL) {
            // 获取键
            sds keystr = dictGetKey(de);
            // 值对象
            robj key, *o = dictGetVal(de);
            long long expire;
            // 初始化 key，因为操作的是 key 字符串对象，而不是直接操作 键的字符串内容
            initStaticStringObject(key,keystr);
            // 获取键的过期数据
            expire = getExpire(db,&key);
            // 保存键值对数据
            if (rdbSaveKeyValuePair(rdb,&key,o,expire) == -1) goto werr;
            // 说明是在 aofrewrite 且开启了 RDB-AOF混合开关 此时就会从父进程去读取增量数据了，需要进行一些特殊处理，我们在 RDB-AOF混合模式的时候详细说，这里挖个坑
            if (flags & RDB_SAVE_AOF_PREAMBLE &&
                rdb->processed_bytes > processed+AOF_READ_DIFF_INTERVAL_BYTES)
            {
                processed = rdb->processed_bytes;
                aofReadDiffFromParent();
            }
        }
        // 释放迭代器
        dictReleaseIterator(di);
    }
    // 先不重置 迭代器变量，下次循环还需要使用
    di = NULL;
    // 如果有相关lua脚本信息，这些信息也需要保存
    if (rsi && dictSize(server.lua_scripts)) {
        di = dictGetIterator(server.lua_scripts);
        // 遍历所有 lua脚本
        while((de = dictNext(di)) != NULL) {
            // 获取脚本内容
            robj *body = dictGetVal(de);
            // 保存脚本内容信息
            if (rdbSaveAuxField(rdb,"lua",3,body->ptr,sdslen(body->ptr)) == -1)
                goto werr;
        }
        dictReleaseIterator(di);
    }
    // 写入结束符
    if (rdbSaveType(rdb,RDB_OPCODE_EOF) == -1) goto werr;
    // 写入CRC64校验和
    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;
}

上面我们看到了一些地方调用 rdbSaveType() 和 rdbWriteRaw() ，不过具体逻辑还是 rioWrite()函数，这个函数主要是 rdb 的 io 操作，用来写入 rdb 文件

写入辅助信息函数 rdbSaveInfoAuxFields()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
/* rdb.c: line 851 */
int rdbSaveInfoAuxFields(rio *rdb, int flags, rdbSaveInfo *rsi) {
    // REDIS 服务器位数
    int redis_bits = (sizeof(void*) == 8) ? 64 : 32;
    // 是否开启 RDB-AOF 混合模式
    int aof_preamble = (flags & RDB_SAVE_AOF_PREAMBLE) != 0;
    // 写入 redis 版本号，版本号内容为字符串类型写入
    if (rdbSaveAuxFieldStrStr(rdb,"redis-ver",REDIS_VERSION) == -1) return -1;
    //  写入 redis 服务器位数信息，位数信息内容为整数类型写入
    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;
    // 写入 从库信息
    if (rsi) {
        if (rdbSaveAuxFieldStrInt(rdb,"repl-stream-db",rsi->repl_stream_db)
            == -1) return -1;
        if (rdbSaveAuxFieldStrStr(rdb,"repl-id",server.replid)
            == -1) return -1;
        if (rdbSaveAuxFieldStrInt(rdb,"repl-offset",server.master_repl_offset)
            == -1) return -1;
    }
    // 写入 是否开启 RDB-AOF 混合模式 标识
    if (rdbSaveAuxFieldStrInt(rdb,"aof-preamble",aof_preamble) == -1) return -1;
    return 1;
}

rdbSaveRawString() 函数主要逻辑为 将 字符串对象以 len+data 形式写入 RDB 文件

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
/* rdb.c: line 375 */
ssize_t rdbSaveRawString(rio *rdb, unsigned char *s, size_t len) {
    int enclen;
    ssize_t n, nwritten = 0;
    // 如果长度小于11，尝试看能不能转成 小整数 形式，以减小存储空间
    if (len <= 11) {
        unsigned char buf[5];
        // 尝试进行 转 int 操作
        if ((enclen = rdbTryIntegerEncoding((char*)s,len,buf)) > 0) {
            // 写入转成功的内容
            if (rdbWriteRaw(rdb,buf,enclen) == -1) return -1;
            return enclen;
        }
    }
    // 如果已经开启了压缩功能并且字符串长度大于 20，则以 LZF 压缩字符串的形式保存
    if (server.rdb_compression && len > 20) {
        // 以 LZF 压缩字符串的形式保存
        n = rdbSaveLzfStringObject(rdb,s,len);
        if (n == -1) return -1;
        if (n > 0) return n;
    }
    // 写入字符串长度
    if ((n = rdbSaveLen(rdb,len)) == -1) return -1;
    nwritten += n;
    if (len > 0) {
        // 写入字符串内容
        if (rdbWriteRaw(rdb,s,len) == -1) return -1;
        nwritten += len;
    }
    return nwritten;
}

压缩字符串保存函数 rdbSaveLzfStringObject()，而在经过压缩过的字符串内容的保存逻辑 rdbSaveLzfBlob() 函数

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
/* rdb.c: line 286 */
// 写入压缩内容
ssize_t rdbSaveLzfBlob(rio *rdb, void *data, size_t compress_len, size_t original_len) {
    unsigned char byte;
    ssize_t n, nwritten = 0;
    // 数据已经被压缩
    byte = (RDB_ENCVAL<<6)|RDB_ENC_LZF;
    // 写入 压缩字符串的 标识
    if ((n = rdbWriteRaw(rdb,&byte,1)) == -1) goto writeerr;
    nwritten += n;
    // 写入 压缩后长度
    if ((n = rdbSaveLen(rdb,compress_len)) == -1) goto writeerr;
    nwritten += n;
    // 写入 原始数据长度
    if ((n = rdbSaveLen(rdb,original_len)) == -1) goto writeerr;
    nwritten += n;
    // 写入 压缩后的数据
    if ((n = rdbWriteRaw(rdb,data,compress_len)) == -1) goto writeerr;
    nwritten += n;
    return nwritten;
writeerr:
    return -1;
}
// 压缩字符串保存入口函数
ssize_t rdbSaveLzfStringObject(rio *rdb, unsigned char *s, size_t len) {
    size_t comprlen, outlen;
    void *out;
    // 要求长度必须 大于4，否则不值得压缩
    if (len <= 4) return 0;
    outlen = len-4;
    // 内存申请失败
    if ((out = zmalloc(outlen+1)) == NULL) return 0;
    // LZF 算法压缩
    comprlen = lzf_compress(s, len, out, outlen);
    if (comprlen == 0) {
        zfree(out);
        return 0;
    }
    // 写入 压缩内容
    ssize_t nwritten = rdbSaveLzfBlob(rdb, out, comprlen, len);
    zfree(out);
    return nwritten;
}

长度编码 写入函数 rdbSaveLen()，跟这个函数逻辑比较类似的为 整数长度编码函数rdbEncodeInteger() ，逻辑也比较简单，看这段的时候可以参考上面的 长度编码部分

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
/* rdb.c: line 110 */
int rdbSaveLen(rio *rdb, uint64_t len) {
    unsigned char buf[2];
    size_t nwritten;
    // 如果长度小于 64
    if (len < (1<<6)) {
        // 获取长度编码，前两位为 00
        buf[0] = (len&0xFF)|(RDB_6BITLEN<<6);
        // 写入长度编码
        if (rdbWriteRaw(rdb,buf,1) == -1) return -1;
        nwritten = 1;
    } else if (len < (1<<14)) {
    // 长度小于 16384
        // 获取长度编码，前两位为 01
        buf[0] = ((len>>8)&0xFF)|(RDB_14BITLEN<<6);
        buf[1] = len&0xFF;
        // 写入长度编码
        if (rdbWriteRaw(rdb,buf,2) == -1) return -1;
        nwritten = 2;
    } else if (len <= UINT32_MAX) {
    // 长度小于 4294967295
        // 获取长度编码，前几位为 RDB_32BITLEN=10000000
        buf[0] = RDB_32BITLEN;
        // 写入长度编码标识
        if (rdbWriteRaw(rdb,buf,1) == -1) return -1;
        // 后续 32位 存储具体长度
        uint32_t len32 = htonl(len);
        // 写入后续的 32位 数据
        if (rdbWriteRaw(rdb,&len32,4) == -1) return -1;
        nwritten = 1+4;
    } else {
        // 获取长度编码，前几位为 RDB_64BITLEN=10000001
        buf[0] = RDB_64BITLEN;
        // 写入长度编码标识
        if (rdbWriteRaw(rdb,buf,1) == -1) return -1;
        // 后续 64位 存储具体长度
        len = htonu64(len);
        // 写入后续的 64位 数据
        if (rdbWriteRaw(rdb,&len,8) == -1) return -1;
        nwritten = 1+8;
    }
    return nwritten;
}

保存逻辑继续走下去，直到键值对保存的入口函数 rdbSaveKeyValuePair()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
/* rdb.c: line 810 */
int rdbSaveKeyValuePair(rio *rdb, robj *key, robj *val, long long expiretime)
{
    // 如果有过期信息
    if (expiretime != -1) {
        // 保存过期信息标识
        if (rdbSaveType(rdb,RDB_OPCODE_EXPIRETIME_MS) == -1) return -1;
        // 保存过期具体数据内容
        if (rdbSaveMillisecondTime(rdb,expiretime) == -1) return -1;
    }
    // 保存键值对 类型的标识
    if (rdbSaveObjectType(rdb,val) == -1) return -1;
    // 保存键值对 键的内容
    if (rdbSaveStringObject(rdb,key) == -1) return -1;
    // 保存键值对 值的内容
    if (rdbSaveObject(rdb,val) == -1) return -1;
    return 1;
}

保存键值对 类型标识函数 rdbSaveObjectType()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
/* rdb.c: line 594 */
int rdbSaveObjectType(rio *rdb, robj *o) {
    switch (o->type) {
    // 字符串类型
    case OBJ_STRING:
        return rdbSaveType(rdb,RDB_TYPE_STRING);
    // 列表类型
    case OBJ_LIST:
        // 列表类型的编码只有 quicklist
        if (o->encoding == OBJ_ENCODING_QUICKLIST)
            return rdbSaveType(rdb,RDB_TYPE_LIST_QUICKLIST);
        else
            serverPanic("Unknown list encoding");
    // 集合类型
    case OBJ_SET:
        // 整数集合编码
        if (o->encoding == OBJ_ENCODING_INTSET)
            return rdbSaveType(rdb,RDB_TYPE_SET_INTSET);
        // hashtable 编码
        else if (o->encoding == OBJ_ENCODING_HT)
            return rdbSaveType(rdb,RDB_TYPE_SET);
        else
            serverPanic("Unknown set encoding");
    // 有序集合类型
    case OBJ_ZSET:
        // 压缩列表编码
        if (o->encoding == OBJ_ENCODING_ZIPLIST)
            return rdbSaveType(rdb,RDB_TYPE_ZSET_ZIPLIST);
        // 跳跃表编码
        else if (o->encoding == OBJ_ENCODING_SKIPLIST)
            return rdbSaveType(rdb,RDB_TYPE_ZSET_2);
        else
            serverPanic("Unknown sorted set encoding");
    // 哈希类型
    case OBJ_HASH:
        // 压缩列表编码
        if (o->encoding == OBJ_ENCODING_ZIPLIST)
            return rdbSaveType(rdb,RDB_TYPE_HASH_ZIPLIST);
        // hashtable 编码
        else if (o->encoding == OBJ_ENCODING_HT)
            return rdbSaveType(rdb,RDB_TYPE_HASH);
        else
            serverPanic("Unknown hash encoding");
    // 模块类型，redis 4 开始引入了模块化机制
    case OBJ_MODULE:
        return rdbSaveType(rdb,RDB_TYPE_MODULE_2);
    default:
        serverPanic("Unknown object type");
    }
    return -1; /* avoid warning */
}

键的保存是以字符串的形式进行的，这个我们跟下面的 字符串类型值保存函数 一块讲解

我们先看一下 值的保存逻辑，函数为 rdbSaveObject()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
/* rdb.c: line 643 */
ssize_t rdbSaveObject(rio *rdb, robj *o) {
    ssize_t n = 0, nwritten = 0;
    // 字符串类型
    if (o->type == OBJ_STRING) {
        if ((n = rdbSaveStringObject(rdb,o)) == -1) return -1;
        nwritten += n;
    } else if (o->type == OBJ_LIST) {
    // 列表类型
    
        // 列表类型只有压缩列表编码
        if (o->encoding == OBJ_ENCODING_QUICKLIST) {
            quicklist *ql = o->ptr;
            quicklistNode *node = ql->head;
            // 写入列表长度
            if ((n = rdbSaveLen(rdb,ql->len)) == -1) return -1;
            nwritten += n;
            
            // 遍历列表中的每个元素
            while(node) {
                // 尝试压缩元素
                if (quicklistNodeIsCompressed(node)) {
                    void *data;
                    // 获取压缩元素数据以及压缩长度
                    size_t compress_len = quicklistGetLzf(node, &data);
                    // 写入 压缩长度、原始长度、压缩数据内容
                    if ((n = rdbSaveLzfBlob(rdb,data,compress_len,node->sz)) == -1) return -1;
                    nwritten += n;
                } else {
                    // 不能压缩元素的话以字符串的形式写入
                    if ((n = rdbSaveRawString(rdb,node->zl,node->sz)) == -1) return -1;
                    nwritten += n;
                }
                node = node->next;
            }
        } else {
            serverPanic("Unknown list encoding");
        }
    } else if (o->type == OBJ_SET) {
    // 集合类型
        
        // hashtable 编码
        if (o->encoding == OBJ_ENCODING_HT) {
            dict *set = o->ptr;
            // 集合迭代器
            dictIterator *di = dictGetIterator(set);
            dictEntry *de;
            // 写入集合长度
            if ((n = rdbSaveLen(rdb,dictSize(set))) == -1) return -1;
            nwritten += n;
            // 遍历集合元素
            while((de = dictNext(di)) != NULL) {
                sds ele = dictGetKey(de);
                // 以字符串的形式写入
                if ((n = rdbSaveRawString(rdb,(unsigned char*)ele,sdslen(ele)))
                    == -1) return -1;
                nwritten += n;
            }
            dictReleaseIterator(di);
        } else if (o->encoding == OBJ_ENCODING_INTSET) {
        // 整数集合
        
            // 直接将 INTSET 转换成字符串对象存放
            size_t l = intsetBlobLen((intset*)o->ptr);
            // 以字符串的形式写入
            if ((n = rdbSaveRawString(rdb,o->ptr,l)) == -1) return -1;
            nwritten += n;
        } else {
            serverPanic("Unknown set encoding");
        }
    } else if (o->type == OBJ_ZSET) {
    // 有序集合类型
    
        // 压缩列表编码
        if (o->encoding == OBJ_ENCODING_ZIPLIST) {
            // 直接将 ziplist 转换成字符串对象存放
            size_t l = ziplistBlobLen((unsigned char*)o->ptr);
            // 以字符串的形式写入
            if ((n = rdbSaveRawString(rdb,o->ptr,l)) == -1) return -1;
            nwritten += n;
        } else if (o->encoding == OBJ_ENCODING_SKIPLIST) {
        // 跳跃表编码
            zset *zs = o->ptr;
            zskiplist *zsl = zs->zsl;
            // 写入集合长度
            if ((n = rdbSaveLen(rdb,zsl->length)) == -1) return -1;
            nwritten += n;
            // 遍历有序集合元素
            zskiplistNode *zn = zsl->tail;
            while (zn != NULL) {
                // 写入元素key
                if ((n = rdbSaveRawString(rdb,
                    (unsigned char*)zn->ele,sdslen(zn->ele))) == -1)
                {
                    return -1;
                }
                nwritten += n;
                // 将元素分值转换成 double 类型，然后以 字符串的形式写入
                if ((n = rdbSaveBinaryDoubleValue(rdb,zn->score)) == -1)
                    return -1;
                nwritten += n;
                zn = zn->backward;
            }
        } else {
            serverPanic("Unknown sorted set encoding");
        }
    } else if (o->type == OBJ_HASH) {
    // 哈希类型
    
        // 压缩列表编码
        if (o->encoding == OBJ_ENCODING_ZIPLIST) {
            // 直接将 ziplist 转换成字符串对象存放
            size_t l = ziplistBlobLen((unsigned char*)o->ptr);
            // 以字符串的形式写入
            if ((n = rdbSaveRawString(rdb,o->ptr,l)) == -1) return -1;
            nwritten += n;
        } else if (o->encoding == OBJ_ENCODING_HT) {
        // hashtable 编码
            // hashtable 迭代器
            dictIterator *di = dictGetIterator(o->ptr);
            dictEntry *de;
            // 写入哈希表长度
            if ((n = rdbSaveLen(rdb,dictSize((dict*)o->ptr))) == -1) return -1;
            nwritten += n;
            // 遍历所有 hashtable 元素
            while((de = dictNext(di)) != NULL) {
                // 获取哈希数据的 key
                sds field = dictGetKey(de);
                // 获取哈希数据的 value
                sds value = dictGetVal(de);
                // 以字符串的形式写入 key 数据
                if ((n = rdbSaveRawString(rdb,(unsigned char*)field,
                        sdslen(field))) == -1) return -1;
                nwritten += n;
                // 以字符串的形式写入 value 数据
                if ((n = rdbSaveRawString(rdb,(unsigned char*)value,
                        sdslen(value))) == -1) return -1;
                nwritten += n;
            }
            dictReleaseIterator(di);
        } else {
            serverPanic("Unknown hash encoding");
        }
    } else if (o->type == OBJ_MODULE) {
    // 模块类型，redis 4 开始引入了模块化机制
        RedisModuleIO io;
        moduleValue *mv = o->ptr;
        moduleType *mt = mv->type;
        moduleInitIOContext(io,mt,rdb);
        // 写入 模块长度
        int retval = rdbSaveLen(rdb,mt->id);
        if (retval == -1) return -1;
        io.bytes += retval;
        // 写入 模块内容数据
        mt->rdb_save(&io,mv->value);
        // 写入 模块结束符
        retval = rdbSaveLen(rdb,RDB_MODULE_OPCODE_EOF);
        if (retval == -1) return -1;
        io.bytes += retval;
        if (io.ctx) {
            moduleFreeContext(io.ctx);
            zfree(io.ctx);
        }
        return io.error ? -1 : (ssize_t)io.bytes;
    } else {
        serverPanic("Unknown object type");
    }
    return nwritten;
}

字符串形式保存函数 rdbSaveStringObject()

1
2
3
4
5
6
7
8
9
10
11
12
13
/* rdb.c: line 427 */
ssize_t rdbSaveStringObject(rio *rdb, robj *obj) {
    // 整数编码
    if (obj->encoding == OBJ_ENCODING_INT) {
        // 整数长度编码 写入
        return rdbSaveLongLongAsStringObject(rdb,(long)obj->ptr);
    } else {
        serverAssertWithInfo(NULL,obj,sdsEncodedObject(obj));
        // 字符串长度写入
        return rdbSaveRawString(rdb,obj->ptr,sdslen(obj->ptr));
    }
}

RDB 文件数据加载逻辑为 上面写入逻辑 的逆操作，入口为 redis server 启动流程中的 loadDataFromDisk() 函数，主要的操作逻辑为 rdbLoadRio()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
/* rdb.c: line 1506 */
int rdbLoadRio(rio *rdb, rdbSaveInfo *rsi, int loading_aof) {
    uint64_t dbid;
    int type, rdbver;
    redisDb *db = server.db+0;
    char buf[1024];
    long long expiretime, now = mstime();
    // 校验和处理函数
    rdb->update_cksum = rdbLoadProgressCallback;
    // 单次进程内存块大小
    rdb->max_processing_chunk = server.loading_process_events_interval_bytes;
    // 开头文件标识
    if (rioRead(rdb,buf,9) == 0) goto eoferr;
    buf[9] = '\0';
    // 判断开头是不是 REDIS
    if (memcmp(buf,"REDIS",5) != 0) {
        serverLog(LL_WARNING,"Wrong signature trying to load DB from file");
        errno = EINVAL;
        return C_ERR;
    }
    // RDB 文件版本
    rdbver = atoi(buf+5);
    // 判断版本是否在可以解析的范围内，一般是向下兼容，比如当前版本的 redis 中 RDB_VERSION=8，那么当前 redis 服务 可以解析  1~8 之间的 rdb 版本
    if (rdbver < 1 || rdbver > RDB_VERSION) {
        serverLog(LL_WARNING,"Can't handle RDB format version %d",rdbver);
        errno = EINVAL;
        return C_ERR;
    }
    // 遍历 RDB 文件
    while(1) {
        robj *key, *val;
        expiretime = -1;
        // 读取当前 type 类型字段
        if ((type = rdbLoadType(rdb)) == -1) goto eoferr;
        // 如果 type 标识为 秒级过期时间信息
        if (type == RDB_OPCODE_EXPIRETIME) {
            // 读取键值对过期时间，单位为 秒
            if ((expiretime = rdbLoadTime(rdb)) == -1) goto eoferr;
            // 获取 value 类型，方便后续解析
            if ((type = rdbLoadType(rdb)) == -1) goto eoferr;
            // 转换为 毫秒级
            expiretime *= 1000;
        } else if (type == RDB_OPCODE_EXPIRETIME_MS) {
        // type 标识为 毫秒级过期时间信息
            // 读取键值对过期时间，单位为 毫秒
            if ((expiretime = rdbLoadMillisecondTime(rdb)) == -1) goto eoferr;
            // 获取 value 类型，方便后续解析
            if ((type = rdbLoadType(rdb)) == -1) goto eoferr;
        } else if (type == RDB_OPCODE_EOF) {
            // 结束符，遍历完毕
            break;
        } else if (type == RDB_OPCODE_SELECTDB) {
        // type 标识为 数据库选择
            // 获取数据库 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;
            continue;
        } else if (type == RDB_OPCODE_RESIZEDB) {
        // type 标识为 数据库大小
        
            uint64_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);
            continue;
        } else if (type == RDB_OPCODE_AUX) {
        // type 标识为 辅助信息
            
            robj *auxkey, *auxval;
            // 获取辅助信息 key
            if ((auxkey = rdbLoadStringObject(rdb)) == NULL) goto eoferr;
            // 获取辅助信息 value
            if ((auxval = rdbLoadStringObject(rdb)) == NULL) goto eoferr;
            if (((char*)auxkey->ptr)[0] == '%') {
                serverLog(LL_NOTICE,"RDB '%s': %s",
                    (char*)auxkey->ptr,
                    (char*)auxval->ptr);
            } else if (!strcasecmp(auxkey->ptr,"repl-stream-db")) {
            // 从库信息
                if (rsi) rsi->repl_stream_db = atoi(auxval->ptr);
            } else if (!strcasecmp(auxkey->ptr,"repl-id")) {
            // 从库 id
                if (rsi && sdslen(auxval->ptr) == CONFIG_RUN_ID_SIZE) {
                    memcpy(rsi->repl_id,auxval->ptr,CONFIG_RUN_ID_SIZE+1);
                    rsi->repl_id_is_set = 1;
                }
            } else if (!strcasecmp(auxkey->ptr,"repl-offset")) {
            // 从库同步偏移
                if (rsi) rsi->repl_offset = strtoll(auxval->ptr,NULL,10);
            } else if (!strcasecmp(auxkey->ptr,"lua")) {
            // lua 信息
                if (luaCreateFunction(NULL,server.lua,auxval) == NULL) {
                    rdbExitReportCorruptRDB(
                        "Can't load Lua script from RDB file! "
                        "BODY: %s", auxval->ptr);
                }
            } else {
                serverLog(LL_DEBUG,"Unrecognized RDB AUX field: '%s'",
                    (char*)auxkey->ptr);
            }
            decrRefCount(auxkey);
            decrRefCount(auxval);
            continue; /* Read type again. */
        }
        // 读取 键
        if ((key = rdbLoadStringObject(rdb)) == NULL) goto eoferr;
        // 读取值
        if ((val = rdbLoadObject(type,rdb)) == NULL) goto eoferr;
        // 如果键值对过期
        if (server.masterhost == NULL && !loading_aof && expiretime != -1 && expiretime < now) {
            // 引用计数 -1
            decrRefCount(key);
            decrRefCount(val);
            continue;
        }
        // 添加键值对到数据库
        dbAdd(db,key,val);
        // 如果需要，设置键值对 过期数据
        if (expiretime != -1) setExpire(NULL,db,key,expiretime);
    
        decrRefCount(key);
    }
    // RDB 文件版本 >=5 的时候开始验证 校验和数据
    if (rdbver >= 5) {
        uint64_t cksum, expected = rdb->cksum;
        // 获取校验和数据
        if (rioRead(rdb,&cksum,8) == 0) goto eoferr;
        // 验证校验和数据
        if (server.rdb_checksum) {
            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");
            }
        }
    }
    return C_OK;
eoferr:
    serverLog(LL_WARNING,"Short read or OOM loading DB. Unrecoverable error, aborting now.");
    rdbExitReportCorruptRDB("Unexpected EOF reading RDB file");
    return C_ERR;
}

本文作者： wettper
本文链接： http://www.web-lovers.com/redis-source-rdb.html
版权声明： 本博客所有文章除特别声明外，均采用 CC BY-NC-SA 3.0 许可协议。转载请注明出处！

赏

谢谢你请我吃糖果

支付宝
微信

扫一扫，分享到微信

微信分享二维码

tag:

    缺失模块。
    1、请确保node版本大于6.2
    2、在博客根目录（注意不是yilia根目录）执行以下命令：
    npm i hexo-generator-json-content --save

    3、在根目录_config.yml里添加配置： 

      jsonContent:
    meta: false
    pages: false
    posts:
      title: true
      date: true
      path: true
      text: false
      raw: false
      content: false
      slug: false
      updated: false
      comments: false
      link: false
      permalink: false
      excerpt: false
      categories: false
      tags: true

wettper <br /><br /> 现 离职状态 程序猿 <br /><br /> 叫醒我的不止阳光，还有梦想！<br /><br /> 喵喵、念念、健身、吉他！