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 一样可靠，但是能够更快地计算出来。