詳解Redis 緩存刪除機制(源碼解析)

更新時間：2021年03月07日 10:22:13 作者：klew

這篇文章主要介紹了Redis 緩存刪除機制(源碼解析)，文中通過示例代碼介紹的非常詳細(xì)，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧

刪除的范圍

過期的 key
在內(nèi)存滿了的情況下，如果繼續(xù)執(zhí)行 set 等命令，且所有 key 都沒有過期，那么會按照緩存淘汰策略選中的 key

過期刪除

redis 中設(shè)置了過期時間的 key 會單獨存儲一份

typedef struct redisDb {
 dict *dict;   // 所有的鍵值對
 dict *expires;  //設(shè)置了過期時間的鍵值對
 // ...
} redisDb;

設(shè)置有效期

Redis 中有 4 個命令可以給 key 設(shè)置過期時間，分別是 expire pexpire expireat pexpireat

設(shè)置相對時間

expire <key> <ttl>：將 key 值的過期時間設(shè)置為 ttl 秒。

// src/expire.c

/* EXPIRE key seconds */
void expireCommand(client *c) {
 expireGenericCommand(c,mstime(),UNIT_SECONDS);
}

pexpire <key> <ttl>：將 key 值的過期時間設(shè)置為 ttl 毫秒。

// src/expire.c

/* PEXPIRE key milliseconds */
void pexpireCommand(client *c) {
 expireGenericCommand(c,mstime(),UNIT_MILLISECONDS);
}

設(shè)置絕對時間

expireat <key> <timestamp>：將 key 值的過期時間設(shè)置為指定的 timestamp 秒數(shù)。

// src/expire.c

/* EXPIREAT key time */
void expireatCommand(client *c) {
 expireGenericCommand(c,0,UNIT_SECONDS);
}

pexpireat <key> <timestamp>：將 key 值的過期時間設(shè)置為指定的 timestamp 毫秒數(shù)。

// src/expire.c

/* PEXPIREAT key ms_time */
void pexpireatCommand(client *c) {
 expireGenericCommand(c,0,UNIT_MILLISECONDS);
}

以上 4 種方法最終都會調(diào)用下面的通用函數(shù) expireGenericCommand ：

// src/expire.c

void expireGenericCommand(client *c, long long basetime, int unit) {
 robj *key = c->argv[1], *param = c->argv[2];
 
 // 獲取數(shù)據(jù)對象
 long long when;
 if (getLongLongFromObjectOrReply(c, param, &when, NULL) != C_OK)
  return;

 // 將時間轉(zhuǎn)化成以 ms 為單位
 if (unit == UNIT_SECONDS) when *= 1000;
 when += basetime;
 // 在 master 節(jié)點上，如果設(shè)置的過期時間小于當(dāng)前時間，那么將命令轉(zhuǎn)化成 DEL 指令
 if (when <= mstime() && !server.loading && !server.masterhost) {
  robj *aux;

  int deleted = server.lazyfree_lazy_expire ? dbAsyncDelete(c->db,key) :
             dbSyncDelete(c->db,key);
  // ...
  // 將刪除命令同步給 slave 和 AOF
  // ...
 } else {
  // 設(shè)置過期時間
  setExpire(c,c->db,key,when);
  // ...
  // 構(gòu)造返回值和發(fā)布對象更新消息
  // ...
  return;
 }
}

設(shè)置過期時間的操作由 setExpire 執(zhí)行，他將 dictEntry 的 union v 中的 s64 設(shè)為過期時間

// src/db.c

void setExpire(client *c, redisDb *db, robj *key, long long when) {
 dictEntry *kde, *de;

 // 找出 db->dict 中對應(yīng)的存儲對象，這里的查詢和用 get 查詢數(shù)據(jù)是邏輯一樣，通過 hashFunc(key) & sizemask 
 // 找到 bucket 后在鏈表中遍歷
 kde = dictFind(db->dict,key->ptr);
 // 找出 db->expires 中對應(yīng)的存儲對象，如果沒有則新建一個
 de = dictAddOrFind(db->expires,dictGetKey(kde));
 // 
 dictSetSignedIntegerVal(de,when);
 // ...
}

#define dictSetSignedIntegerVal(entry, _val_) \
 do { (entry)->v.s64 = _val_; } while(0)

db->expires 中存儲的 dictEntry 表示的是過期 key 和過期時間，存儲過期時間的 v 是一個 union ，可見在 redis 中不同使用場景或不同編碼下 v 的意義不同

