走近源碼：Redis如何清除過時key

「叮……」，美好的週六就這麼被一陣釘釘消息吵醒了。

業務組的同窗告訴我說不少用戶的賬號今天被強制下線。咱們的賬號系統正常的邏輯是用戶登陸一次後，token的有效期能夠維持一天的時間。如今的問題是用戶大概每10分鐘左右就須要從新登陸一次。這種狀況通常有兩種緣由：一、token生成時出問題。二、驗證token時出現問題。

經過檢查日誌，我發現是驗證token時，Redis中已經沒有對應的token了。而且肯定了生成新的token時，set到Redis中的有效期是正確的，那麼就基本能夠肯定是Redis的問題了。

因而又去檢查了Redis的監控，發如今那段時間Redis因爲內存佔用太高強制清理了幾回key。但從日誌上來看，這段時間並無出現流量暴漲的狀況，並且Redis中key的數量也沒有顯著增長。那是什麼緣由致使Redis內存佔用太高呢？肯定了Redis內存升高不是咱們形成的以後，咱們又聯繫了業務組的同窗協助他們，他們表示最近確實有上線，而且新上線的功能有使用到Redis。但我仍然感受很奇怪，爲何Redis中的key沒有增多，而且沒看到有其餘業務的key。通過一番詢問，才瞭解到，業務組同窗使用的是這個Redis的db1，而我用的（和剛查的）是db0。這裏確實是我在排查問題時出現了疏忽。

那麼Redis的不一樣db之間會互相影響嗎？一般狀況下，咱們使用不一樣的db進行數據隔離，這沒問題。但Redis進行清理時，並非只清理數據量佔用最大的那個db，而是會對全部的db進行清理。在這以前我並非很瞭解這方面知識，這裏也只是根據現象進行的猜想。

好奇心驅使我來驗證一下這個想法。因而我決定直接來看Redis的源碼。清理key相關的代碼在evict.c文件中。

Redis中會保存一個「過時key池」，這個池子中存放了一些可能會被清理的key。其中保存的數據結構以下：

struct evictionPoolEntry {
    unsigned long long idle;    /* Object idle time (inverse frequency for LFU) */
    sds key;                    /* Key name. */
    sds cached;                 /* Cached SDS object for key name. */
    int dbid;                   /* Key DB number. */
};

其中idle是對象空閒時間，在Reids中，key的過時算法有兩種：一種是近似LRU，一種是LFU。默認使用的是近似LRU。

近似LRU

在解釋近似LRU以前，先來簡單瞭解一下LRU。當Redis的內存佔用超過咱們設置的maxmemory時，會把長時間沒有使用的key清理掉。按照LRU算法，咱們須要對全部key（也能夠設置成只淘汰有過時時間的key）按照空閒時間進行排序，而後淘汰掉空閒時間最大的那部分數據，使得Redis的內存佔用降到一個合理的值。

LRU算法的缺點是，咱們須要維護一個所有（或只有過時時間）key的列表，還要按照最近使用時間排序。這會消耗大量內存，而且每次使用key時更新排序也會佔用額外的CPU資源。對於Redis這樣對性能要求很高的系統來講是不被容許的。

所以，Redis採用了一種近似LRU的算法。當Redis接收到新的寫入命令，而內存又不夠時，就會觸發近似LRU算法來強制清理一些key。具體清理的步驟是，Redis會對key進行採樣，一般是取5個，而後會把過時的key放到咱們上面說的「過時池」中，過時池中的key是按照空閒時間來排序的，Redis會優先清理掉空閒時間最長的key，直到內存小於maxmemory。

近似LRU算法的清理效果圖如圖（圖片來自Redis官方文檔）

這麼說可能不夠清楚，咱們直接上代碼。

源碼分析

上圖展現了代碼中近似LRU算法的主要邏輯調用路徑。

其中主要邏輯是在freeMemoryIfNeeded函數中

首先調用getMaxmemoryState函數判斷當前內存的狀態

int getMaxmemoryState(size_t *total, size_t *logical, size_t *tofree, float *level) {
    size_t mem_reported, mem_used, mem_tofree;

    mem_reported = zmalloc_used_memory();
    if (total) *total = mem_reported;

    int return_ok_asap = !server.maxmemory || mem_reported <= server.maxmemory;
    if (return_ok_asap && !level) return C_OK;

    mem_used = mem_reported;
    size_t overhead = freeMemoryGetNotCountedMemory();
    mem_used = (mem_used > overhead) ? mem_used-overhead : 0;

    if (level) {
        if (!server.maxmemory) {
            *level = 0;
        } else {
            *level = (float)mem_used / (float)server.maxmemory;
        }
    }

    if (return_ok_asap) return C_OK;

    if (mem_used <= server.maxmemory) return C_OK;

    mem_tofree = mem_used - server.maxmemory;

    if (logical) *logical = mem_used;
    if (tofree) *tofree = mem_tofree;

    return C_ERR;
}

若是使用內存低於maxmemory的話，就返回C_OK，不然返回C_ERR。另外，這個函數還經過傳遞指針型的參數來返回一些額外的信息。

• total：已使用的字節總數，不管是C_OK仍是C_ERR都有效。
• logical：已使用的內存減去slave或AOF緩衝區後的大小，只有返回C_ERR時有效。
• tofree：須要釋放的內存大小，只有返回C_ERR時有效。
• level：已使用內存的比例，一般是0到1之間，當超出內存限制時，就大於1。不管是C_OK仍是C_ERR都有效。

