探索Redis設計與實現9：數據庫redisDb與鍵過時刪除策略

本文轉自互聯網

本系列文章將整理到我在GitHub上的《Java面試指南》倉庫，更多精彩內容請到個人倉庫裏查看

https://github.com/h2pl/Java-Tutorial 喜歡的話麻煩點下Star哈

文章首發於個人我的博客：

www.how2playlife.com 本文是微信公衆號【Java技術江湖】的《探索Redis設計與實現》其中一篇，本文部份內容來源於網絡，爲了把本文主題講得清晰透徹，也整合了不少我認爲不錯的技術博客內容，引用其中了一些比較好的博客文章，若有侵權，請聯繫做者。

該系列博文會告訴你如何從入門到進階，Redis基本的使用方法，Redis的基本數據結構，以及一些進階的使用方法，同時也須要進一步瞭解Redis的底層數據結構，再接着，還會帶來Redis主從複製、集羣、分佈式鎖等方面的相關內容，以及做爲緩存的一些使用方法和注意事項，以便讓你更完整地瞭解整個Redis相關的技術體系，造成本身的知識框架。

若是對本系列文章有什麼建議，或者是有什麼疑問的話，也能夠關注公衆號【Java技術江湖】聯繫做者，歡迎你參與本系列博文的創做和修訂。

一. 數據庫 Redis的數據庫使用字典做爲底層實現，數據庫的增、刪、查、改都是構建在字典的操做之上的。 redis服務器將全部數據庫都保存在服務器狀態結構redisServer(redis.h/redisServer)的db數組（應該是一個鏈表）裏：