typedef struct dictEntry {
 void *key;
 union {
  void *val;
  uint64_t u64;
  int64_t s64;
  double d;
 } v;
 struct dictEntry *next;
} dictEntry;

查詢過期時間

ttl key 返回 key 剩余過期秒數(shù)。

// src/expire.c

/* TTL key */
void ttlCommand(client *c) {
 ttlGenericCommand(c, 0);
}

pttl key 返回 key 剩余過期的毫秒數(shù)。

// src/expire.c

/* PTTL key */
void pttlCommand(client *c) {
 ttlGenericCommand(c, 1);
}

以上 2 種查看方式最終都會調(diào)用下面的通用函數(shù) ttlGenericCommand ：

// src/expire.c

/* Implements TTL and PTTL */
void ttlGenericCommand(client *c, int output_ms) {
 // ...
 // key 不存在時報錯
 // ...
 
 // 獲取過期時間，如果沒有過期時間則
 expire = getExpire(c->db,c->argv[1]);
 if (expire != -1) {
  ttl = expire-mstime();
  if (ttl < 0) ttl = 0;
 }
 
 if (ttl == -1) {
  addReplyLongLong(c,-1);
 } else {
  // 根據(jù)指定的單位返回結(jié)果，以秒為單位時向上取整
  addReplyLongLong(c,output_ms ? ttl : ((ttl+500)/1000));
 }
}

獲取過期時間的操作由 getExpire 執(zhí)行，在 db->expires 中查詢到對象后，獲取 union v 中的成員 s64

// src/expire.c

// 返回過期時間的絕對時間
long long getExpire(redisDb *db, robj *key) {
 dictEntry *de;

 // 查詢對象
 if (dictSize(db->expires) == 0 ||
  // 如果返回為 NULL 表示沒有設(shè)置過期時間，向上返回 -1
  (de = dictFind(db->expires,key->ptr)) == NULL) return -1;
 
 // 獲取 v.s64
 return dictGetSignedIntegerVal(de);
}

#define dictGetSignedIntegerVal(he) ((he)->v.s64)

過期策略

Redis 綜合使用惰性刪除和定期掃描實現(xiàn)

惰性刪除

每次訪問時會調(diào)用 expireIfNeeded 判斷 key 是否過期，如果過期就刪除該鍵，否則返回鍵對應(yīng)的值。單獨使用這種策略可能會浪費很多內(nèi)存。

// src/db.c

int expireIfNeeded(redisDb *db, robj *key) {
 mstime_t when = getExpire(db,key);
 mstime_t now;
 
 // 沒有設(shè)置過期時間，直接返回
 if (when < 0) return 0;

 // 從硬盤中加載數(shù)據(jù)時不執(zhí)行過期操作
 if (server.loading) return 0;

 // 參考 GitHub Issue #1525
 // 對于 master，在執(zhí)行 Lua Script 的過程中，可能會用某個 key 是否存在當(dāng)作判斷條件
 // 為了避免一個腳本中前后條件不一致，將當(dāng)前時間強制設(shè)為腳本開始時間 
 now = server.lua_caller ? server.lua_time_start : mstime();

 // 對于 slave，返回此時 key 是否已過期，但不執(zhí)行后續(xù)刪除操作
 if (server.masterhost != NULL) return now > when;

 // key 未過期
 if (now <= when) return 0;

 // 統(tǒng)計過期 key 的個數(shù)
 server.stat_expiredkeys++;
 // 向所有的 slave 和 AOF 文件寫入一條 DEL 指令
 propagateExpire(db,key,server.lazyfree_lazy_expire);
 // 向 keyspace channel 中發(fā)布一條 key 過期的消息
 notifyKeyspaceEvent(NOTIFY_EXPIRED,
  "expired",key,db->id);
 // 根據(jù)配置決定是同步刪除還是異步刪除（僅刪除引用，由后臺線程執(zhí)行物理刪除）
 return server.lazyfree_lazy_expire ? dbAsyncDelete(db,key) :
           dbSyncDelete(db,key);
}

特殊處理

在 master 節(jié)點執(zhí)行 Lua 腳本時

參考 GitHub Issue #1525，對于 master，在執(zhí)行 Lua Script 的過程中，可能會用某個 key 是否存在當(dāng)作判斷條件。為了避免一個腳本中前后條件不一致，將當(dāng)前時間強制設(shè)為腳本開始時間。
例如多次執(zhí)行如下 Lua 腳本 /tmp/myscript.lua 出現(xiàn)的結(jié)果可能不一致

