【Redis5源碼學習】淺析redis命令之randomkey篇

baiyan

命令語法

命令含義：從當前選定數據庫隨機返回一個key
命令格式：

RANDOMKEY

命令實戰：

127.0.0.1:6379> keys *
1) "kkk"
2) "key1"

127.0.0.1:6379> randomkey
"key1"
127.0.0.1:6379> randomkey
"kkk"

返回值： 隨機的鍵；若是數據庫爲空則返回nil

源碼分析

主體流程

keys命令對應的處理函數是randomKeyCommand()：

void randomkeyCommand(client *c) {
    robj *key; // 存儲獲取到的key

    if ((key = dbRandomKey(c->db)) == NULL) { // 調用核心函數dbRandomKey()
        addReply(c,shared.nullbulk); // 返回nil
        return;
    }

    addReplyBulk(c,key); // 返回key
    decrRefCount(key); // 減小引用計數
}

隨機鍵生成以及過時判斷

randomKeyCommand()調用了dbRandomKey()函數來真正生成一個隨機鍵：

robj *dbRandomKey(redisDb *db) {
    dictEntry *de;
    int maxtries = 100;
    int allvolatile = dictSize(db->dict) == dictSize(db->expires);

    while(1) {
        sds key;
        robj *keyobj;

        de = dictGetRandomKey(db->dict); // 獲取隨機的一個dictEntry
        if (de == NULL) return NULL; // 獲取失敗返回NULL

        key = dictGetKey(de); // 獲取dictEntry中的key
        keyobj = createStringObject(key,sdslen(key)); // 根據key字符串生成robj
        if (dictFind(db->expires,key)) { // 去過時字典裏查找這個鍵
            ...
            if (expireIfNeeded(db,keyobj)) { // 判斷鍵是否過時
                decrRefCount(keyobj); // 若是過時了，刪掉這個鍵並減小引用計數
                continue; // 當前鍵過時了不能返回，只返回不過時的鍵，進行下一次隨機生成
            }
        }
        return keyobj;
    }
}

那麼這一層的主邏輯又調用了dictGetRandomKey()，獲取隨機的一個dictEntry。假設咱們已經獲取到了隨機生成的dictEntry，咱們隨後取出key。因爲不能返回過時的key，因此咱們須要先判斷鍵是否過時，若是過時了就不能返回了，直接continue；若是不過時就能夠返回。

真正獲取隨機鍵的算法

那麼咱們繼續跟進dictGetRandomKey()函數，看一下究竟使用了什麼算法，來隨機生成dictEntry：

dictEntry *dictGetRandomKey(dict *d)
{
    dictEntry *he, *orighe;
    unsigned long h;
    int listlen, listele;

    if (dictSize(d) == 0) return NULL; // 傳進來的字典爲空，根本不用生成
    if (dictIsRehashing(d)) _dictRehashStep(d); // 執行一次rehash操做
    if (dictIsRehashing(d)) { // 若是正在rehash，注意要保證從兩個哈希表中均勻分配隨機種子
        do {
            h = d->rehashidx + (random() % (d->ht[0].size +d->ht[1].size - d->rehashidx)); //計算隨機哈希值，這個哈希值必定是在rehashidx的後部
            he = (h >= d->ht[0].size) ? d->ht[1].table[h - d->ht[0].size] : d->ht[0].table[h];// 根據上面計算的哈希值拿到對應的bucket
        } while(he == NULL); // 一直循環計算，取最後一個計算結果不爲空的bucket
    } else { // 不在rehash，只有一個哈希表
        do {
            h = random() & d->ht[0].sizemask; // 直接計算哈希值
            he = d->ht[0].table[h]; // 取出哈希表上第h個bucket
        } while(he == NULL); // 一直循環計算，取最後一個計算結果不爲空的bucket
    }

    // 如今咱們獲得了一個不爲空的bucket，而這個bucket的後面還掛接了一個或多個dictEntry（鏈地址法解決哈希衝突），因此一樣須要計算一個隨機索引，來判斷究竟訪問哪個dickEntry鏈表結點
    listlen = 0;
    orighe = he;
    while(he) {
        he = he->next;
        listlen++; // 計算鏈表長度
    }
    listele = random() % listlen; // 隨機數對鏈表長度取餘，肯定獲取哪個結點
    he = orighe;
    while(listele--) he = he->next; // 從前到後遍歷這個bucket上的鏈表，找到這個結點
    return he; // 最終返回這個結點
}

這個函數首先會進行字典爲空的判斷。而後會進行一個單步rehash操做，這一點和調用如dictAdd()等字典函數的效果是同樣的，都是漸進式rehash技術的一部分。在這裏咱們首先複習一下字典的總體結構：

因爲rehash會影響隨機數種子的生成，因此根據當前字典是否正在進行rehash操做，須要分兩種狀況討論：
第一種：正在進行rehash操做。 那麼當前字典的結構爲：有一部分鍵在第一個哈希表上、其他的鍵在第二個哈希表上。爲了均勻分配兩個哈希表可能被取到的機率，須要將兩個哈希表結合考慮。其算法爲：

h = d->rehashidx + (random() % (d->ht[0].size + d->ht[1].size - d->rehashidx)); //計算隨機哈希值，這個哈希值必定是在rehashidx的後部

這裏將一個隨機數對兩個哈希表大小之和減去rehashidx取餘。這樣的取餘操做能夠保證這個哈希值會隨機落在索引大於rehashidx位置的bucket上。由於rehashidx表示rehash的進度。這個rehashidx表示在第一個哈希表上在這個索引以前的數據，即[0, rehashidx-1]，這個閉區間上的數據已經在被rehash到第二個哈希表上了。而大於等於這個rehashidx的元素仍在第一個哈希表上。因此，這樣就保證了任何一個結果h上的bucket，都是非空有值的。接下來只須要判斷這個h值在哪一個哈希表上，而後去哈希表上對應位置上的bucket取值便可：

he = (h >= d->ht[0].size) ? d->ht[1].table[h - d->ht[0].size] : d->ht[0].table[h];

第二種：沒有進行rehash操做。 那麼全部鍵都在惟一一個第一個字典上，這種狀況就很是簡單了，能夠直接對字典長度求餘，或者對字典的sizemask進行按位與運算，均可以保證計算後的結果落在哈希表內。redis選擇的是後者：

h = random() & d->ht[0].sizemask; // 經過對sizemask的按位與運算計算哈希值
he = d->ht[0].table[h]; // 取出哈希表上第h個bucket

接下來，咱們找到了一個非空的bucket，可是尚未結束。因爲可能存在哈希衝突，redis採用鏈地址法解決哈希衝突，因此會在一個bucket後面掛接多個dictEntry，造成一個鏈表。因此，還須要思考究竟要取哪個鏈表結點上的dictEntry。這個算法就比較簡單了，直接利用random()的結果，對鏈表長度求餘便可：

listele = random() % listlen; // 隨機數對鏈表長度取餘，肯定獲取哪個結點
while(listele--) he = he->next; // 從前到後遍歷這個bucket上的鏈表，找到這個結點

到此爲止，咱們就找到了一個隨機bucket上的一個隨機dictEntry結點，那麼就能夠返回給客戶端啦。