前言

通過之前的源碼閱讀與分析, 我們通過服務的啟動, 數據流的接受與處理, 整體DB結構, 詳細的存儲數據結構等方面的學習對於redis6.0有了一個較為系統的認知。尤其是其中的一些優秀的設計我們在學習完成後也要深入的加以分析和思考, 是否可以將這些經驗借鑑到我們實際的工作中呢？這次就和大家一起討論一下我學習完成後的收穫。

伺服器模型

在Redis6.0之前的版本伺服器採用單進程單線程的處理方式, 優點就是避免了並發的鎖開銷, 缺點是不能充分利用CPU的多核處理。在現在業務場景中, CPU通常不會成為負載的主要瓶頸, 更多在於內存和網絡。Redis 的多線程網絡模型實際上並不是一個標準的 Multi-Reactors/Master-Workers 模型, 在6.0.0的版本中, I/O線程只是負責完成I/O流的處理任務, 當然主線程也會承擔部分I/O任務, 真正處理命令執行的在主線程中完成。這樣依然可以保持無鎖處理指令。這樣既保持了原系統的兼容性，又能利用多核提升 I/O 性能。這給我帶來的思考是我們不必要完全拘泥於歷史經驗的典型處理方法。針對於不同的場景可以做我們自己的伺服器處理策略。

分享一個case, 在高並發的分布式系統中, 對於同一個用戶請求的處理可能涉及到加鎖, 事務操作, 冪等性校驗等情況。類比於redis對於命令的處理, 本質上是相同的, 用戶請求的數據相當於set操作, 資料庫中的數據相當於RedisDB中待操作的對象, 在業務中我們一般會通過分布式鎖控制並發, 使用資料庫的隔離級別控制事務鎖, 使用資料庫唯一鍵做冪等控制等。這種通用性方案可以在很大程度上避免錯誤的發生, 但是性能上卻受到一定的影響，而且處理不當會經常有死鎖的發生, 那麼我們是否可以參考Redis在多線程上的處理方式實現業務呢？答案是可以的。

如圖, 在server中我們實現一個典型的Master-Workers架構的tcp伺服器, 可以根據配置文件配置指定的線程池數量處理用戶請求。

// 創建processfunc (process *taskProcess) createGoPool() { for i := 0; i < process.maxGo; i++ { go func(processId int) { ch := make(chan *server.ServerContext, 128) process.goPool.Store(processId, ch) for { select { case data, ok := <-ch: if !ok { fmt.Println("process is not ok ", processId) } // 處理請求數據 process.dealData(data) } } }(i) }}

當請求到來的時候分發規則就是我們可以自定義的, 加入我們的業務可以根據UID來進行請求劃分, 則可以根據UID%threadNum指定processId來分發請求到指定的goroutine中來處理, 偽代碼如下:

