redis源码阅读笔记(9)——RDB，AOF持久化

玩龙天子

1. 持久化
Redis提供了两种持久化数据到硬盘的方式。
RDB:数据库里所有记录的一个快照
AOF(append only file):原汁原味地记录了每次操作命令的历史记录，相当于一个log
如果还没了解过持久化功能的话，请先阅读Redis官网上的persistence手册中文翻译版。
RDB以二进制方式存储，每条记录只记一次，文件大小更紧凑，这样恢复起来更快。但是写这个RDB需要时间，这样会造成宕机后数据丢失。
AOF以文本文件方式存储，把每条命令都记下来，会有冗余，但是写aof快，只要往文件末尾追加记录即可，可以使得宕机后可以恢复更多的数据。
接下来可以阅读Redis 设计与实现里的RDB和AOF部分。
2. RDB的save
    SAVE 直接调用 rdbSave ，阻塞 Redis 主进程，直到保存完成为止。
    BGSAVE 则 fork 出一个子进程，子进程调用 rdbSave ，并在保存完成之后向主进程发送信号，通知保存已完成。
2.1 redis代码核心点
我们看下BGSAVE的代码,主要关注如何fork子进程

//简化以后的流程
int rdbSaveBackground(char *filename) {
pid_t childpid;
long long start;
// 如果 BGSAVE 已经在执行，那么出错
if (server.rdb_child_pid != -1) return REDIS_ERR;
// ......
if ((childpid = fork()) == 0) {
int retval;
/* Child */
// 执行保存操作
retval = rdbSave(filename);
// ......
// 向父进程发送信号
exitFromChild((retval == REDIS_OK) ? 0 : 1);
} else {
/* Parent */
// 如果 fork() 出错，那么报告错误
if (childpid == -1) {
return REDIS_ERR;
}
// 打印 BGSAVE 开始的日志
redisLog(REDIS_NOTICE,"Background saving started by pid %d",childpid);
// 记录负责执行 BGSAVE 的子进程 ID
server.rdb_child_pid = childpid;
return REDIS_OK;
}
return REDIS_OK; /* unreached */
}
void exitFromChild(int retcode) {
_exit(retcode); //_exit调用后，会发信号给父进程
}

解释一下，父进程fork出一个子进程。
子进程如果做完了，则调用_exit函数，即可通知到父进程。
如下代码则是父进程如何接收信号

/*  处理 BGSAVE 完成时发送的信号 */
void backgroundSaveDoneHandler(int exitcode, int bysignal) {
// BGSAVE 成功
if (!bysignal && exitcode == 0) {
redisLog(REDIS_NOTICE,
"Background saving terminated with success");
//......
}
// 更新服务器状态
server.rdb_child_pid = -1;
}
//简化以后的流程, 这是 Redis 的时间中断器，每秒调用 server.hz 次。
int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
pid_t pid;
// 接收子进程发来的信号，非阻塞
if ((pid = wait3(&statloc,WNOHANG,NULL)) != 0) {
// BGSAVE 执行完毕
if (pid == server.rdb_child_pid) {
backgroundSaveDoneHandler(exitcode,bysignal);
// BGREWRITEAOF 执行完毕
} else if (pid == server.aof_child_pid) {
backgroundRewriteDoneHandler(exitcode,bysignal);
}
}
}


以上代码便是核心关键，serverCron可以暂且理解成一个定时要执行的函数，wait3函数意思可以理解成查看一下子进程有没有完成，有了则调用backgroundSaveDoneHandler，没有的话不阻塞，继续干自己的活，过一段时间再来查看子进程是否完成。
如果看不懂，需要补习一下c的进程交互的知识（fork，wait，_exit）。
3. C语言里的fork函数
c语言:fork函数详解
3.1 写个程序进一步理解fork，wait，_exit

#include <stdio.h>
#include <unistd.h>
#include <sys/wait.h>
int main (void)
{
pid_t fpid; //fpid表示fork函数返回的值
int count=0;
fpid=fork();
if (fpid < 0)
printf("error in fork!");
else if (fpid == 0)
{
printf("i am the child process, my process id is %d, parent process id is %d\n",getpid(), getppid());
printf("我是儿子\n");
fflush(stdout);
count++;
sleep(10);
}
else
{
printf("i am the parent process, my process id is %d, parent process id is %d\n",getpid(), getppid());
printf("我是孩子他爹\n");
fflush(stdout);
count++;
int statloc;
pid_t pid;
while (1) {
if ((pid = wait3(&statloc,1,NULL)) != 0) {
//if ((pid = wait(&statloc)) != 0) {
printf("收到儿子 %d的信号\n",pid);
fflush(stdout);
break;
} else {
printf("还没收到儿子 的信号\n");
fflush(stdout);
sleep(2);
}
}
}
printf("统计结果是: %d\n",count);
fflush(stdout);
return 0;
}


