详解Redis 缓存删除机制(源码解析)

目录

• 删除的范围
• 过期删除

• 设置有效期
• 查询过期时间
• 过期策略

• 配置最大内存限制
• 淘汰策略

typedef struct redisDb {
 dict *dict;   // 所有的键值对
 dict *expires;  //设置了过期时间的键值对
 // ...
} redisDb;

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

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

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

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

// src/expire.c
void expireGenericCommand(client *c, long long basetime, int unit) {
 robj *key = c->argv[1], *param = c->argv[2];
 // 获取数据对象
 long long when;
 if (getLongLongFromObjectOrReply(c, param, &when, NULL) != C_OK)
  return;
 // 将时间转化成以 ms 为单位
 if (unit == UNIT_SECONDS) when *= 1000;
 when += basetime;
 // 在 master 节点上，如果设置的过期时间小于当前时间，那么将命令转化成 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 {
  // 设置过期时间
  setExpire(c,c->db,key,when);
  // ...
  // 构造返回值和发布对象更新消息
  // ...
  return;
 }
}

// src/db.c
void setExpire(client *c, redisDb *db, robj *key, long long when) {
 dictEntry *kde, *de;
 // 找出 db->dict 中对应的存储对象，这里的查询和用 get 查询数据是逻辑一样，通过 hashFunc(key) & sizemask 
 // 找到 bucket 后在链表中遍历
 kde = dictFind(db->dict,key->ptr);
 // 找出 db->expires 中对应的存储对象，如果没有则新建一个
 de = dictAddOrFind(db->expires,dictGetKey(kde));
 // 
 dictSetSignedIntegerVal(de,when);
 // ...
}
#define dictSetSignedIntegerVal(entry, _val_) \
 do { (entry)->v.s64 = _val_; } while(0)

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

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

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

// 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 {
  // 根据指定的单位返回结果，以秒为单位时向上取整
  addReplyLongLong(c,output_ms ? ttl : ((ttl+500)/1000));
 }
}

// src/expire.c
// 返回过期时间的绝对时间
long long getExpire(redisDb *db, robj *key) {
 dictEntry *de;
 // 查询对象
 if (dictSize(db->expires) == 0 ||
  // 如果返回为 NULL 表示没有设置过期时间，向上返回 -1
  (de = dictFind(db->expires,key->ptr)) == NULL) return -1;
 // 获取 v.s64
 return dictGetSignedIntegerVal(de);
}
#define dictGetSignedIntegerVal(he) ((he)->v.s64)

// src/db.c
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;
 // 参考 GitHub Issue #1525
 // 对于 master，在执行 Lua Script 的过程中，可能会用某个 key 是否存在当作判断条件
 // 为了避免一个脚本中前后条件不一致，将当前时间强制设为脚本开始时间 
 now = server.lua_caller ? server.lua_time_start : mstime();
 // 对于 slave，返回此时 key 是否已过期，但不执行后续删除操作
 if (server.masterhost != NULL) return now > when;
 // key 未过期
 if (now <= when) return 0;
 // 统计过期 key 的个数
 server.stat_expiredkeys++;
 // 向所有的 slave 和 AOF 文件写入一条 DEL 指令
 propagateExpire(db,key,server.lazyfree_lazy_expire);
 // 向 keyspace channel 中发布一条 key 过期的消息
 notifyKeyspaceEvent(NOTIFY_EXPIRED,
  "expired",key,db->id);
 // 根据配置决定是同步删除还是异步删除（仅删除引用，由后台线程执行物理删除）
 return server.lazyfree_lazy_expire ? dbAsyncDelete(db,key) :
           dbSyncDelete(db,key);
}

-- /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

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

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);
  // 记录下一个执行的 db，这样如果因为超时意外退出，下次可以继续从这个 db 开始，
  // 从而在所有 db 上均匀执行清除操作
  current_db++;
  do {
   // ...
   // 跳过没有设置过期时间的 key 等不需要执行的情况
   // ...
   // 抽样个数，默认为 20
   if (num > ACTIVE_EXPIRE_CYCLE_LOOKUPS_PER_LOOP)
    num = ACTIVE_EXPIRE_CYCLE_LOOKUPS_PER_LOOP;
   // 从设置了过期时间的 key 中随机抽取 20 个
   while (num--) {
    dictEntry *de;
    long long ttl;
    // 随机挑选 dict 中的一个 key
    if ((de = dictGetRandomKey(db->expires)) == NULL) break;
    ttl = dictGetSignedIntegerVal(de)-now;
    // 执行删除，具体删除操作和惰性删除中类似
    if (activeExpireCycleTryExpire(db,de,now)) expired++;
    // ...
   }
   // ...
   // 更新统计数据等操作
   // ...
  // 如果每次删除的 key 超过了样本数的 25%，说明过期键占的比例较高，需要再重复执行依次
  } while (expired > ACTIVE_EXPIRE_CYCLE_LOOKUPS_PER_LOOP/4);
 }
 // ...
}

