Redis數據庫

時間 2019-11-19

原文原文鏈接

1 Redis數據庫簡介

　　redis的全部數據庫都是保存在redisServer結構體的db數組中，db數組中的每一個元素都是redisDb結構體表明一個數據庫。db.c主要是封裝了數據庫的底層操做實現，操做dict和expires兩個字典。全部數據庫的鍵值隊都是保存在dict字典(即內存中)，而expires用來保存設置過時時間的鍵。redis用rdb和aof來進行數據的持久化，僅僅是內存數據備份，當內存不足時，經過內存策略刪除部分鍵值對，不會持久化要刪除的的鍵值隊。redis

struct redisServer {  
    ……  
    redisDb *db;//redis server的全部數據褲  
    int dbnum;//redis server的數據庫的個數  
    ……  
};

/* Redis database representation. There are multiple databases identified
 * by integers from 0 (the default database) up to the max configured
 * database. The database number is the 'id' field in the structure. */
typedef struct redisDb {

    // 數據庫鍵空間，保存着數據庫中的全部鍵值對
    dict *dict;                 /* The keyspace for this DB */

    // 鍵的過時時間，字典的鍵爲鍵，字典的值爲過時事件 UNIX 時間戳
    dict *expires;              /* Timeout of keys with a timeout set */

    // 正處於阻塞狀態的鍵
    dict *blocking_keys;        /* Keys with clients waiting for data (BLPOP) */

    // 能夠解除阻塞的鍵
    dict *ready_keys;           /* Blocked keys that received a PUSH */

    // 正在被 WATCH 命令監視的鍵，即事務中被監視的key
    dict *watched_keys;         /* WATCHED keys for MULTI/EXEC CAS */

    struct evictionPoolEntry *eviction_pool;    /* Eviction pool of keys */

    // 數據庫號碼
    int id;                     /* Database ID */

    // 數據庫的鍵的平均 TTL ，統計信息
    long long avg_ttl;          /* Average TTL, just for stats */

} redisDb;

　　下圖是一個RedisDb的示例，該數據庫存放有五個鍵值對，分別是sRedis，INums，hBooks，SortNum和sNums，它們各自都有本身的值對象，另外，其中有三個鍵設置了過時時間，當前數據庫是服務器的第0號數據庫。算法

2 Redis數據庫的實現

2.1 Redis數據庫的切換

每個數據庫的結構體都有一個id用來標識該數據庫的編號，Redis的配置文件redis.conf中提供了以下參數來控制Redis在初始化的時候須要建立多少個數據庫。數據庫

databases 16  // 表示該服務器須要建立16個數據庫

Redis提供了SELECT命令，來選擇當前使用的數據庫。其操做以下：數組

127.0.0.1:6379> select 1  // 初始爲0號數據庫，此時選擇編碼爲1的數據庫
OK
127.0.0.1:6379[1]> select 2 // [1]表明當前數據庫編號，此時選擇數據庫2
OK
127.0.0.1:6379[2]> // [2]表明當前數據庫編號爲2

客戶端redisClient結構體的db屬性記錄了客戶端的當前目標數據庫，若是客戶端的目標數據庫爲1號數據庫，則將指向redisServer中db[1]. 　　服務器

/* selectDb source code
 * 將客戶端的目標數據庫切換爲 id 所指定的數據庫
 */
int selectDb(redisClient *c, int id) {

    // 確保 id 在正確範圍內
    if (id < 0 || id >= server.dbnum)
        return REDIS_ERR;

    // 切換數據庫（更新客戶端數據庫指針）
    c->db = &server.db[id];

    return REDIS_OK;
}

2.2 Redis 鍵空間的操做(redisDb.dict)

Redis數據庫中存放的數據都是以鍵值對形式存在，其充分利用了字典結構的高效索引特性，其中：dom

字典的鍵：一般是一個字符串對象
字典的值：但是是字符串，哈希，鏈表，集合和有序集合

Redis爲數據庫的鍵空間操做提供了下列操做函數：ide

/* 從數據庫中取出指定鍵對應的值對象，如不存在則返回NULL */ 
robj *lookupKey(redisDb *db, robj *key, int flags);
/* 先刪除過時鍵，再從數據庫中取出指定鍵對應的值對象，如不存在則返回NULL 
 * 底層調用lookupKey函數
 */
robj *lookupKeyReadWithFlags(redisDb *db, robj *key, int flags);
/* 先刪除過時鍵，以讀操做的方式從數據庫中取出指定鍵對應的值對象
 * 如不存在則返回NULL，底層調用lookupKey函數
 */