func (dispatcher *Dispatcher) parseMsg(info *RequestInfo) { // 解析body數據 requestBody := &server.RequestMsgBody{} err := json.Unmarshal(info.msg, &requestBody) if err != nil { writeinfo := fmt.Sprintf("解析發生錯誤: %+v, 消息丟棄: %s", err, string(info.msg)) write, _ := comm.GbkToUtf8([]byte(writeinfo)) info.conn.Write(write) return } // 如果存在當前用戶請求的ID, 則使用同一線程處理, 否則使用空閒線程處理並記錄用戶ID與線程ID var processId int if val, isExistsUidProcessId := dispatcher.activeUserProcessMap.Load(requestBody.Md.Uid); buyOk { processId = val.(int) } else { processId = NewTaskProcess.GetIdleProcess() } // 將數據分發 taskData := &server.ServerContext{ Conn: info.conn, Body: requestBody, } // 發送數據分發消息 ch, ok := NewTaskProcess.goPool.Load(processId) if ok { ch.(chan *server.ServerContext) <- taskData } // 維護UID對processId的關係 // 發送成功，保存map & 計數 if requestBody.Md.Uid > 0 { dispatcher.activeUserProcessMap.Store(requestBody.Md.Uid, processId) dispatcher.activeUserCount(requestBody.Md.Uid, 1) }}

基於這樣的方案可以將用戶請求執行單線程的處理, 不用擔心任何的鎖消耗, 當然在負載均衡網關層還需要添加請求分發的插件, 將請求與集群的server進行綁定, 這樣就可以模擬實現單線程處理用戶的請求, 在並發量極高的時候, 可以極大避免各種鎖帶來的開銷。並且處理能力也可以提升30%。當然這樣並不是通用的處理方案, 是基於特殊的業務場景做特殊的實現。也是對我們的一種啟發。

漸進式rehash

當資料庫中存在上百萬的key, 占用數GB內存的時候, 需要擴容就是一件非常耗時的操作, 同時CPU使用率也會飆升。對此redis使用了漸進式的擴容策略, 複習一下redishashtable的結構:

#dict字典的數據結構typedef struct dict{ dictType *type; //直線dictType結構，dictType結構中包含自定義的函數，這些函數使得key和value能夠存儲任何類型的數據 void *privdata; //私有數據，保存著dictType結構中函數的 參數 dictht ht[2]; //兩張哈希表 long rehashidx; //rehash的標記，rehashidx=-1表示沒有進行rehash，rehash時每遷移一個桶就對rehashidx加一 int itreators; //正在疊代的疊代器數量} #dict結構中ht[0]、ht[1]哈希表的數據結構typedef struct dictht{ dictEntry[] table; //存放一個數組的地址，數組中存放哈希節點dictEntry的地址 unsingned long size; //哈希表table的大小，出始大小為4 unsingned long sizemask; //用於將hash值映射到table位置的索引，大小為（size-1） unsingned long used; //記錄哈希表已有節點（鍵值對）的數量}

當達到擴容條件時, 會將rehashidx置為0標識rehash開始, 在漸進式rehash進行期間，字典的刪除、查找、更新等操作會在兩個哈希表上進行。比如說，要在字典裡面查找一個鍵的話，程序會先在ht[0]裡面進行查找，如果沒找到的話，就會繼續到ht[1]裡面進行查找, 新添加到字典的鍵值對一律會被保存到 ht[1] 裡面, 以此操作在某一個時間點, 會將ht[0]的數據遷移完成, 並且將ht[1]置為ht[0], ht[1]置為NULL, 為下一次rehash做準備。
在一次和新浪同學的交流中, 發現他們也是使用了類似的方法來更新本地緩存, 我們知道新浪曾經是最大的redis集群, 在熱點新聞爆發的時候點讚, 評論, 轉發，熱點新聞等會量級會指數量級攀升, 如果使用Redis性能也是遠遠不夠, 所以他們同時也使用了大量的本地緩存, 使用本地緩存就要涉及到緩存的及時更新, 在更新大量本地緩存是也同樣不可避免的要涉及到讀寫鎖開銷, 大量擴容帶來的CPU消耗等, 為此他們也是採用了兩套map來規避此類問題, 業務模型大致如下:

當請求到來的時候會優先讀取本地緩存(內存)數據, 同時本地緩存會有各種來源的設置, 當數據量比較大的時候就會有非常大的cpu抖動, 同時導致請求出現較大的波動，造成不好的體驗，所以在結構中有兩個map, 一個為只讀map, 另一個為可寫map, 讀取緩存時只讀取只讀map，更新緩存時, 將所有緩存寫入到可寫map, 更新寫入完成後, 將最新緩存map加速賦值給只讀map, 同時清空更新map, 偽代碼如下:

package mainimport "sync"type localcache struct { lock sync.RWMutex Cache map[string]interface{} update map[string]interface{}}func (c *localCache) Get(key string) interface{} { c.lock.RLock() defer c.lock.RUnlock() cache, ok := c.Cache[key] if ok { return cache } return nil}func (c *localCache) Set(key string, val interface{}) { c.update[key] = val}// 初始化待更新數據的長度func (c *localCache) MakeUpdateCache(len int) { c.update = make(map[string]interface{}, len)}func (c *localCache) Rehash() { c.lock.RLock() defer c.lock.RUnlock() c.Cache = c.update c.MakeUpdateCache(0)}

這樣只會在更新db時不會頻繁的對於一個map進行擴容操作, 同時可以保證Cache欄位中不會存在歷史的髒數據。只有在執行Rehash()方法完成時才會執行內存分配的動作, 而且在讀多寫少的情況下, RWMutex性能可以達到互斥鎖的8倍左右(參考Go 語言高性能編程)。這個是基於漸進式rehash可以帶給我的關於自己做本地緩存的思考, 但是這個方案我自己並未實踐, 所以各種細節考慮還有待改進, 也歡迎討論。

動態內存分配

分享一個工作中的問題, 有一個case現象是當請求並發量比較大的時候會出現cpu使用率飆升, 查看了相關接口的業務邏輯相對比較簡單, 從磁碟中加載數據到內存map中, 然後返回map中的key。然後開始查看系統調用關係, 發現有非常對的cpu耗時在php的內存分配上mmap, debug發現使用了php數組的+=操作進行對象合併, 而且每次操作數據比較大, 所以造成了cpu使用率較高, 解決方案也很簡單, 可以把類似的操作改為共享內存操作, 通過lstat調用我們已經獲取需要保存文件的總體大小, 我們只需要一次性開闢一塊內存存儲即可, 可以使用共享內存操作:

$shmid = shmop_open($systemid, $mode, $permissions, $size);

參數四是需要開闢的內存大小, 即可減少內存分配的次數, 降低頻繁的內存分配。

數據結構

Redis中實現了很多優秀的數據結構, 有些不僅僅是典型的實現, 而是根據實際需求做了改進, 如Skiplist中加入了向後的指針等。關於數據結構及算法設計可能在大家平常的工作中並不會有太多時間去深入思考與應用, 業界也有比較成熟, 可靠的方案可借鑑, 但其實如果拋開現有的方案, 我們需要的數據結構及算法在工作中處處可見, 如父子節點的組織構造與查找, 自己構造權重隊列等場景。

前一段時間工作有一個場景是給客戶端下發一組原因選項, 每個選項下面可能存在一個二級選項, 所以這個場景可以是一個無限極分類, 初次的實現中是通過遞歸調用來查找的。後面優化使用樹的深度遍歷來查找, 個人思考是能夠使用合適的數據結構實現不同的業務有助於提升自己的思維能力。

總結

至此關於Redis6.0.0版本的源碼分析就準備告一個段落, 在此過程中發現對自己的一個鞭策是除了學習知識, 還需要通過思考吸收和消化知識, 後續的學習中我們也同樣要有執行力和思考力 :)

作者：cooper_li
連結：https://juejin.cn/post/7105697337615319070