// 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;
 // 没有数据，返回为 NULL，外层函数接收到 NULL 后会中断过期操作的执行
 if (dictSize(d) == 0) return NULL;
 // 根据 rehashidx 参数判断是否正在执行 rehash，如果正在执行，
 // 则先执行 rehash 中的一个步骤
 if (dictIsRehashing(d)) _dictRehashStep(d);
 if (dictIsRehashing(d)) {
  do {
   // 正在执行 rehash，所以两个 ht 中的对象都要考虑
   //
   // 由于正在执行 rehash，所以可以肯定 ht[0] 中下标小于等于 rehashidx 的 bucket
   // 肯定没有数据，所以只从 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 时计算下标的规则 hashFunc(key) & sizemake
   // 这里 random() & sizemask 是随机取一个下标
   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++;
 }
 // 随机取链表中某个对象的下标
 listele = random() % listlen;
 he = orighe;
 // 重新遍历链表获取指定下标的对象
 while(listele--) he = he->next;
 return he;
}

// 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);
 }
}

// 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> 触发 EOF
  if (filename[0] == '-' && filename[1] == '\0') {
   fp = stdin;
  } else {
  // 第一个参数不是 -，则尝试打开这个参数指定的文件
   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 后面的参数，直接加到 config 中
 if (options) {
  config = sdscat(config,"\n");
  config = sdscat(config,options);
 }
 loadServerConfigFromString(config);
 sdsfree(config);
}

// 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);
  // 将配置命令转化成小写
  sdstolower(argv[0]);
  // 根据配置命令初始化配置，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 将字符串形式的配置转化成对应的 long long 值
   // 例如 1kb -> 1024
   server.maxmemory = memtoll(argv[1],NULL);
  }
 }
}

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

if (0) {
 // ...
} config_set_memory_field("maxmemory",server.maxmemory) {
 // 配置不为 0，表示之前限制过内存
 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();
 }
 // ...
}

// 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;
 }

typedef struct redisObject {
 // 对象类型
 unsigned type:4;
 // 对象编码
 unsigned encoding:4;
 // LRU 算法和 LFU 算法公用 lru 这个字段
 // 
 // LRU_BITS 默认为 24，因此最大只能存储 194 天的时间戳,
 // 创建对象时会写入这个字段，访问对象时会更新这个字段，
 // 超过之后再从 0 开始计算
 unsigned lru:LRU_BITS;
 int refcount;
 void *ptr;
} robj;

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

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

unsigned int getLRUClock(void) {
 // mstime() 获取 unix 时间戳，单位时毫秒
 // 除以 LRU_CLOCK_RESOLUTION(值为 1000)，将时间戳转化为秒
 return (mstime()/LRU_CLOCK_RESOLUTION) & LRU_CLOCK_MAX;
}

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

// src/evict.c
// 以秒为精度计算对象距离上一次访问的间隔时间，然后转化成毫秒返回
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;
 }
}

uint8_t LFULogIncr(uint8_t counter) {
 // counter 值已经到达了 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;
}

// 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  |
+--------+------------+------------+------------+------------+------------+

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) {
 // 取当前时间戳（单位：分钟）
 unsigned long now = LFUGetTimeInMinutes();
 // 计算时间差
 if (now >= ldt) return now-ldt;
 return 65535-ldt+now;
}
// 获取当前时间戳，以分钟为单位，取低 8 位
unsigned long LFUGetTimeInMinutes(void) {
 return (server.unixtime/60) & 65535;
}

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

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 淘汰
  // 那么还要将对象转化成数据字典中对应的对象，然后取其值
  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 越大的越应该被先淘汰
  //
  // 如果使用 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 值进行比较，找出应该比当前对象先淘汰出去的对象
  while (k < EVPOOL_SIZE &&
    pool[k].key &&
    pool[k].idle < idle) k++;
  if (k == 0 && pool[EVPOOL_SIZE-1].key != NULL) {
   // 没有发现更需要被淘汰的对象，并且 pool 中也没有多余的位置
   // 那么当前对象仍然留在 samples 中
   continue;
  } else if (k < EVPOOL_SIZE && pool[k].key == NULL) {
   // 没有发现更需要被淘汰的对象，但 pool 中有多余的位置
   // 于是将这个对象插入 pool 中
  } else {
   //     当前对象
   //      |
   //      V
   // Pool: [ 0 1 2 3 ...k-1 k ... EVPOOL_SIZE-1]
   // 为了保证 pool 中的数据按 idle 从小到大排列，这里将当前对象插入第 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);
   }
  }
  // ...
  // 将当前对象的属性赋值到下标为 k 的元素
  // ...
 }
}