Redis源码分析--RDB实现源码阅读

重要说明，在看这篇文章之前，最好先通过剖析Redis RDB文件 了解RDB文件的结构；

RDB相关源码在rdb.c中；通过saveCommand(redisClient *c) 和bgsaveCommand(redisClient *c) 两个方法可知，RDB持久化业务逻辑在rdbSave(server.rdb_filename)和rdbSaveBackground(server.rdb_filename这两个方法中；一个通过执行"save"触发，另一个通过执行"bgsave"或者save seconds changes条件满足时(在redis.c的serverCron中)触发：

redis.c里serverCron中通过调用rdbSaveBackground(server.rdb_filename)触发bgsave的部分代码：

if (server.dirty >= sp->changes &&
    server.unixtime-server.lastsave > sp->seconds &&
    (server.unixtime-server.lastbgsave_try >
     REDIS_BGSAVE_RETRY_DELAY ||
     server.lastbgsave_status == REDIS_OK))
{
    redisLog(REDIS_NOTICE,"%d changes in %d seconds. Saving...",
        sp->changes, (int)sp->seconds);
    rdbSaveBackground(server.rdb_filename);
    break;
}

通过阅读rdbSaveBackground(char *filename)的源码可知，其最终的实现还是调用rdbSave(char *filename)，只不过是通过fork()出的子进程来执行罢了，所以bgsave和save的实现是殊途同归：

int rdbSaveBackground(char *filename) {
    pid_t childpid;
    long long start;

    // 如果已经有RDB持久化任务，那么rdb_child_pid的值就不是-1，那么返回REDIS_ERR；
    if (server.rdb_child_pid != -1) return REDIS_ERR;

    server.dirty_before_bgsave = server.dirty;
    server.lastbgsave_try = time(NULL);

    // 记录RDB持久化开始时间
    start = ustime();
    //fork一个子进程，
    if ((childpid = fork()) == 0) {
        // 如果fork()的结果childpid为0，即当前进程为fork的子进程，那么接下来调用rdbSave()进程持久化；
        int retval;

        /* Child */
        closeListeningSockets(0);
        redisSetProcTitle("redis-rdb-bgsave");
        // bgsave事实上就是通过fork的子进程调用rdbSave()实现, rdbSave()就是save命令业务实现；
        retval = rdbSave(filename);
        if (retval == REDIS_OK) {
            size_t private_dirty = zmalloc_get_private_dirty();

            if (private_dirty) {
                // RDB持久化成功后，如果是notice级别的日志，那么log输出RDB过程中copy-on-write使用的内存
                redisLog(REDIS_NOTICE,
                    "RDB: %zu MB of memory used by copy-on-write",
                    private_dirty/(1024*1024));
            }
        }
        exitFromChild((retval == REDIS_OK) ? 0 : 1);
    } else {
        // 父进程更新redisServer记录一些信息，例如：fork进程消耗的时间stat_fork_time, 
        /* Parent */
        server.stat_fork_time = ustime()-start;
       // 更新redisServer记录fork速率：每秒多少G；zmalloc_used_memory()的单位是字节，所以通过除以(1024*1024*1024),得到GB；由于记录的fork_time即fork时间是微妙，所以*1000000，得到每秒钟fork多少GB的速度；
        server.stat_fork_rate = (double) zmalloc_used_memory() * 1000000 / server.stat_fork_time / (1024*1024*1024); /* GB per second. */
        latencyAddSampleIfNeeded("fork",server.stat_fork_time/1000);
        // 如果fork子进程出错，即childpid为-1，更新redisServer，记录最后一次bgsave状态是REDIS_ERR;
        if (childpid == -1) {
            server.lastbgsave_status = REDIS_ERR;
            redisLog(REDIS_WARNING,"Can't save in background: fork: %s",
                strerror(errno));
            return REDIS_ERR;
        }
        redisLog(REDIS_NOTICE,"Background saving started by pid %d",childpid);
        // 最后在redisServer中记录的save开始时间重置为空，并记录执行bgsave的子进程id，即child_pid；
        server.rdb_save_time_start = time(NULL);
        server.rdb_child_pid = childpid;
        server.rdb_child_type = REDIS_RDB_CHILD_TYPE_DISK;
        updateDictResizePolicy();
        return REDIS_OK;
    }
    return REDIS_OK; /* unreached */
}

RDB持久化实现：

/* Save the DB on disk. Return REDIS_ERR on error, REDIS_OK on success. */
int rdbSave(char *filename) {
    char tmpfile[256];
    FILE *fp;
    rio rdb;
    int error;
    // 文件临时文件名为temp-${pid}.rdb
    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;
    }

    rioInitWithFile(&rdb,fp);
    // RDB持久化的核心实现；
    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;

    // 重命名rdb文件的命名；
    /* Use RENAME to make sure the DB file is changed atomically only
     * if the generate DB file is ok. */
    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;
    }
    redisLog(REDIS_NOTICE,"DB saved on disk");
    server.dirty = 0;
    server.lastsave = time(NULL);
    server.lastbgsave_status = REDIS_OK;
    return REDIS_OK;

werr:
    redisLog(REDIS_WARNING,"Write error saving DB on disk: %s", strerror(errno));
    fclose(fp);
    unlink(tmpfile);
    return REDIS_ERR;
}

rdbSaveRio--RDB持久化实现的核心代码--根据RDB文件协议将所有redis中的key-value写入rdb文件中:

/* Produces a dump of the database in RDB format sending it to the specified
 * Redis I/O channel. On success REDIS_OK is returned, otherwise REDIS_ERR
 * is returned and part of the output, or all the output, can be
 * missing because of I/O errors.
 *
 * When the function returns REDIS_ERR and if 'error' is not NULL, the
 * integer pointed by 'error' is set to the value of errno just after the I/O
 * error. */
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;
    // rdb文件中最先写入的内容就是magic，magic就是REDIS这个字符串+4位版本号
    snprintf(magic,sizeof(magic),"REDIS%04d",REDIS_RDB_VERSION);
    if (rdbWriteRaw(rdb,magic,9) == -1) goto werr;

    // 遍历所有db重写rdb文件；
    for (j = 0; j < server.dbnum; j++) {
        redisDb *db = server.db+j;
        dict *d = db->dict;
        // 如果db的size为0，即没有任何key，那么跳过，遍历下一个db；
        if (dictSize(d) == 0) continue;
        di = dictGetSafeIterator(d);
        if (!di) return REDIS_ERR;

        // 写入REDIS_RDB_OPCODE_SELECTDB，这个值redis定义为254，即FE，再通过rdbSaveLen合入当前dbnum，例如当前db为0，那么写入FE 00
        /* Write the SELECT DB opcode */
        if (rdbSaveType(rdb,REDIS_RDB_OPCODE_SELECTDB) == -1) goto werr;
        if (rdbSaveLen(rdb,j) == -1) goto werr;

        // 如注释所表达的，迭代遍历db这个dict的每一个entry；
        /* Iterate this DB writing every entry */
        while((de = dictNext(di)) != NULL) {
            // 先得到当前entry的key(sds类型)和value（redisObject类型）；
            sds keystr = dictGetKey(de);
            robj key, *o = dictGetVal(de);
            long long expire;

            initStaticStringObject(key,keystr);
            // 从redisDb的expire这个dict中查询过期时间属性值；
            expire = getExpire(db,&key);
            // 每个entry（redis中的key和其value）rdb持久化的核心代码
            if (rdbSaveKeyValuePair(rdb,&key,o,expire,now) == -1) goto werr;
        }
        dictReleaseIterator(di);
    }
    di = NULL; /* So that we don't release it again on error. */

    // 遍历所有db后，写入EOF这个opcode，REDIS_RDB_OPCODE_EOF申明为255，即FF，所以是写入FF到rdb文件中；FF是redis对rdb文件结束的定义；
    /* EOF opcode */
    if (rdbSaveType(rdb,REDIS_RDB_OPCODE_EOF) == -1) goto werr;

    // 最后写入8个字节长度的checksum值到rdb文件尾部；
    /* CRC64 checksum. It will be zero if checksum computation is disabled, the
     * loading code skips the check in this case. */
    cksum = rdb->cksum;
    memrev64ifbe(&cksum);
    if (rioWrite(rdb,&cksum,8) == 0) goto werr;
    return REDIS_OK;

werr:
    if (error) *error = errno;
    if (di) dictReleaseIterator(di);
    return REDIS_ERR;
}

每个entry（key-value）rdb持久化的核心代码：

/* Save a key-value pair, with expire time, type, key, value.
 * On error -1 is returned.
 * On success if the key was actually saved 1 is returned, otherwise 0
 * is returned (the key was already expired). */
int rdbSaveKeyValuePair(rio *rdb, robj *key, robj *val,
                        long long expiretime, long long now)
{
    /* Save the expire time */
    if (expiretime != -1) {
        // 如果过期时间少于当前时间，那么表示该key已经失效，返回不做任何保存；
        /* If this key is already expired skip it */
        if (expiretime < now) return 0;
        // 如果当前遍历的entry有失效时间属性，那么保存REDIS_RDB_OPCODE_EXPIRETIME_MS即252，即"FC"以及失效时间到rdb文件中，
        if (rdbSaveType(rdb,REDIS_RDB_OPCODE_EXPIRETIME_MS) == -1) return -1;
        if (rdbSaveMillisecondTime(rdb,expiretime) == -1) return -1;
    }

    // 接下来保存redis key的类型，key，以及value到rdb文件中；
    /* Save type, key, value */
    if (rdbSaveObjectType(rdb,val) == -1) return -1;
    if (rdbSaveStringObject(rdb,key) == -1) return -1;
    if (rdbSaveObject(rdb,val) == -1) return -1;
    return 1;
}

通过上面的源码分析得到最终rdb文件的格式如下：

REDIS // RDB协议约束的固定字符串
0006 // redis的版本号
FE 00 // 表示当前接下来的key都是db=0中的key；
FC 1506327609 // 表示key失效时间点为1506327609
0 // 表示key的属性是string类型；
username // key
afei // value
FF // 表示遍历完成
y73e9iq1 // checksum值

备注:
#define REDIS_RDB_TYPE_STRING 0
#define REDIS_RDB_TYPE_LIST 1
#define REDIS_RDB_TYPE_SET 2
#define REDIS_RDB_TYPE_ZSET 3
#define REDIS_RDB_TYPE_HASH 4