过去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) /* 将后台保存数据库操作封装成命令的模式 */ ~~~