“叮……”，美好的周六就這么被一陣釘釘消息吵醒了。

業務組的同學告訴我說很多用戶的帳號今天被強制下線。我們的帳號系統正常的邏輯是用戶登錄一次后，token的有效期可以維持一天的時間。現在的問題是用戶大概每10分鐘左右就需要重新登錄一次。這種情況一般有兩種原因：

• 1、token生成時出問題。
• 2、驗證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，這樣業務再出現類似問題就不會影響到我們的帳號服務了，整體的影響范圍也會變得更加可控。

原文鏈接：https://www.cnblogs.com/Jackeyzhe/p/12616624.html