运行结果如下

i am the child process, my process id is 4260, parent process id is 6660
我是儿子
i am the parent process, my process id is 6660, parent process id is 1
我是孩子他爹
还没收到儿子 的信号
还没收到儿子 的信号
还没收到儿子 的信号
还没收到儿子 的信号
还没收到儿子 的信号
统计结果是: 1
还没收到儿子 的信号
收到儿子 4260的信号
统计结果是: 1

可以看到用了wait3的话，父亲每隔2秒看一下孩子，如果没有音讯，可以做自己的事情(非阻塞的)，过一会儿再来看看。
而如果把wait3改成wait的话（把代码里wait3那行注释掉，wait那行注释去掉）
运行结果如下

i am the child process, my process id is 10904, parent process id is 9056
我是儿子
i am the parent process, my process id is 9056, parent process id is 1
我是孩子他爹
统计结果是: 1
收到儿子 10904的信号
统计结果是: 1

可以看到父亲傻傻地等待孩子的音讯（阻塞的）等了10秒，这之间他不能做别的事情。
4. RDB和AOF的格式
这一部分自己跑个实例，然后打开dump.rdb和appendonly.aof，对照网上的资料或是协议分析一下就可以了，比较枯燥，不再赘述。
5. LZF压缩
5.1 Redis里的LZF
RDB因为存的是二进制，所以可以对长的字符串做压缩。如果字符串长度大于20，并且服务器开启了LZF压缩功能，那么保存压缩之后的数据。
用到的类库是liblzf
redis把以下4个源文件原封不动的拷过来了。
lzf_c.c
lzf_d.c
lzf.h
lzfP.h
5.2 Java里的LZF
java也有lzf的类库，可参考如下网址：
https://github.com/ning/compress
https://github.com/ning/jvm-compressor-benchmark/wiki
6. CRC校验
6.1 Redis里的CRC
RDB文件的末尾8个字节是CRC校验和（循环冗余校验）.
算法都在下面几个文件里
crc16.c
crc64.c
crc64.h
redis采用了crc-64-jones算法，
rio.h中的定义

struct _rio {
// 校验和计算函数，每次有写入/读取新数据时都要计算一次
void (*update_cksum)(struct _rio *, const void *buf, size_t len);
// 当前校验和
uint64_t cksum;
};
typedef struct _rio rio;
//将 buf 中的 len 字节写入到 r 中。
static inline size_t rioWrite(rio *r, const void *buf, size_t len) {
while (len) {
size_t bytes_to_write = (r->max_processing_chunk && r->max_processing_chunk < len) ? r->max_processing_chunk : len;
// 如果crc函数指针被赋值过，则调用它更新crc值
if (r->update_cksum) r->update_cksum(r,buf,bytes_to_write);
if (r->write(r,buf,bytes_to_write) == 0)
return 0;
buf = (char*)buf + bytes_to_write;
}
return 1;
}

如上定义了一个函数指针void (*update_cksum)，
rioWrite可以理解成写rdb文件时会调用的函数，每调用一次，它就会去调用函数指针，以这样的方式来更新crc值：
r->update_cksum()
rio.c

/*
* 通用校验和计算函数
*/
void rioGenericUpdateChecksum(rio *r, const void *buf, size_t len) {
r->cksum = crc64(r->cksum,buf,len);
}

这个函数就是简单的调用一下crc64.c里写的crc算法
rdb.c中

int rdbSave(char *filename) {
// 设置校验和函数，如果需要校验，则把函数指针赋值给rdb.update_cksum
if (server.rdb_checksum)
rdb.update_cksum = rioGenericUpdateChecksum;
// ......
cksum = rdb.cksum;
rioWrite(&rdb,&cksum,8);
}

最后可以看到rdbSave的时候会去调用rioWrite，而最后则是写入crc校验和。
6.2 Java里的CRC
JDK里也有crc的算法实现,不过是32位的
java.util.zip.CRC32
java.util.zip.Adler32
Adler-32 校验和几乎与 CRC-32 一样可靠，但是能够更快地计算出来。