? ? ? ? ? ?過去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) /* 將后臺保存數據庫操作封裝成命令的模式 */ ~~~