? ? ? ? ? 堅持了一個月左右的時間，從最開始的對Redis的代碼做分類，從struct結構體分析開始，到最后分析main主程序結束，中間，各大模塊的代碼逐個擊破，學習，總之，收獲了非常多，好久沒有這么久的耐心把一個框架學透，學習一個框架，會用那只是小小的一部分，能把背后的原理吃透才是真功夫。在這個學習的最后階段，是時候要來點干貨了，我把這1個多月來的一些總結的一些比較好的代碼，和設計思想總結出來了，原本想湊成10大精彩設計的，可后來感覺每個點都挺精彩的，還是做成了11大優秀設計，包證讓你打開研究，這里無關語言，重在一種編程的思想和設計，希望大家能好好領會。(下面的排序無關緊要，我只是按照時間順序下來。后面的鏈接為我寫的相關文章，如果想具體了解，請點擊請入) ? ? ??1.hyperloglog基量統計算法的實現([http://blog.csdn.net/androidlushangderen/article/details/40683763](http://blog.csdn.net/androidlushangderen/article/details/40683763))。說到這個，比較搞笑的一點是，我剛剛開始竟然以為是某種類型的日志，和slowLog一樣，后來才明白，這是一種基量統計算法，類似的算法還有LLC，HLLC是他的升級版本。 ? ? ? ?2.zmalloc內存分配的重新實現([http://blog.csdn.net/androidlushangderen/article/details/40659331]())。Redis的作者在內存分配上顯然是早有準備，不會傻傻的還是調用系統的mallo和free方法，人家在這里做了一個小小的封裝，便于管理者更方便的控制系統的內存，下面是一個小小的結構體的聲明，看到這個大家估計會明白。 ~~~ /* 調用zmalloc申請size個大小的空間 */ void *zmalloc(size_t size) { //實際調用的還是malloc函數 void *ptr = malloc(size+PREFIX_SIZE); //如果申請的結果為null，說明發生了oom,調用oom的處理方法 if (!ptr) zmalloc_oom_handler(size); #ifdef HAVE_MALLOC_SIZE //更新used_memory的大小 update_zmalloc_stat_alloc(zmalloc_size(ptr)); return ptr; #else *((size_t*)ptr) = size; update_zmalloc_stat_alloc(size+PREFIX_SIZE); return (char*)ptr+PREFIX_SIZE; #endif } ~~~ ? ? ? ? ??3.multi事務操作([http://blog.csdn.net/androidlushangderen/article/details/40392209]())。Redis中的事務操作給我一種煥然一新的感覺，作者在做此設計的時候，用到了key，和watch key的概念，一個key維護了一個所有watch他的所有Client列表，一個Client自身也擁有一個他所監視的所有key，如果一個key被touch了，所有同樣見識此key的客戶端的下一步操作統統失效，具體怎么實現，請猛點后面的鏈接。 ? ? ? ? ??4.redis-benchmark性能測試([http://blog.csdn.net/androidlushangderen/article/details/40211907]())。Redis在這里出現了一個性能統計的概念，比較高大上的感覺，與調用了很多latency延時類的方法，就是判斷延時的情況來看性能的好壞的。 ? ? ? ?5.zipmap壓縮結構的設計([http://blog.csdn.net/androidlushangderen/article/details/39994599]())。Redis在內存處理上可謂是想盡了辦法，ziplist壓縮列表和zipmap壓縮圖就是非常典型的設計。與往常的結構體內直接放一個int64類型的整形變量，這樣就占了8個字節，但是一般情況下，我們保存的數值都比較小，1個字節差不多就夠了，所有就浪費了7個字節，所以zip壓縮系列結構體，就可以動態分配字節應對不同的情況，這個設計非常精彩，要確定這個key-value 的位置，通過前面保留的長度做偏移量的定位。 ? ? ??6.sparkline微線圖的重新設計([http://blog.csdn.net/androidlushangderen/article/details/39964591]())。Redis的sparkline的出現應該又是幫我掃盲了，人家可以用字符串的形式輸出一張類似折線圖的表，利用了采集的很多歌Sample的樣本點，這個類多用于分析統計中出現，比如latency.c延時分析類中用到了。 ? ? ?7.對象引用計數實現內存管理([http://blog.csdn.net/androidlushangderen/article/details/40716469]())。我們知道管理對象的生命周期一般有2種方法，1個是根搜索法（JVM中用的就是這個），另一個就是引用計數法，而Redis就給我們對此方法的實現，下面是對象增引用和減少引用的實現: ~~~ /* robj對象增減引用計數,遞增robj中的refcount的值 */ void incrRefCount(robj *o) { //遞增robj中的refcount的值 o->refcount++; } ~~~ ~~~ /* 遞減robj中的引用計數，引用到0后，釋放對象 */ void decrRefCount(robj *o) { //如果之前的引用計數已經<=0了，說明出現異常情況了 if (o->refcount <= 0) redisPanic("decrRefCount against refcount <= 0"); if (o->refcount == 1) { //如果之前的引用計數為1，再遞減一次，恰好內有被任何對象引用了，所以就可以釋放對象了 switch(o->type) { case REDIS_STRING: freeStringObject(o); break; case REDIS_LIST: freeListObject(o); break; case REDIS_SET: freeSetObject(o); break; case REDIS_ZSET: freeZsetObject(o); break; case REDIS_HASH: freeHashObject(o); break; default: redisPanic("Unknown object type"); break; } zfree(o); } else { //其他對于>1的引用計數的情況，只需要按常規的遞減引用計數即可 o->refcount--; } } ~~~ 減少引用的方法實現是重點。 ? ? ?8.fork子進程實現后臺程序([http://blog.csdn.net/androidlushangderen/article/details/40266579]())。fork創建子線程實現后臺程序的操作，我還是第一次見能這么用的，以前完全不知道fork能怎么使用的，這次真的是漲知識了。里面關鍵的一點是fork方法在子線程和父線程中的返回值不同做處理，父線程返回子線程的PID號，在子線程中返回的是0. ~~~ /* 后臺進行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 */ } ~~~ ? ? ? ??9.long long 類型轉為String類型方法([http://blog.csdn.net/androidlushangderen/article/details/40649623](http://blog.csdn.net/androidlushangderen/article/details/40649623))。以前做過很多字符串轉數值和數值轉字符串的算法實現，也許你的功能是實現了，但是效率呢，但面對的是非常長的long long類型的數字時，效率可能會更低。Redis在這里給我們提供了一個很好的思路，平時我們/10的計算，再%1o求余數，人家直接來了個/100的，然后直接通過字符串數組和余數值直接的映射，進行計算。算法如下； ~~~ /* Convert a long long into a string. Returns the number of * characters needed to represent the number. * If the buffer is not big enough to store the string, 0 is returned. * * Based on the following article (that apparently does not provide a * novel approach but only publicizes an already used technique): * * https://www.facebook.com/notes/facebook-engineering/three-optimization-tips-for-c/10151361643253920 * * Modified in order to handle signed integers since the original code was * designed for unsigned integers. */ /* long long類型轉化為string類型 */ int ll2string(char* dst, size_t dstlen, long long svalue) { static const char digits[201] = "0001020304050607080910111213141516171819" "2021222324252627282930313233343536373839" "4041424344454647484950515253545556575859" "6061626364656667686970717273747576777879" "8081828384858687888990919293949596979899"; int negative; unsigned long long value; /* The main loop works with 64bit unsigned integers for simplicity, so * we convert the number here and remember if it is negative. */ /* 在這里做正負號的判斷處理 */ if (svalue < 0) { if (svalue != LLONG_MIN) { value = -svalue; } else { value = ((unsigned long long) LLONG_MAX)+1; } negative = 1; } else { value = svalue; negative = 0; } /* Check length. */ uint32_t const length = digits10(value)+negative; if (length >= dstlen) return 0; /* Null term. */ uint32_t next = length; dst[next] = '\0'; next--; while (value >= 100) { //做值的換算 int const i = (value % 100) * 2; value /= 100; //i所代表的余數值用digits字符數組中的對應數字代替了 dst[next] = digits[i + 1]; dst[next - 1] = digits[i]; next -= 2; } /* Handle last 1-2 digits. */ if (value < 10) { dst[next] = '0' + (uint32_t) value; } else { int i = (uint32_t) value * 2; dst[next] = digits[i + 1]; dst[next - 1] = digits[i]; } /* Add sign. */ if (negative) dst[0] = '-'; return length; } ~~~ ? ? ? ?10.正則表達式的實現算法([http://blog.csdn.net/androidlushangderen/article/details/40649623](http://blog.csdn.net/androidlushangderen/article/details/40649623))。正則表達式在我們平時用的可是非常多的，可有多少知道，正則表達式是如何實現通過簡單的模式進程匹配，背后的原理實現到底怎么樣呢，為什么?就可以代表任何一個字符接著往后匹配，*代表的是所有字符，要實現這樣一個算法，也不是那么容易的哦，Redis就實現了這么一個算法，算是撿到寶了吧。 ~~~ /* Glob-style pattern matching. */ /*支持glob-style的通配符格式,如*表示任意一個或多個字符,?表示任意字符,[abc]表示方括號中任意一個字母。*/ int stringmatchlen(const char *pattern, int patternLen, const char *string, int stringLen, int nocase) { while(patternLen) { switch(pattern[0]) { case '*': while (pattern[1] == '*') { //如果出現的是**，說明一定匹配 pattern++; patternLen--; } if (patternLen == 1) return 1; /* match */ while(stringLen) { if (stringmatchlen(pattern+1, patternLen-1, string, stringLen, nocase)) return 1; /* match */ string++; stringLen--; } return 0; /* no match */ break; case '?': if (stringLen == 0) return 0; /* no match */ /* 因為？能代表任何字符，所以，匹配的字符再往后挪一個字符 */ string++; stringLen--; break; case '[': { int not, match; pattern++; patternLen--; not = pattern[0] == '^'; if (not) { pattern++; patternLen--; } match = 0; while(1) { if (pattern[0] == '\\') { //如果遇到轉義符，則模式字符往后移一個位置 pattern++; patternLen--; if (pattern[0] == string[0]) match = 1; } else if (pattern[0] == ']') { //直到遇到另外一個我中括號，則停止 break; } else if (patternLen == 0) { pattern--; patternLen++; break; } else if (pattern[1] == '-' && patternLen >= 3) { int start = pattern[0]; int end = pattern[2]; int c = string[0]; if (start > end) { int t = start; start = end; end = t; } if (nocase) { start = tolower(start); end = tolower(end); c = tolower(c); } pattern += 2; patternLen -= 2; if (c >= start && c <= end) match = 1; } else { if (!nocase) { if (pattern[0] == string[0]) match = 1; } else { if (tolower((int)pattern[0]) == tolower((int)string[0])) match = 1; } } pattern++; patternLen--; } if (not) match = !match; if (!match) return 0; /* no match */ string++; stringLen--; break; } case '\\': if (patternLen >= 2) { pattern++; patternLen--; } /* fall through */ default: /* 如果沒有正則表達式的關鍵字符，則直接比較 */ if (!nocase) { if (pattern[0] != string[0]) //不相等，直接不匹配 return 0; /* no match */ } else { if (tolower((int)pattern[0]) != tolower((int)string[0])) return 0; /* no match */ } string++; stringLen--; break; } pattern++; patternLen--; if (stringLen == 0) { while(*pattern == '*') { pattern++; patternLen--; } break; } } if (patternLen == 0 && stringLen == 0) //如果匹配字符和模式字符匹配的長度都減少到0了，說明匹配成功了 return 1; return 0; } ~~~ ? ? ? ?11.Redis的drand48()隨機算法重實現([http://blog.csdn.net/androidlushangderen/article/details/40582189](http://blog.csdn.net/androidlushangderen/article/details/40582189))。Redis隨機算法的實現作為11大設計的最后一個，并不是說這個設計相比前面有多么的爛，因為我覺得比較有特點，我就追加了一個上去。由于Redis的作者考慮到隨機算法的在不同的操作系統可能會表現出不同的特性，所以不建議采用math.rand()方法，而是基于drand48()的算法重新實現了一個。具體什么叫drand48().請猛點鏈接處。 ? ? ?好了，以上就是我印象中的Redis中比較優秀的設計。其實在Redis的很多還有很多優秀代碼的痕跡，由于篇幅有限，等待著讀者自己去學習，發現。