牛骨文教育服务平台(让学习变的简单)

           过去2,3天内把redis内部的测试相关包分析了一遍,总体感觉还是比较容易的,总共5个文件,也让我们涨了一下见识,什么叫内置的测试函数。今天,我把目标进行了转移,下面我准备继续学习与代码逻辑稍稍无关的模块,数据层,在我的分类中,就是在Data的文件包。在这个里面,首当其冲,我研究了rdb.c,直接与数据库操作相关。什么叫数据库操作相关呢,最直接的意思就是,数据库的相关操作到最后到会直接映射到这个文件中的函数操作。所以在理解这些操作之前,我得先介绍一下里面的一些东西,免得会比较乱。我们知道,redis内部支持很多中类型,

1.list列表

2.hash类型

3.set类型

4.string类型

其中如果是list列表类型,其实内部的编码方式又可分为2种,linkedList普通链表模式,ziplist压缩列表模式,所以说里面的代码里的类型是非常多的,所以建议读者阅读学习源码的时候,不要搞混了。rdb中的数据存储的基本格式为[len][data],前面使用字节表示的长度,后面是真实的数据,当然我这说的是普通的字符串类型的key:value的值,如果是纯数字,直接用字节表示值,根据值的大小分配不同的字节表示,不得不说,redis在数据存储方面上,把数据存储的内存消耗降到了极致。比如只要是在数据库中保存的长度等数字的,必须经过计算判断,然后再分配相应的字节保存(跟前面压缩列表等的原理类型):

/* Load an encoded length. The "isencoded" argument is set to 1 if the length
 * is not actually a length but an "encoding type". See the REDIS_RDB_ENC_*
 * definitions in rdb.h for more information. */
/* 加载长度,也需要根据编码方式,读取不同的buf获取长度 */
uint32_t rdbLoadLen(rio *rdb, int *isencoded) {
    unsigned char buf[2];
    uint32_t len;
    int type;

    if (isencoded) *isencoded = 0;
    if (rioRead(rdb,buf,1) == 0) return REDIS_RDB_LENERR;
    type = (buf[0]&0xC0)>>6;
    if (type == REDIS_RDB_ENCVAL) {
        /* Read a 6 bit encoding type. */
        if (isencoded) *isencoded = 1;
        return buf[0]&0x3F;
    } else if (type == REDIS_RDB_6BITLEN) {
        /* Read a 6 bit len. */
        return buf[0]&0x3F;
    } else if (type == REDIS_RDB_14BITLEN) {
        /* Read a 14 bit len. */
        if (rioRead(rdb,buf+1,1) == 0) return REDIS_RDB_LENERR;
        return ((buf[0]&0x3F)<<8)|buf[1];
    } else {
        /* Read a 32 bit len. */
        if (rioRead(rdb,&len,4) == 0) return REDIS_RDB_LENERR;
        return ntohl(len);
    }
}

只要通过编码方式存储的字符串,普通字符串都要先经过压缩再存入,取出的时候先做解压操作:

/* rdb加载字符串对象的泛型方法 */
robj *rdbGenericLoadStringObject(rio *rdb, int encode) {
    int isencoded;
    uint32_t len;
    sds val;

    len = rdbLoadLen(rdb,&isencoded);
    if (isencoded) {
    	//返回值主要为加载数值对象,和获取解压后的字符串对象
        switch(len) {
        case REDIS_RDB_ENC_INT8:
        case REDIS_RDB_ENC_INT16:
        case REDIS_RDB_ENC_INT32:
            return rdbLoadIntegerObject(rdb,len,encode);
        case REDIS_RDB_ENC_LZF:
            return rdbLoadLzfStringObject(rdb);
        default:
            redisPanic("Unknown RDB encoding type");
        }
    }

	//无编码方式,直接读取rdb
    if (len == REDIS_RDB_LENERR) return NULL;
    val = sdsnewlen(NULL,len);
    if (len && rioRead(rdb,val,len) == 0) {
        sdsfree(val);
        return NULL;
    }
    return createObject(REDIS_STRING,val);
}

综上,我总结了几点,redis数据量在存储数据上的做的调优
1.长度等数值数据存储,根据数值大小的不同,分配不同的字节存储,1个字节,2个字节,后面直接到5个字节,避免直接像int32,int64一样,直接占去4,8个字节。一般字符串的长度都是比较小的,如果每个字符串的长度是10,你用4,8个字节去存的话,大大的浪费空间了。
2.字符串等非数值存储,redis在这里采用了lzf压缩算法,当然取出的时候,你要进行解压,或者你从最开始的时候不选择的压缩存储,而是直接存储。

所以,这样的设计非常棒,数据库的任何操作结果都会最终赋值到robj->ptr上:

if (o->encoding == REDIS_ENCODING_INTSET) {
                /* Fetch integer value from element */
                if (isObjectRepresentableAsLongLong(ele,&llval) == REDIS_OK) {
                	//最后都会通过吧值赋在obj->ptr上
                    o->ptr = intsetAdd(o->ptr,llval,NULL);
                } else {
                    setTypeConvert(o,REDIS_ENCODING_HT);
                    dictExpand(o->ptr,len);
                }
            }

在这些个方法里面,还有一个比较特殊的后台保存到数据库的方法,为什么会有这样的操作呢,因为redis其实和mencached一样,是内存数据库,如果对数据的操作都直接是写入磁盘,I/O开销肯定很大,所以一般内存数据库都是先把操作结构都存放在内存中,等到了内存的数据满了,再持久化到磁盘中,就是保存数据库操作到文件中了。redis在这里还很人性化的提供了backgroundSave()的方式:,如果这个问题出现在java里面,我的直接做法肯定开个线程让他直接运行Save的方法就行了,但是想在C语言中实现这种类似多线程的操作,我还真想不出来,最终他的答案是fork(),在Linux编程中,肯定接触过了这个方法,在C语言的应用编程中基本没看到过,我也是头次领略到fork方法还能这么用,先看看原方法调用细节:

/* 后台进行rbd保存操作 */
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();
    //利用fork()创建子进程用来实现rdb的保存操作
    //此时有2个进程在执行这段函数的代码,在子进行程返回的pid为0,
    //所以会执行下面的代码,在父进程中返回的代码为孩子的pid,不为0,所以执行else分支的代码
    //在父进程中放返回-1代表创建子进程失败
    if ((childpid = fork()) == 0) {
    	//在这个if判断的代码就是在子线程中后执行的操作
        int retval;

        /* Child */
        closeListeningSockets(0);
        redisSetProcTitle("redis-rdb-bgsave");
        //这个就是刚刚说的rdbSave()操作
        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 */
        server.stat_fork_time = ustime()-start;
        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);
        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);
        server.rdb_child_pid = childpid;
        updateDictResizePolicy();
        return REDIS_OK;
    }
    return REDIS_OK; /* unreached */
}

父进程fork()出的子线程是基本完全复用父亲线程的,所以也就是说,父子线程都会执行这个函数,但是唯一的区别是执行fork函数返回值是不同的,子线程因为是被fork出来的,返回的就是0代表自身,父亲线程就是返回子线程的PID,然后根据返回的PID不同,执行不同的操作,子线程就完全独立于父亲线程,做自己的保存操作。这也是头次我知道了fork还能这么用。下面亮出.h头文件中的API,其实和.c文件里的差了很多的方法:

int rdbSaveType(rio *rdb, unsigned char type); /* 保存类型操作 */
int rdbLoadType(rio *rdb); /* 加载RDB中的格式类型 */
int rdbSaveTime(rio *rdb, time_t t);
time_t rdbLoadTime(rio *rdb); /* 加载时间,都是间接调用的是rioRead()方法 */
int rdbSaveLen(rio *rdb, uint32_t len); /* 保存一个字符串对象的长度时,根据长度的不同,分不同的编码方式 */
uint32_t rdbLoadLen(rio *rdb, int *isencoded); /* 加载长度,也需要根据编码方式,读取不同的buf获取长度 */
int rdbSaveObjectType(rio *rdb, robj *o); /* 根据robj中的编码方式,保存到rbd中 */
int rdbLoadObjectType(rio *rdb); /* 加载rbd中的obj Type */
int rdbLoad(char *filename); /* 加载rdb数据库文件 */
int rdbSaveBackground(char *filename); /* 后台进行rbd保存操作 */
void rdbRemoveTempFile(pid_t childpid); /* 移除子进程操作的相关保存rdb文件 */
int rdbSave(char *filename); /* 保存rdb数据库的内容到磁盘中 */
int rdbSaveObject(rio *rdb, robj *o); /* 保存redis obj对象到rdb中 */
off_t rdbSavedObjectLen(robj *o); /* 获取保存后的长度,其实就是获取了保存数据时计算的偏移量 */
off_t rdbSavedObjectPages(robj *o);
robj *rdbLoadObject(int type, rio *rdb); /* 加载redis obj对象,有特定的Type类型 */
void backgroundSaveDoneHandler(int exitcode, int bysignal); /* 后台保存数据库操作完成后的处理方法 */
int rdbSaveKeyValuePair(rio *rdb, robj *key, robj *val, long long expiretime, long long now);
robj *rdbLoadStringObject(rio *rdb); /* 无编码方式加载字符串对象 */
void saveCommand(redisClient *c) /* 将保存操作封装成命令的形式 */
void bgsaveCommand(redisClient *c) /* 将后台保存数据库操作封装成命令的模式 */