-- /tmp/myscript.lua

if redis.call("exists",KEYS[1]) == 1
then
 redis.call("incr","mycounter")
end

if redis.call("exists",KEYS[1]) == 1
then
 return redis.call("incr","mycounter")
end

具體復(fù)現(xiàn)操作可以參考下面的 bash 腳本：

while [ 1 ]
do
 redis-cli set x foo px 100 > /dev/null
 sleep 0.092
 redis-cli --eval /tmp/myscript.lua x > /dev/null
 sleep 0.1
 redis-cli get mycounter
 redis-cli -p 6380 get mycounter
done

對于 slave 節(jié)點

在 slave 節(jié)點上，key 的刪除操作由 master 發(fā)來的 DEL 執(zhí)行，因此這里只返回是否過期的結(jié)果給客戶端，而不執(zhí)行刪除操作

正在從 RDB 和 AOF 讀取數(shù)據(jù)時跳過這個步驟

定期掃描

系統(tǒng)每隔一段時間就定期掃描一次，發(fā)現(xiàn)過期的鍵就進行刪除。單獨使用這種策略可能出現(xiàn)鍵已經(jīng)過期但沒有刪除的情況
Redis 默認(rèn)每 100ms 執(zhí)行一次（通過 hz 參數(shù)配置，執(zhí)行周期為 1s/hz）過期掃描。由于 redisDb 中設(shè)置了過期時間的 key 會單獨存儲，所以不會出現(xiàn)掃描所有 key 的情況
具體步驟由 activeExpireCycle 函數(shù)執(zhí)行

activeExpireCycle、incrementallyRehash 等后臺操作都是由 databasesCron 觸發(fā)的

void activeExpireCycle(int type) {
 // ...
 
 // 依次遍歷各個 db
 for (j = 0; j < dbs_per_call && timelimit_exit == 0; j++) {
  int expired;
  redisDb *db = server.db+(current_db % server.dbnum);

  // 記錄下一個執(zhí)行的 db，這樣如果因為超時意外退出，下次可以繼續(xù)從這個 db 開始，
  // 從而在所有 db 上均勻執(zhí)行清除操作
  current_db++;

  do {
   // ...
   // 跳過沒有設(shè)置過期時間的 key 等不需要執(zhí)行的情況
   // ...

   // 抽樣個數(shù)，默認(rèn)為 20
   if (num > ACTIVE_EXPIRE_CYCLE_LOOKUPS_PER_LOOP)
    num = ACTIVE_EXPIRE_CYCLE_LOOKUPS_PER_LOOP;

   // 從設(shè)置了過期時間的 key 中隨機抽取 20 個
   while (num--) {
    dictEntry *de;
    long long ttl;

    // 隨機挑選 dict 中的一個 key
    if ((de = dictGetRandomKey(db->expires)) == NULL) break;
    ttl = dictGetSignedIntegerVal(de)-now;
    // 執(zhí)行刪除，具體刪除操作和惰性刪除中類似
    if (activeExpireCycleTryExpire(db,de,now)) expired++;
    // ...
   }
   // ...
   // 更新統(tǒng)計數(shù)據(jù)等操作
   // ...
  // 如果每次刪除的 key 超過了樣本數(shù)的 25%，說明過期鍵占的比例較高，需要再重復(fù)執(zhí)行依次
  } while (expired > ACTIVE_EXPIRE_CYCLE_LOOKUPS_PER_LOOP/4);
 }
 // ...
}

隨機抽樣由 dictGetRandomKey 執(zhí)行

// src/dict.c

/* Return a random entry from the hash table. Useful to
 * implement randomized algorithms */