struct redisServer { //.. // 數據庫數組，保存着服務器中全部的數據庫 redisDb *db; //.. } 在初始化服務器時，程序會根據服務器狀態的dbnum屬性來決定應該建立多少個數據庫：

struct redisServer { // .. //服務器中數據庫的數量 int dbnum; //.. } dbnum屬性的值是由服務器配置的database選項決定的，默認值爲16；

2、切換數據庫原理 每一個Redis客戶端都有本身的目標數據庫，每當客戶端執行數據庫的讀寫命令時，目標數據庫就會成爲這些命令的操做對象。

127.0.0.1:6379> set msg 'Hello world' OK 127.0.0.1:6379> get msg "Hello world" 127.0.0.1:6379> select 2 OK 127.0.0.1:6379[2]> get msg (nil) 127.0.0.1:6379[2]> 在服務器內部，客戶端狀態redisClient結構(redis.h/redisClient)的db屬性記錄了客戶端當前的目標數據庫，這個屬性是一個指向redisDb結構(redis.h/redisDb)的指針：

typedef struct redisClient { //.. // 客戶端當前正在使用的數據庫 redisDb *db; //.. } redisClient; redisClient.db指針指向redisServer.db數組中的一個元素，而被指向的元素就是當前客戶端的目標數據庫。 咱們就能夠經過修改redisClient指針，讓他指向服務器中的不一樣數據庫，從而實現切換數據庫的功能–這就是select命令的實現原理。 實現代碼：

int selectDb(redisClient *c, int id) { // 確保 id 在正確範圍內 if (id < 0 || id >= server.dbnum) return REDIS_ERR; // 切換數據庫（更新指針） c->db = &server.db[id]; return REDIS_OK; } 3、數據庫的鍵空間 一、數據庫的結構（咱們只分析鍵空間和鍵過時時間） typedef struct redisDb { // 數據庫鍵空間，保存着數據庫中的全部鍵值對 dict dict; / The keyspace for this DB */ // 鍵的過時時間，字典的鍵爲鍵，字典的值爲過時事件 UNIX 時間戳 dict expires; / Timeout of keys with a timeout set / // 數據庫號碼 int id; / Database ID / // 數據庫的鍵的平均 TTL ，統計信息 long long avg_ttl; / Average TTL, just for stats */ //.. } redisDb

上圖是一個RedisDb的示例，該數據庫存放有五個鍵值對，分別是sRedis，INums，hBooks，SortNum和sNums，它們各自都有本身的值對象，另外，其中有三個鍵設置了過時時間，當前數據庫是服務器的第0號數據庫。如今，咱們就從源碼角度分析這個數據庫結構： 咱們知道，Redis是一個鍵值對數據庫服務器，服務器中的每個數據庫都是一個redis.h/redisDb結構，其中，結構中的dict字典保存了數據庫中全部的鍵值對，咱們就將這個字典成爲鍵空間。 Redis數據庫的數據都是以鍵值對的形式存在，其充分利用了字典高效索引的特色。 a、鍵空間的鍵就是數據庫中的鍵，通常都是字符串對象； b、鍵空間的值就是數據庫中的值，能夠是5種類型對象（字符串、列表、哈希、集合和有序集合）之一。 數據庫的鍵空間結構分析完了，咱們先看看數據庫的初始化。

二、鍵空間的初始化 在redis.c中，咱們能夠找到鍵空間的初始化操做：

//建立並初始化數據庫結構 for (j = 0; j < server.dbnum; j++) { // 建立每一個數據庫的鍵空間 server.db[j].dict = dictCreate(&dbDictType,NULL); // ... // 設定當前數據庫的編號 server.db[j].id = j; } 初始化以後就是對鍵空間的操做了。

三、鍵空間的操做 我先把一些常見的鍵空間操做函數列出來：

// 從數據庫中取出鍵key的值對象，若不存在就返回NULL robj *lookupKey(redisDb *db, robj *key);

/* 先刪除過時鍵，以讀操做的方式從數據庫中取出指定鍵對應的值對象

• 並根據是否成功找到值，更新服務器的命中或不命中信息,
• 如不存在則返回NULL，底層調用lookupKey函數 */ robj *lookupKeyRead(redisDb *db, robj *key);

/* 先刪除過時鍵，以寫操做的方式從數據庫中取出指定鍵對應的值對象

• 如不存在則返回NULL，底層調用lookupKey函數，
• 不會更新服務器的命中或不命中信息 */ robj *lookupKeyWrite(redisDb *db, robj *key);

/* 先刪除過時鍵，以讀操做的方式從數據庫中取出指定鍵對應的值對象

• 如不存在則返回NULL，底層調用lookupKeyRead函數
• 此操做須要向客戶端回覆 */ robj *lookupKeyReadOrReply(redisClient *c, robj *key, robj *reply);

/* 先刪除過時鍵，以寫操做的方式從數據庫中取出指定鍵對應的值對象

• 如不存在則返回NULL，底層調用lookupKeyWrite函數
• 此操做須要向客戶端回覆 */ robj *lookupKeyWriteOrReply(redisClient *c, robj *key, robj *reply);

/* 添加元素到指定數據庫 */ void dbAdd(redisDb *db, robj *key, robj val); / 重寫指定鍵的值 */ void dbOverwrite(redisDb *db, robj *key, robj val); / 設定指定鍵的值 */ void setKey(redisDb *db, robj *key, robj val); / 判斷指定鍵是否存在 */ int dbExists(redisDb *db, robj key); / 隨機返回數據庫中的鍵 */ robj *dbRandomKey(redisDb db); / 刪除指定鍵 */ int dbDelete(redisDb *db, robj key); / 清空全部數據庫，返回鍵值對的個數 / long long emptyDb(void(callback)(void)); 下面我選取幾個比較典型的操做函數分析一下：

查找鍵值對函數–lookupKey robj *lookupKey(redisDb *db, robj *key) { // 查找鍵空間 dictEntry *de = dictFind(db->dict,key->ptr); // 節點存在 if (de) { // 取出該鍵對應的值 robj *val = dictGetVal(de); // 更新時間信息 if (server.rdb_child_pid == -1 && server.aof_child_pid == -1) val->lru = LRU_CLOCK(); // 返回值 return val; } else { // 節點不存在 return NULL; } }

添加鍵值對–dbAdd 添加鍵值對使咱們常用到的函數，底層由dbAdd()函數實現，傳入的參數是待添加的數據庫，鍵對象和值對象，源碼以下：

void dbAdd(redisDb *db, robj *key, robj *val) { // 複製鍵名 sds copy = sdsdup(key->ptr); // 嘗試添加鍵值對 int retval = dictAdd(db->dict, copy, val); // 若是鍵已經存在，那麼中止 redisAssertWithInfo(NULL,key,retval == REDIS_OK); // 若是開啓了集羣模式，那麼將鍵保存到槽裏面 if (server.cluster_enabled) slotToKeyAdd(key); }

好了，關於鍵空間操做函數就分析到這，其餘函數(在文件db.c中)你們能夠本身去分析，有問題的話能夠回帖，咱們能夠一塊兒討論！

4、數據庫的過時鍵操做 在前面咱們說到，redisDb結構中有一個expires指針（概況圖能夠看上圖），該指針指向一個字典結構，字典中保存了全部鍵的過時時間，該字典稱爲過時字典。 過時字典的初始化：

// 建立並初始化數據庫結構 for (j = 0; j < server.dbnum; j++) { // 建立每一個數據庫的過時時間字典 server.db[j].expires = dictCreate(&keyptrDictType,NULL); // 設定當前數據庫的編號 server.db[j].id = j; // .. }

a、過時字典的鍵是一個指針，指向鍵空間中的某一個鍵對象（就是某一個數據庫鍵）； b、過時字典的值是一個long long類型的整數，這個整數保存了鍵所指向的數據庫鍵的時間戳–一個毫秒精度的unix時間戳。 下面咱們就來分析過時鍵的處理函數：

一、過時鍵處理函數 設置鍵的過時時間–setExpire() /*

將鍵 key 的過時時間設爲 when */ void setExpire(redisDb *db, robj *key, long long when) { dictEntry *kde, *de; // 從鍵空間中取出鍵key kde = dictFind(db->dict,key->ptr); // 若是鍵空間找不到該鍵，報錯 redisAssertWithInfo(NULL,key,kde != NULL); // 向過時字典中添加該鍵 de = dictReplaceRaw(db->expires,dictGetKey(kde)); // 設置鍵的過時時間 // 這裏是直接使用整數值來保存過時時間，不是用 INT 編碼的 String 對象 dictSetSignedIntegerVal(de,when); } 獲取鍵的過時時間–getExpire() long long getExpire(redisDb *db, robj *key) { dictEntry *de; // 若是過時鍵不存在，那麼直接返回 if (dictSize(db->expires) == 0 || (de = dictFind(db->expires,key->ptr)) == NULL) return -1; redisAssertWithInfo(NULL,key,dictFind(db->dict,key->ptr) != NULL); // 返回過時時間 return dictGetSignedIntegerVal(de); }

刪除鍵的過時時間–removeExpire() // 移除鍵 key 的過時時間 int removeExpire(redisDb *db, robj *key) { // 確保鍵帶有過時時間 redisAssertWithInfo(NULL,key,dictFind(db->dict,key->ptr) != NULL); // 刪除過時時間 return dictDelete(db->expires,key->ptr) == DICT_OK; }

二、過時鍵刪除策略 經過前面的介紹，你們應該都知道數據庫鍵的過時時間都保存在過時字典裏，那假如一個鍵過時了，那麼這個過時鍵是何時被刪除的呢？如今來看看redis的過時鍵的刪除策略： a、定時刪除：在設置鍵的過時時間的同時，建立一個定時器，在定時結束的時候，將該鍵刪除； b、惰性刪除：聽任鍵過時無論，在訪問該鍵的時候，判斷該鍵的過時時間是否已經到了，若是過時時間已經到了，就執行刪除操做； c、按期刪除：每隔一段時間，對數據庫中的鍵進行一次遍歷，刪除過時的鍵。 其中定時刪除能夠及時刪除數據庫中的過時鍵，並釋放過時鍵所佔用的內存，可是它爲每個設置了過時時間的鍵都開了一個定時器，使的cpu的負載變高，會對服務器的響應時間和吞吐量形成影響。 惰性刪除有效的克服了定時刪除對CPU的影響，可是，若是一個過時鍵很長時間沒有被訪問到，且若存在大量這種過時鍵時，勢必會佔用很大的內存空間，致使內存消耗過大。 定時刪除能夠算是上述兩種策略的折中。設定一個定時器，每隔一段時間遍歷數據庫，刪除其中的過時鍵，有效的緩解了定時刪除對CPU的佔用以及惰性刪除對內存的佔用。 在實際應用中，Redis採用了惰性刪除和定時刪除兩種策略來對過時鍵進行處理，上面提到的lookupKeyWrite等函數中就利用到了惰性刪除策略，定時刪除策略則是在根據服務器的例行處理程序serverCron來執行刪除操做，該程序每100ms調用一次。

惰性刪除函數–expireIfNeeded() 源碼以下：

/* 檢查key是否已通過期，若是是的話，將它從數據庫中刪除

並將刪除命令寫入AOF文件以及附屬節點(主從複製和AOF持久化相關) 返回0表明該鍵尚未過時，或者沒有設置過時時間 返回1表明該鍵由於過時而被刪除 */ int expireIfNeeded(redisDb *db, robj *key) { // 獲取該鍵的過時時間 mstime_t when = getExpire(db,key); mstime_t now; // 該鍵沒有設定過時時間 if (when < 0) return 0; // 服務器正在加載數據的時候，不要處理 if (server.loading) return 0; // lua腳本相關 now = server.lua_caller ? server.lua_time_start : mstime(); // 主從複製相關，附屬節點不主動刪除key if (server.masterhost != NULL) return now > when; // 該鍵尚未過時 if (now <= when) return 0; // 刪除過時鍵 server.stat_expiredkeys++; // 將刪除命令傳播到AOF文件和附屬節點 propagateExpire(db,key); // 發送鍵空間操做時間通知 notifyKeyspaceEvent(NOTIFY_EXPIRED, "expired",key,db->id); // 將該鍵從數據庫中刪除 return dbDelete(db,key); } 按期刪除策略 過時鍵的按期刪除策略由redis.c/activeExpireCycle()函數實現，服務器週期性地操做redis.c/serverCron()（每隔100ms執行一次）時，會調用activeExpireCycle()函數，分屢次遍歷服務器中的各個數據庫，從數據庫中的expires字典中隨機檢查一部分鍵的過時時間，並刪除其中的過時鍵。 刪除過時鍵的操做由activeExpireCycleTryExpire函數(activeExpireCycle()調用了該函數)執行，其源碼以下：

/* 檢查鍵的過時時間，如過時直接刪除*/ int activeExpireCycleTryExpire(redisDb *db, dictEntry *de, long long now) { // 獲取過時時間 long long t = dictGetSignedIntegerVal(de); if (now > t) { // 執行到此說明過時 // 建立該鍵的副本 sds key = dictGetKey(de); robj *keyobj = createStringObject(key,sdslen(key)); // 將刪除命令傳播到AOF和附屬節點 propagateExpire(db,keyobj); // 在數據庫中刪除該鍵 dbDelete(db,keyobj); // 發送事件通知 notifyKeyspaceEvent(NOTIFY_EXPIRED, "expired",keyobj,db->id); // 臨時鍵對象的引用計數減1 decrRefCount(keyobj); // 服務器的過時鍵計數加1 // 該參數影響每次處理的數據庫個數 server.stat_expiredkeys++; return 1; } else { return 0; } }

刪除過時鍵對AOF、RDB和主從複製都有影響，等到了介紹相關功能時再討論。 今天就先到這裏~