判斷完內存狀態之後，若是內存沒有超過使用限制就會直接返回，不然就繼續向下執行。此時咱們已經知道須要釋放多少內存空間了，下面就開始進行釋放內存的操做了。每次釋放內存都會記錄釋放內存的大小，直到釋放的內存不小於tofree。

首先根據maxmemory_policy進行判斷，對於不一樣的清除策略有不一樣的實現方法，咱們來看LRU的具體實現。

for (i = 0; i < server.dbnum; i++) {
  db = server.db+i;
  dict = (server.maxmemory_policy & MAXMEMORY_FLAG_ALLKEYS) ?
    db->dict : db->expires;
  if ((keys = dictSize(dict)) != 0) {
    evictionPoolPopulate(i, dict, db->dict, pool);
    total_keys += keys;
  }
}

首先是填充「過時池」，這裏遍歷了每個db（驗證了我最開始的想法），調用evictionPoolPopulate函數進行填充。

void evictionPoolPopulate(int dbid, dict *sampledict, dict *keydict, struct evictionPoolEntry *pool) {
    int j, k, count;
    dictEntry *samples[server.maxmemory_samples];

    count = dictGetSomeKeys(sampledict,samples,server.maxmemory_samples);
    for (j = 0; j < count; j++) {
        unsigned long long idle;
        sds key;
        robj *o;
        dictEntry *de;

        de = samples[j];
        key = dictGetKey(de);
				/* some code */
        if (server.maxmemory_policy & MAXMEMORY_FLAG_LRU) {
            idle = estimateObjectIdleTime(o);
        }

        /* some code */
        k = 0;
        while (k < EVPOOL_SIZE &&
               pool[k].key &&
               pool[k].idle < idle) k++;
        if (k == 0 && pool[EVPOOL_SIZE-1].key != NULL) {
            continue;
        } else if (k < EVPOOL_SIZE && pool[k].key == NULL) {
        } else {
            if (pool[EVPOOL_SIZE-1].key == NULL) {
                sds cached = pool[EVPOOL_SIZE-1].cached;
                memmove(pool+k+1,pool+k,
                    sizeof(pool[0])*(EVPOOL_SIZE-k-1));
                pool[k].cached = cached;
            } else {
                k--;
                sds cached = pool[0].cached; /* Save SDS before overwriting. */
                if (pool[0].key != pool[0].cached) sdsfree(pool[0].key);
                memmove(pool,pool+1,sizeof(pool[0])*k);
                pool[k].cached = cached;
            }
        }
        /* some code */
    }
}

因爲篇幅緣由，我截取了部分代碼，經過這段代碼咱們能夠看到，Redis首先是採樣了一部分key，這裏採樣數量maxmemory_samples一般是5，咱們也能夠本身設置，採樣數量越大，結果就越接近LRU算法的結果，帶來的影響是性能隨之變差。

採樣以後咱們須要得到每一個key的空閒時間，而後將其填充到「過時池」中的指定位置。這裏「過時池」是按照空閒時間從小到大排序的，也就是說，idle大大key排在最右邊。

填充完「過時池」以後，會從後向前獲取到最適合清理的key。

/* Go backward from best to worst element to evict. */
for (k = EVPOOL_SIZE-1; k >= 0; k--) {
  if (pool[k].key == NULL) continue;
  bestdbid = pool[k].dbid;

  if (server.maxmemory_policy & MAXMEMORY_FLAG_ALLKEYS) {
    de = dictFind(server.db[pool[k].dbid].dict,
                  pool[k].key);
  } else {
    de = dictFind(server.db[pool[k].dbid].expires,
                  pool[k].key);
  }
  /* some code */
  if (de) {
    bestkey = dictGetKey(de);
    break;
  }
}

找到須要刪除的key後，就須要根據設置清理策略進行同步/異步清理。

if (server.lazyfree_lazy_eviction)
  dbAsyncDelete(db,keyobj);
else
  dbSyncDelete(db,keyobj)

最後記下本次清理的空間大小，用來在循環條件判斷是否要繼續清理。

delta -= (long long) zmalloc_used_memory();
mem_freed += delta;

清理策略

最後咱們來看一下Redis支持的幾種清理策略

• noeviction：不會繼續處理寫請求（DEL能夠繼續處理）。
• allkeys-lru：對全部key的近似LRU
• volatile-lru：使用近似LRU算法淘汰設置了過時時間的key
• allkeys-random：從全部key中隨機淘汰一些key
• volatile-random：對全部設置了過時時間的key隨機淘汰
• volatile-ttl：淘汰有效期最短的一部分key

Redis4.0開始支持了LFU策略，和LRU相似，它分爲兩種：

• volatile-lfu：使用LFU算法淘汰設置了過時時間的key
• allkeys-lfu：從所有key中進行淘汰，使用LFU

寫在最後

如今我知道了Redis在內存達到上限時作了哪些事了。之後出問題時也就不會只檢查本身的db了。

關於此次事故的後續處理，我首先是讓業務同窗回滾了代碼，而後讓他們使用一個單獨的Redis，這樣業務再出現相似問題就不會影響到咱們的賬號服務了，總體的影響範圍也會變得更加可控。