深入剖析 redis RDB 持久化策略

struct _rio {

    // 函数指针，包括读操作，写操作和文件指针移动操作

    /* Backend functions.

     * Since this functions do not tolerate short writes or reads the return

     * value is simplified to: zero on error, non zero on complete success. */

    size_t (*read)(struct _rio *, void *buf, size_t len);

    size_t (*write)(struct _rio *, const void *buf, size_t len);

    off_t (*tell)(struct _rio *);

    // 校验和计算函数

    /* The update_cksum method if not NULL is used to compute the checksum of

     * all the data that was read or written so far. The method should be

     * designed so that can be called with the current checksum, and the buf

     * and len fields pointing to the new block of data to add to the checksum

     * computation. */

    void (*update_cksum)(struct _rio *, const void *buf, size_t len);

    // 校验和

    /* The current checksum */

    uint64_t cksum;

    // 已经读取或者写入的字符数

    /* number of bytes read or written */

    size_t processed_bytes;

    // 每次最多能处理的字符数

    /* maximum single read or write chunk size */

    size_t max_processing_chunk;

    // 可以是一个内存总的字符串，也可以是一个文件描述符

    /* Backend-specific vars. */

    union {

        struct {

            sds ptr;

            // 偏移量

            off_t pos;

        } buffer;

        struct {

            FILE *fp;

            // 偏移量

            off_t buffered; /* Bytes written since last fsync. */

            off_t autosync; /* fsync after 'autosync' bytes written. */

        } file;

    } io;

};

typedef struct _rio rio;

// 适用于内存缓存

static const rio rioBufferIO = {

    rioBufferRead,

    rioBufferWrite,

    rioBufferTell,

    NULL,           /* update_checksum */

    0,              /* current checksum */

    0,              /* bytes read or written */

    0,              /* read/write chunk size */

    { { NULL, 0 } } /* union for io-specific vars */

};

// 适用于文件 IO

static const rio rioFileIO = {

    rioFileRead,

    rioFileWrite,

    rioFileTell,

    NULL,           /* update_checksum */

    0,              /* current checksum */

    0,              /* bytes read or written */

    0,              /* read/write chunk size */

    { { NULL, 0 } } /* union for io-specific vars */

};

// 备份主程序

/* Save the DB on disk. Return REDIS_ERR on error, REDIS_OK on success */

int rdbSave(char *filename) {

    dictIterator *di = NULL;

    dictEntry *de;

    char tmpfile[256];

    char magic[10];

    int j;

    long long now = mstime();

    FILE *fp;

    rio rdb;

    uint64_t cksum;

    // 打开文件，准备写

    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;

    }

    // 初始化 rdb 结构体。rdb 结构体内指定了读写文件的函数，已写/读字符统计等数据

    rioInitWithFile(&rdb,fp);

    if (server.rdb_checksum) // 校验和

        rdb.update_cksum = rioGenericUpdateChecksum;

    // 先写入版本号

    snprintf(magic,sizeof(magic),"REDIS%04d",REDIS_RDB_VERSION);

    if (rdbWriteRaw(&rdb,magic,9) == -1) goto werr;

    for (j = 0; j dict;

        if (dictSize(d) == 0) continue;

        // 字典迭代器

        di = dictGetSafeIterator(d);

        if (!di) {

            fclose(fp);

            return REDIS_ERR;

        }

        // 写入 RDB 操作码

        /* Write the SELECT DB opcode */

        if (rdbSaveType(&rdb,REDIS_RDB_OPCODE_SELECTDB) == -1) goto werr;

        // 写入数据库序号

        if (rdbSaveLen(&rdb,j) == -1) goto werr;

        // 写入数据库中每一个数据项

        /* Iterate this DB writing every entry */

        while((de = dictNext(di)) != NULL) {

            sds keystr = dictGetKey(de);

            robj key,

                *o = dictGetVal(de);

            long long expire;

            // 将 keystr 封装在 robj 里

            initStaticStringObject(key,keystr);

            // 获取过期时间

            expire = getExpire(db,&key);

            // 开始写入磁盘

            if (rdbSaveKeyValuePair(&rdb,&key,o,expire,now) == -1) goto werr;

        }

        dictReleaseIterator(di);

    }

    di = NULL; /* So that we don't release it again on error. */

    // RDB 结束码

    /* EOF opcode */

    if (rdbSaveType(&rdb,REDIS_RDB_OPCODE_EOF) == -1) goto werr;

    // 校验和

    /* 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);

    rioWrite(&rdb,&cksum,8);

    // 同步到磁盘

    /* Make sure data will not remain on the OS's output buffers */

    fflush(fp);

    fsync(fileno(fp));

    fclose(fp);

    // 修改临时文件名为指定文件名

    /* 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:

    // 清理工作，关闭文件描述符等

    fclose(fp);

    unlink(tmpfile);

    redisLog(REDIS_WARNING,"Write error saving DB on disk: %s", strerror(errno));

    if (di) dictReleaseIterator(di);

    return REDIS_ERR;

}

// bgsaveCommand(),serverCron(),syncCommand(),updateSlavesWaitingBgsave() 会调用 rdbSaveBackground()

int rdbSaveBackground(char *filename) {

    pid_t childpid;

    long long start;

    // 已经有后台程序了，拒绝再次执行

    if (server.rdb_child_pid != -1) return REDIS_ERR;

    server.dirty_before_bgsave = server.dirty;

    // 记录这次尝试执行持久化操作的时间

    server.lastbgsave_try = time(NULL);

    start = ustime();

    if ((childpid = fork()) == 0) {

        int retval;

        // 取消监听

        /* Child */

        closeListeningSockets(0);

        redisSetProcTitle("redis-rdb-bgsave");

        // 执行备份主程序

        retval = rdbSave(filename);

        // 脏数据，其实就是子进程所消耗的内存大小

        if (retval == REDIS_OK) {

            // 获取脏数据大小

            size_t private_dirty = zmalloc_get_private_dirty();

            // 记录脏数据

            if (private_dirty) {

                redisLog(REDIS_NOTICE,

                    "RDB: %zu MB of memory used by copy-on-write",

                    private_dirty/(1024*1024));

            }

        }

        // 退出子进程

        exitFromChild((retval == REDIS_OK) ? 0 : 1);

    } else {

        /* Parent */

        // 计算 fork 消耗的时间

        server.stat_fork_time = ustime()-start;

        // fork 出错

        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);

        // 记录保存的起始时间

        server.rdb_save_time_start = time(NULL);

        // 子进程 ID

        server.rdb_child_pid = childpid;

        updateDictResizePolicy();

        return REDIS_OK;

    }

    return REDIS_OK; /* unreached */

}

int rdbSaveKeyValuePair(rio *rdb, robj *key, robj *val,

                        long long expiretime, long long now)

{

    // 过期时间

    /* Save the expire time */

    if (expiretime != -1) {

        /* If this key is already expired skip it */

        if (expiretime

<p>如果对 redis RDB 数据格式细则感兴趣，欢迎访问我的 github & 欢迎讨论。</p>

<h3 id="参考文档">参考文档</h3>

<p>http://redis.io/topics/persistence</p>

<p>—-</p>

<p>捣乱 2014-3-26</p>

<p>http://daoluan.net</p>

    <p class="copyright">

        原文地址：深入剖析 redis RDB 持久化策略, 感谢原作者分享。

    </p>