robj *lookupKeyRead(redisDb *db, robj *key);
/* 先刪除過時鍵，以寫操做的方式從數據庫中取出指定鍵對應的值對象
 * 如不存在則返回NULL，底層調用lookupKeyReadWithFlags函數
 */
robj *lookupKeyWrite(redisDb *db, robj *key);
/* 先刪除過時鍵，以讀操做的方式從數據庫中取出指定鍵對應的值對象
 * 如不存在則返回NULL，底層調用lookupKeyRead函數
 * 此操做須要向客戶端回覆
 */
robj *lookupKeyReadOrReply(client *c, robj *key, robj *reply);
/* 先刪除過時鍵，以寫操做的方式從數據庫中取出指定鍵對應的值對象
 * 如不存在則返回NULL，底層調用lookupKeyWrite函數
 * 此操做須要向客戶端回覆
 */
robj *lookupKeyWriteOrReply(client *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*));

2.3 數據庫過時鍵空間的操做（db->expires）

（1）過時鍵刪除策略函數

若是一個鍵設置了刪除時間，那麼面臨的問題是以怎樣的策略去刪除該鍵。咱們很容易理解下面三個刪除策略：源碼分析

定時刪除：若是一個鍵設置了過時時間，就爲其建立一個定時器，在定時器結束時，馬上對該鍵執行刪除操做
惰性刪除：在訪問該鍵時，判斷其過時時間是否到了，若是已過時，則執行刪除操做
按期刪除：每一個一段時間，對數據庫中的鍵進行一次遍歷，刪除其中的一些過時鍵，至於刪除多少過時鍵以及檢查檢查多少個數據庫，則由算法決定

　　其中，定時刪除能夠及時的刪除過時鍵，但它爲每個設定了過時時間的鍵都開了一個定時器，使得CPU的負載變高，從而致使服務器的響應時間和吞吐量收到影響。this

　　惰性刪除有效的克服了定時刪除對CPU的影響，可是，若是一個鍵長時間沒有被訪問，且這個鍵已通過期好久了，顯然，大量的過時鍵會佔用內存，從而致使內存上的消耗過大。

　　按期刪除能夠算是上述兩種策略的折中。設定一個定時器，每隔一段時間遍歷數據庫，刪除其中的過時鍵，有效的緩解了定時刪除對CPU的佔用以及惰性刪除對內存的佔用。

　　Redis採用了惰性刪除和按期刪除兩種策略來對過時鍵進行處理，在上面的lookupKeyWrite等函數中就利用到了惰性刪除策略，定時刪除策略則是在根據服務器的例行處理程序serverCron來執行刪除操做，該程序每100ms調用一次。

（2）惰性刪除

惰性刪除由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);
}

int removeExpire(redisDb *db, robj *key);//從expires中刪除key的信息  
void setExpire(redisDb *db, robj *key, long long when);//設置key的過時時間  
long long getExpire(redisDb *db, robj *key);//獲取key的過時時間  

//過時健傳播  
void propagateExpire(redisDb *db, robj *key, int lazy) {  
    robj *argv[2];  
    argv[0] = lazy ? shared.unlink : shared.del;  
    argv[1] = key;  
    incrRefCount(argv[0]);  
    incrRefCount(argv[1]);  
    if (server.aof_state != AOF_OFF)  
        //將過時健以刪除命令寫入aof文件  
        feedAppendOnlyFile(server.delCommand,db->id,argv,2);  
    //傳播過時健給slave  
    replicationFeedSlaves(server.slaves,db->id,argv,2);  
    decrRefCount(argv[0]);  
    decrRefCount(argv[1]);  
}

（3）按期刪除

　　Redis定義了一個例行處理程序serverCron，該程序每隔100ms執行一次，在其執行過程當中會調用databasesCron函數，這個函數裏面纔會調用真正的按期刪除函數activeExpireCycle。該函數每次執行時遍歷指定個數的數據庫，而後從expires字典中隨機取出一個帶過時時間的鍵，檢查它是否過時，如過時直接刪除。每隔100ms處理數據庫的個數由CRON_DBS_PER_CALL參數決定，該參數的默認值以下：

#define CRON_DBS_PER_CALL 16  // 每次處理16個數據庫

刪除過時鍵的操做由activeExpireCycleTryExpire函數執行，其源碼以下：

/* 檢查鍵的過時時間，如過時直接刪除*/
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;
    }
}