dictEntry *dictGetRandomKey(dict *d)
{
 dictEntry *he, *orighe;
 unsigned long h;
 int listlen, listele;

 // 沒有數(shù)據(jù)，返回為 NULL，外層函數(shù)接收到 NULL 后會中斷過期操作的執(zhí)行
 if (dictSize(d) == 0) return NULL;
 // 根據(jù) rehashidx 參數(shù)判斷是否正在執(zhí)行 rehash，如果正在執(zhí)行，
 // 則先執(zhí)行 rehash 中的一個步驟
 if (dictIsRehashing(d)) _dictRehashStep(d);
 
 if (dictIsRehashing(d)) {
  do {
   // 正在執(zhí)行 rehash，所以兩個 ht 中的對象都要考慮
   //
   // 由于正在執(zhí)行 rehash，所以可以肯定 ht[0] 中下標(biāo)小于等于 rehashidx 的 bucket
   // 肯定沒有數(shù)據(jù)，所以只從 ht[0] 中大于 rehashidx 的 bucket 和 ht[1] 中抽取
   h = d->rehashidx + (random() % (d->ht[0].size +
           d->ht[1].size -
           d->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);
 } else {
  do {
   // 參考寫入 ht 時計算下標(biāo)的規(guī)則 hashFunc(key) & sizemake
   // 這里 random() & sizemask 是隨機取一個下標(biāo)
   h = random() & d->ht[0].sizemask;
   he = d->ht[0].table[h];
  // 取到空 bucket 時重試
  } while(he == NULL);
 }
 
 // 到這一步 he 是 ht[n] 中某個 bucket 中完整的鏈表
 // 所以還要從這個鏈表中隨機取一個對象
 
 // 遍歷計算整個鏈表的長度
 listlen = 0;
 orighe = he;
 while(he) {
  he = he->next;
  listlen++;
 }
 // 隨機取鏈表中某個對象的下標(biāo)
 listele = random() % listlen;
 he = orighe;
 // 重新遍歷鏈表獲取指定下標(biāo)的對象
 while(listele--) he = he->next;
 return he;
}

緩存淘汰

配置最大內(nèi)存限制

在 redis.conf 中配置

redis server 啟動時加載配置文件和命令行參數(shù)中的 maxmemory ，存入 Server 對象的 maxmemory 字段
main 中在 redis server 啟動時執(zhí)行初始化等操作，其中會執(zhí)行加載配置文件的 loadServerConfig 函數(shù)

// src/server.c
int main(int argc, char **argv) {
 // ..
 // 加載配置
 loadServerConfig(configfile,options);
 // ..
 // 警告過小的配置
 if (server.maxmemory > 0 && server.maxmemory < 1024*1024) {
  serverLog(LL_WARNING,"WARNING: You specified a maxmemory value that is less than 1MB (current value is %llu bytes). Are you sure this is what you really want?", server.maxmemory);
 }
}

loadServerConfig 中將配置文件、stdin、命令行參數(shù)加載到 config 字符串中，然后調(diào)用 loadServerConfigFromString

// src/config.c
void loadServerConfig(char *filename, char *options) {
 sds config = sdsempty();
 char buf[CONFIG_MAX_LINE+1];
 
 // 加載配置文件
 if (filename) {
  FILE *fp;

  // 啟動命令為 ./redis-server - 則從 stdin 中讀取，需要用 <C-D> 觸發(fā) EOF
  if (filename[0] == '-' && filename[1] == '\0') {
   fp = stdin;
  } else {
  // 第一個參數(shù)不是 -，則嘗試打開這個參數(shù)指定的文件
   if ((fp = fopen(filename,"r")) == NULL) {
    serverLog(LL_WARNING,
     "Fatal error, can't open config file '%s'", filename);
    exit(1);
   }
  }
  // 將配置文件中的每一行追加到 config 中
  while(fgets(buf,CONFIG_MAX_LINE+1,fp) != NULL)
   config = sdscat(config,buf);
  if (fp != stdin) fclose(fp);
 }
 // 添加其他選項，例如 ./redis-server --port 8080 后面的參數(shù)，直接加到 config 中
 if (options) {
  config = sdscat(config,"\n");
  config = sdscat(config,options);
 }
 loadServerConfigFromString(config);
 sdsfree(config);
}

loadServerConfigFromString 從上一步中的 config 字符串中逐行讀取配置，并寫入 server 對象

// src/config.c
void loadServerConfigFromString(char *config) {
 // ...
 
 // 按行讀取配置文件
 lines = sdssplitlen(config,strlen(config),"\n",1,&totlines);
 for (i = 0; i < totlines; i++) {
 // 跳過無效的配置和注釋
  // ...
  argv = sdssplitargs(lines[i],&argc);
  
  // 將配置命令轉(zhuǎn)化成小寫
  sdstolower(argv[0]);

  // 根據(jù)配置命令初始化配置，strcasecmp 比較
  if (!strcasecmp(argv[0],"timeout") && argc == 2) {
   server.maxidletime = atoi(argv[1]);
   if (server.maxidletime < 0) {
    err = "Invalid timeout value"; goto loaderr;
   }
  // ...
  } else if (!strcasecmp(argv[0],"maxmemory") && argc == 2) {
   // memtoll 將字符串形式的配置轉(zhuǎn)化成對應(yīng)的 long long 值
   // 例如 1kb -> 1024
   server.maxmemory = memtoll(argv[1],NULL);
  }
 }
}

使用 CONFIG SET 命令配置

Redis Server 接收到客戶端的 CONFIG SET 命令后調(diào)用 configSetCommand 函數(shù)
服務(wù)端接收到命令時將命令和參數(shù)存入 Redis Server 的 argc 和 argv

argc: 4
argv: 0  1 2   3
  config set maxmemory 10mb

動態(tài)配置 maxmemory 后會立即嘗試觸發(fā)內(nèi)存回收，而修改其他內(nèi)存相關(guān)配置(例如: maxmemory_policy)時不會觸發(fā)

if (0) {
 // ...
} config_set_memory_field("maxmemory",server.maxmemory) {
 // 配置不為 0，表示之前限制過內(nèi)存
 if (server.maxmemory) {
  if (server.maxmemory < zmalloc_used_memory()) {
   serverLog(LL_WARNING,"WARNING: the new maxmemory value set via CONFIG SET is smaller than the current memory usage. This will result in keys eviction and/or inability to accept new write commands depending on the maxmemory-policy.");
  }
  freeMemoryIfNeeded();
 }
 // ...
}

32 位機器的內(nèi)存限制

對于 64 位機器，將 maxmemory 設(shè)置為 0 表示不限制內(nèi)存，但由于 32 位尋址空間最多只有 4 GB，所以默認(rèn)內(nèi)存限制設(shè)為 3 GB，緩存淘汰策略設(shè)為 noeviction

// src/server.c
// ...
if (server.arch_bits == 32 && server.maxmemory == 0) {
 serverLog(LL_WARNING,"Warning: 32 bit instance detected but no memory limit set. Setting 3 GB maxmemory limit with 'noeviction' policy now.");
 server.maxmemory = 3072LL*(1024*1024); /* 3 GB */
 server.maxmemory_policy = MAXMEMORY_NO_EVICTION;
 }

淘汰策略

淘汰策略使用 CONFIG SET maxmemory-policy 配置

默認(rèn)：

**noeviction: **內(nèi)存滿了后對于 set 等命令直接返回錯誤

針對所有 key：

allkeys-lru: 在所有 key 的范圍內(nèi)使用 LRU 算法執(zhí)行刪除，如果內(nèi)存仍然不夠，則報錯
**allkeys-lfu: **在所有 key 的范圍內(nèi)使用 LRU 算法執(zhí)行刪除，如果內(nèi)存仍然不夠，則報錯
**allkeys-random: **在所有 key 的范圍內(nèi)隨機執(zhí)行刪除，如果內(nèi)存仍然不夠，則報錯

針對設(shè)置了過期時間的 key：

**volatile-lru: **在設(shè)置了過期時間的 key 中使用 LRU 算法執(zhí)行刪除，如果內(nèi)存仍然不夠，則報錯
**volatile-lfu: **在設(shè)置了過期時間的 key 中使用 LRU 算法執(zhí)行刪除，如果內(nèi)存仍然不夠，則報錯
**volatile-random: **在設(shè)置了過期時間的 key 中隨機執(zhí)行刪除，如果內(nèi)存仍然不夠，則報錯
**volatile-ttl: **刪除即將過期的 key，如果內(nèi)存仍然不夠，則報錯

Redis 在執(zhí)行淘汰之前會計算部分對象的 idle 值，使用不同淘汰策略時計算 idle 值的方法也不同， idle 值越大表示這個值越需要優(yōu)先刪除。

下面主要介紹 LRU 和 LFU 中 idle 值的計算方法

LRU 淘汰策略

抽樣刪除，樣本數(shù)量通過 CONFIG SET maxmemory-samples 100 控制，對應(yīng) RedisObject 中的 maxmemory_samples 參數(shù)，抽樣數(shù)量越多和傳統(tǒng)的 LRU 算法越接近

優(yōu)化策略

為了避免傳統(tǒng)的 LRU 算法通常使用 hashmap + 鏈表實現(xiàn)帶來的開銷，Redis 進行了如下優(yōu)化：

RedisObject 結(jié)構(gòu)中設(shè)置了一個 lru 字段，用來記錄數(shù)據(jù)的訪問時間戳，而不是每次調(diào)整對象在鏈表中的位置

typedef struct redisObject {
 // 對象類型
 unsigned type:4;
 // 對象編碼
 unsigned encoding:4;
 // LRU 算法和 LFU 算法公用 lru 這個字段
 // 
 // LRU_BITS 默認(rèn)為 24，因此最大只能存儲 194 天的時間戳,
 // 創(chuàng)建對象時會寫入這個字段，訪問對象時會更新這個字段，
 // 超過之后再從 0 開始計算
 unsigned lru:LRU_BITS;
 int refcount;
 void *ptr;
} robj;

使用抽樣數(shù)組代替鏈表，后續(xù)在候選集合中根據(jù) lru 字段值的大小進行篩選，避免了鏈表帶來的開銷。候選集合中的對象用 evictionPoolEntry 表示

struct evictionPoolEntry {
 unsigned long long idle; // 用于淘汰排序，在不同算法中意義不同
 sds key; // 鍵的名字
 // ...
};

計算方法

全局對象 lru_clock 記錄了當(dāng)前的 unix 時間戳，由 serverCron 調(diào)用 updateCachedTime 默認(rèn)每 100 ms 更新一次。更新頻率與 hz 參數(shù)有關(guān)， 1s/hz 即為更新間隔時間。
LRU_CLOCK_RESOLUTION 的值為 1000，因此使用 LRU_CLOCK 函數(shù)獲取 lru_clock 時，如果每秒更新頻率在 1 次以上，會使用全局變量中緩存的 lrulcock

unsigned int LRU_CLOCK(void) {
 unsigned int lruclock;
 if (1000/server.hz <= LRU_CLOCK_RESOLUTION) {
  atomicGet(server.lruclock,lruclock);
 } else {
  lruclock = getLRUClock();
 }
 return lruclock;
}

如果更新頻率不到每秒 1 次，則會用函數(shù) getLRUClock 實時計算 lruclock

unsigned int getLRUClock(void) {
 // mstime() 獲取 unix 時間戳，單位時毫秒
 // 除以 LRU_CLOCK_RESOLUTION(值為 1000)，將時間戳轉(zhuǎn)化為秒
 return (mstime()/LRU_CLOCK_RESOLUTION) & LRU_CLOCK_MAX;
}

其中 LRU_CLOCK_MAX 表示 lru_clock 最大的可能值，這個值與 redisObject 中 lru 最大的可能值一樣，定義如下：

#define LRU_CLOCK_MAX ((1<<LRU_BITS)-1)

所以最終比較時 lru_clock 和 robj.lru 的值都在 [0, LRU_CLOCK_MAX] 的范圍內(nèi)。
從邏輯上講，當(dāng)前時間戳應(yīng)該永遠(yuǎn)大于上次訪問的時間戳 ,因此正常的計算規(guī)則應(yīng)該是 lru_clock-robj.lru 。
但是由于 lru_clock 和 robj.lru 是當(dāng)前時間戳取模后的值，所以可能出現(xiàn) lru_clock 小于 robj.lru 的情況，所以這種情況下計算規(guī)則應(yīng)該改為 lru_clock+194天-robj.lru
但是對于 lru_clock 和 robj.lru 間隔超過 194 天的情況仍然無法判斷，所以更能存在刪除不準(zhǔn)確的情況。
將上述的邏輯組合起來就是 LRU 算法下獲取 idle 值的函數(shù):

// src/evict.c

// 以秒為精度計算對象距離上一次訪問的間隔時間，然后轉(zhuǎn)化成毫秒返回
unsigned long long estimateObjectIdleTime(robj *o) {
 unsigned long long lruclock = LRU_CLOCK();
 if (lruclock >= o->lru) {
  return (lruclock - o->lru) * LRU_CLOCK_RESOLUTION;
 } else {
  return (lruclock + (LRU_CLOCK_MAX - o->lru)) *
     LRU_CLOCK_RESOLUTION;
 }
}

在 Redis 3.0 中，當(dāng)取樣數(shù)量設(shè)為 10 時，已經(jīng)和傳統(tǒng)的 LRU 算法效果很接近了

LFU 淘汰策略

LFU 算法復(fù)用 robj.lru 字段，將這個 24 bit 的字段拆分成了兩部分：

ldt(last decrement time，單位：分鐘)：lru 字段的前 16bit，表示數(shù)據(jù)的訪問時間戳，最多只能存儲 45 天。
counter 值：lru 字段的后 8bit，表示數(shù)據(jù)的訪問頻率

遞增策略

counter 能表示的最大值是 255，因此 counter 與訪問次數(shù)不能是線性關(guān)系，這里采用的計算步驟如下：

隨機取 0 到 1 之間的隨機數(shù) r
比較 r 與 1/((counter-LFU_INIT_VAL)*lfu_log_factor+1) 的大小，其中 LFU_INIT_VAL 是常量，默認(rèn)為 5，lfu_log_factor 是可配置參數(shù)，默認(rèn)為 10
如果 r 小則 counter 增加 1，否則 counter 不變

實現(xiàn)代碼如下：

uint8_t LFULogIncr(uint8_t counter) {
 // counter 值已經(jīng)到達了 255，不能再增加，直接返回
 if (counter == 255) return 255;
 double r = (double)rand()/RAND_MAX;
 double baseval = counter - LFU_INIT_VAL; // LFU_INIT_VAL 值為 5
 if (baseval < 0) baseval = 0;
 double p = 1.0/(baseval*server.lfu_log_factor+1);
 if (r < p) counter++;
 return counter;
}

訪問次數(shù)與 counter 值之間大概是對數(shù)關(guān)系，counter 值越大，增速越低

// https://redis.io/topics/lru-cache

+--------+------------+------------+------------+------------+------------+
| factor | 100 hits | 1000 hits | 100K hits | 1M hits | 10M hits |
+--------+------------+------------+------------+------------+------------+
| 0  | 104  | 255  | 255  | 255  | 255  |
+--------+------------+------------+------------+------------+------------+
| 1  | 18   | 49   | 255  | 255  | 255  |
+--------+------------+------------+------------+------------+------------+
| 10  | 10   | 18   | 142  | 255  | 255  |
+--------+------------+------------+------------+------------+------------+
| 100 | 8   | 11   | 49   | 143  | 255  |
+--------+------------+------------+------------+------------+------------+

衰減策略

除了訪問對象時 counter 需要增加，對于一段時間內(nèi)沒有訪問的對象還要相應(yīng)地減少 counter 值，遞減的速率由 lfu-decay-time 參數(shù)控制。
counter 衰減步驟如下：

取當(dāng)前時間戳（單位：分鐘）的低 16 位記為 now ，計算與 ldt 的差值。這里與 LRU 算法中計算 lru_clock 和 robj.lru 時可能出現(xiàn)一樣的問題，由于 ldt 最多只能表示 45 天，所以如果距離對象上次訪問超過 45 天，則無法準(zhǔn)確計算訪問的時間間隔

unsigned long LFUDecrAndReturn(robj *o) {
 // 取高 16 位
 unsigned long ldt = o->lru >> 8;
 // 取低 8 位
 unsigned long counter = o->lru & 255;
 // 如果 lfu_decay_time 為 0，則步修改 counter，否則將 counter 減少 LFUTimeElapsed(ldt)/lfu_decay_time
 unsigned long num_periods = server.lfu_decay_time ? LFUTimeElapsed(ldt) / server.lfu_decay_time : 0;
 if (num_periods)
  // 保證 counter 的最小值位 0
  counter = (num_periods > counter) ? 0 : counter - num_periods;
 return counter;
}

// 計算距離上次訪問的間隔時間
unsigned long LFUTimeElapsed(unsigned long ldt) {
 // 取當(dāng)前時間戳（單位：分鐘）
 unsigned long now = LFUGetTimeInMinutes();
 // 計算時間差
 if (now >= ldt) return now-ldt;
 return 65535-ldt+now;
}

// 獲取當(dāng)前時間戳，以分鐘為單位，取低 8 位
unsigned long LFUGetTimeInMinutes(void) {
 return (server.unixtime/60) & 65535;
}

如果 lfu_decay_time 為 0，則步修改 counter，否則將 counter 減少 LFUTimeElapsed(ldt)/lfu_decay_time

例如，在 lfu_decay_time 為 1 的情況下，如果有 N 分鐘沒有訪問這個對象，那么 counter 值減 N
每次訪問一個對象時都會調(diào)用 updateLFU 更新 counter 的值：

void updateLFU(robj *val) {
 unsigned long counter = LFUDecrAndReturn(val);
 counter = LFULogIncr(counter);
 val->lru = (LFUGetTimeInMinutes()<<8) | counter;
}

執(zhí)行淘汰

當(dāng) Redis 需要淘汰一批數(shù)據(jù)時，會調(diào)用 evictionPoolPopulate 獲取一批待刪除對象，根據(jù)設(shè)置的淘汰范圍的不同，會決定傳遞給 evictionPoolPopulate 的 sampledict 參數(shù)是存有全部數(shù)據(jù)的 db->dict 還是只有設(shè)置了過期時間的數(shù)據(jù)的 db->expires

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

 // 隨機獲取 server.maxmemory_samples 個對象，寫入 samples 中
 count = dictGetSomeKeys(sampledict,samples,server.maxmemory_samples);
 // 遍歷每個對象
 for (j = 0; j < count; j++) {
  // ...
  // 初始化
  // ...

  de = samples[j];
  key = dictGetKey(de);

  // 如果獲取樣本的字典不是 db->dict（還可能是 db->expires），并且不是按 volatile-ttl 淘汰
  // 那么還要將對象轉(zhuǎn)化成數(shù)據(jù)字典中對應(yīng)的對象，然后取其值
  if (server.maxmemory_policy != MAXMEMORY_VOLATILE_TTL) {
   if (sampledict != keydict) de = dictFind(keydict, key);
   
   // #define dictGetVal(he) ((he)->v.val)
   // 這里還是利用 union 的特性，如果是 db->dict 中的元素，返回的是鍵的值
   // 如果是 db->expires 中的元素，返回的是過期時間
   o = dictGetVal(de);
  }

  // 按各算法計算 idle 分值，idle 越大的越應(yīng)該被先淘汰
  //
  // 如果使用 LRU 淘汰算法，則計算對象的空閑時間
  if (server.maxmemory_policy & MAXMEMORY_FLAG_LRU) {
   idle = estimateObjectIdleTime(o);
  // 使用 LFU 淘汰算法，
  } else if (server.maxmemory_policy & MAXMEMORY_FLAG_LFU) {
   idle = 255-LFUDecrAndReturn(o);
  // 使用 volatile-ttl 算法，用 ULLONG_MAX 減去過期時間作為分值
  } else if (server.maxmemory_policy == MAXMEMORY_VOLATILE_TTL) {
   idle = ULLONG_MAX - (long)dictGetVal(de);
  } else {
   serverPanic("Unknown eviction policy in evictionPoolPopulate()");
  }

  k = 0;
  // 與原 pool 中的 idle 值進行比較，找出應(yīng)該比當(dāng)前對象先淘汰出去的對象
  while (k < EVPOOL_SIZE &&
    pool[k].key &&
    pool[k].idle < idle) k++;
  if (k == 0 && pool[EVPOOL_SIZE-1].key != NULL) {
   // 沒有發(fā)現(xiàn)更需要被淘汰的對象，并且 pool 中也沒有多余的位置
   // 那么當(dāng)前對象仍然留在 samples 中
   continue;
  } else if (k < EVPOOL_SIZE && pool[k].key == NULL) {
   // 沒有發(fā)現(xiàn)更需要被淘汰的對象，但 pool 中有多余的位置
   // 于是將這個對象插入 pool 中
  } else {
   //     當(dāng)前對象
   //      |
   //      V
   // Pool: [ 0 1 2 3 ...k-1 k ... EVPOOL_SIZE-1]
   // 為了保證 pool 中的數(shù)據(jù)按 idle 從小到大排列，這里將當(dāng)前對象插入第 k 個對象后面的位置
   if (pool[EVPOOL_SIZE-1].key == NULL) {
    // pool 的右邊還有空余的位置，因此將從第 k 個開始后面的元素整體后移
    memmove(pool+k+1,pool+k,
     sizeof(pool[0])*(EVPOOL_SIZE-k-1));
   } else {
    // pool 的右邊沒有空余的位置了，那么將 pool 中前 k 個元素整體左移
    sds cached = pool[0].cached;
    memmove(pool,pool+1,sizeof(pool[0])*k);
   }
  }
  // ...
  // 將當(dāng)前對象的屬性賦值到下標(biāo)為 k 的元素
  // ...
 }
}

完成上述操作后，pool 中剩下的就是新篩選出來的最需要淘汰的對象了。

在 freeMemoryIfNeeded 中會調(diào)用 evictionPoolPopulate 來篩選需要淘汰的對象，每次刪除一個，直到釋放了足夠的內(nèi)存。如果始終無法達到內(nèi)存需求，則會報錯。

到此這篇關(guān)于Redis 緩存刪除機制(源碼解析)的文章就介紹到這了,更多相關(guān)Redis 緩存刪除內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: