详解Redis 缓存删除机制(源码解析)
删除的范围
- 过期的 key
- 在内存满了的情况下，如果继续执行 set 等命令，且所有 key 都没有过期，那么会按照缓存淘汰策略选中的 key
过期删除
redis 中设置了过期时间的 key 会单独存储一份
typedef struct redisDb {
dict *dict; // 所有的键值对
dict *expires; //设置了过期时间的键值对
// ...
} redisDb;
设置有效期
Redis 中有 4 个命令可以给 key 设置过期时间，分别是 expire pexpire expireat pexpireat
设置相对时间
expire <key> <ttl>：将 key 值的过期时间设置为 ttl 秒。
// src/expire.c
/* EXPIRE key seconds */
void expireCommand(client *c) {
expireGenericCommand(c,mstime(),UNIT_SECONDS);
}
pexpire <key> <ttl>：将 key 值的过期时间设置为 ttl 毫秒。
// src/expire.c
/* PEXPIRE key milliseconds */
void pexpireCommand(client *c) {
expireGenericCommand(c,mstime(),UNIT_MILLISECONDS);
}
设置绝对时间
expireat <key> <timestamp>：将 key 值的过期时间设置为指定的 timestamp 秒数。
// src/expire.c
/* EXPIREAT key time */
void expireatCommand(client *c) {
expireGenericCommand(c,0,UNIT_SECONDS);
}
pexpireat <key> <timestamp>：将 key 值的过期时间设置为指定的 timestamp 毫秒数。
// src/expire.c
/* PEXPIREAT key ms_time */
void pexpireatCommand(client *c) {
expireGenericCommand(c,0,UNIT_MILLISECONDS);
}
以上 4 种方法最终都会调用下面的通用函数 expireGenericCommand ：
// 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;
}
}
设置过期时间的操作由 setExpire 执行，他将 dictEntry 的 union v 中的 s64 设为过期时间
// 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)
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 剩余过期秒数。
// src/expire.c
/* TTL key */
void ttlCommand(client *c) {
ttlGenericCommand(c, 0);
}
pttl key 返回 key 剩余过期的毫秒数。
// src/expire.c
/* PTTL key */
void pttlCommand(client *c) {
ttlGenericCommand(c, 1);
}
以上 2 种查看方式最终都会调用下面的通用函数 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 {
// 根据指定的单位返回结果，以秒为单位时向上取整
addReplyLongLong(c,output_ms ? ttl : ((ttl+500)/1000));
}
}
获取过期时间的操作由 getExpire 执行，在 db->expires 中查询到对象后，获取 union v 中的成员 s64
// 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)
过期策略
Redis 综合使用 惰性删除 和 定期扫描 实现
惰性删除
每次访问时会调用 expireIfNeeded 判断 key 是否过期，如果过期就删除该键，否则返回键对应的值。单独使用这种策略可能会浪费很多内存。
// 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);
}
特殊处理
在 master 节点执行 Lua 脚本时
参考 GitHub Issue #1525，对于 master，在执行 Lua Script 的过程中，可能会用某个 key 是否存在当作判断条件。为了避免一个脚本中前后条件不一致，将当前时间强制设为脚本开始时间。
例如多次执行如下 Lua 脚本 /tmp/myscript.lua 出现的结果可能不一致
-- /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
具体复现操作可以参考下面的 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 节点
在 slave 节点上，key 的删除操作由 master 发来的 DEL 执行，因此这里只返回是否过期的结果给客户端，而不执行删除操作
正在从 RDB 和 AOF 读取数据时跳过这个步骤
定期扫描
系统每隔一段时间就定期扫描一次，发现过期的键就进行删除。单独使用这种策略可能出现键已经过期但没有删除的情况
Redis 默认每 100ms 执行一次（通过 hz 参数配置，执行周期为 1s/hz）过期扫描。由于 redisDb 中设置了过期时间的 key 会单独存储，所以不会出现扫描所有 key 的情况
具体步骤由 activeExpireCycle 函数执行
activeExpireCycle、incrementallyRehash 等后台操作都是由 databasesCron 触发的
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);
}
// ...
}
随机抽样由 dictGetRandomKey 执行
// 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;
}
缓存淘汰
配置最大内存限制
在 redis.conf 中配置
redis server 启动时加载配置文件和命令行参数中的 maxmemory ，存入 Server 对象的 maxmemory 字段
main 中在 redis server 启动时执行初始化等操作，其中会执行加载配置文件的 loadServerConfig 函数
// 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、命令行参数加载到 config 字符串中，然后调用 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> 触发 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);
}
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);
// 将配置命令转化成小写
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);
}
}
}
使用 CONFIG SET 命令配置
Redis Server 接收到客户端的 CONFIG SET 命令后调用 configSetCommand 函数
服务端接收到命令时将命令和参数存入 Redis Server 的 argc 和 argv
argc: 4 argv: 0 1 2 3 config set maxmemory 10mb
动态配置 maxmemory 后会立即尝试触发内存回收，而修改其他内存相关配置(例如: maxmemory_policy)时不会触发
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();
}
// ...
}
32 位机器的内存限制
对于 64 位机器，将 maxmemory 设置为 0 表示不限制内存，但由于 32 位寻址空间最多只有 4 GB，所以默认内存限制设为 3 GB，缓存淘汰策略设为 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 配置
默认：
- **noeviction: **内存满了后对于 set 等命令直接返回错误
针对所有 key：
- allkeys-lru: 在所有 key 的范围内使用 LRU 算法执行删除，如果内存仍然不够，则报错
- **allkeys-lfu: **在所有 key 的范围内使用 LRU 算法执行删除，如果内存仍然不够，则报错
- **allkeys-random: **在所有 key 的范围内随机执行删除，如果内存仍然不够，则报错
针对设置了过期时间的 key：
- **volatile-lru: **在设置了过期时间的 key 中使用 LRU 算法执行删除，如果内存仍然不够，则报错
- **volatile-lfu: **在设置了过期时间的 key 中使用 LRU 算法执行删除，如果内存仍然不够，则报错
- **volatile-random: **在设置了过期时间的 key 中随机执行删除，如果内存仍然不够，则报错
- **volatile-ttl: **删除即将过期的 key，如果内存仍然不够，则报错
Redis 在执行淘汰之前会计算部分对象的 idle 值，使用不同淘汰策略时计算 idle 值的方法也不同， idle 值越大表示这个值越需要优先删除。
下面主要介绍 LRU 和 LFU 中 idle 值的计算方法
LRU 淘汰策略
抽样删除，样本数量通过 CONFIG SET maxmemory-samples 100 控制，对应 RedisObject 中的 maxmemory_samples 参数，抽样数量越多和传统的 LRU 算法越接近
优化策略
为了避免传统的 LRU 算法通常使用 hashmap + 链表实现带来的开销，Redis 进行了如下优化：
RedisObject 结构中设置了一个 lru 字段，用来记录数据的访问时间戳，而不是每次调整对象在链表中的位置
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;
使用抽样数组代替链表，后续在候选集合中根据 lru 字段值的大小进行筛选，避免了链表带来的开销。候选集合中的对象用 evictionPoolEntry 表示
struct evictionPoolEntry {
unsigned long long idle; // 用于淘汰排序，在不同算法中意义不同
sds key; // 键的名字
// ...
};
计算方法
全局对象 lru_clock 记录了当前的 unix 时间戳，由 serverCron 调用 updateCachedTime 默认每 100 ms 更新一次。更新频率与 hz 参数有关， 1s/hz 即为更新间隔时间。
LRU_CLOCK_RESOLUTION 的值为 1000，因此使用 LRU_CLOCK 函数获取 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 次，则会用函数 getLRUClock 实时计算 lruclock
unsigned int getLRUClock(void) {
// mstime() 获取 unix 时间戳，单位时毫秒
// 除以 LRU_CLOCK_RESOLUTION(值为 1000)，将时间戳转化为秒
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] 的范围内。
从逻辑上讲， 当前时间戳应该永远大于上次访问的时间戳 ,因此正常的计算规则应该是 lru_clock-robj.lru 。
但是由于 lru_clock 和 robj.lru 是当前时间戳取模后的值，所以可能出现 lru_clock 小于 robj.lru 的情况，所以这种情况下计算规则应该改为 lru_clock+194天-robj.lru
但是对于 lru_clock 和 robj.lru 间隔超过 194 天的情况仍然无法判断，所以更能存在删除不准确的情况。
将上述的逻辑组合起来就是 LRU 算法下获取 idle 值的函数:
// 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;
}
}
在 Redis 3.0 中，当取样数量设为 10 时，已经和传统的 LRU 算法效果很接近了
LFU 淘汰策略
LFU 算法复用 robj.lru 字段，将这个 24 bit 的字段拆分成了两部分：
- ldt(last decrement time，单位：分钟)：lru 字段的前 16bit，表示数据的访问时间戳，最多只能存储 45 天。
- counter 值：lru 字段的后 8bit，表示数据的访问频率
递增策略
counter 能表示的最大值是 255，因此 counter 与访问次数不能是线性关系，这里采用的计算步骤如下：
- 随机取 0 到 1 之间的随机数 r
- 比较 r 与 1/((counter-LFU_INIT_VAL)*lfu_log_factor+1) 的大小，其中 LFU_INIT_VAL 是常量，默认为 5，lfu_log_factor 是可配置参数，默认为 10
- 如果 r 小则 counter 增加 1，否则 counter 不变
实现代码如下：
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;
}
访问次数与 counter 值之间大概是对数关系，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 需要增加，对于一段时间内没有访问的对象还要相应地减少 counter 值，递减的速率由 lfu-decay-time 参数控制。
counter 衰减步骤如下：
取当前时间戳（单位：分钟）的低 16 位记为 now ，计算与 ldt 的差值。这里与 LRU 算法中计算 lru_clock 和 robj.lru 时可能出现一样的问题，由于 ldt 最多只能表示 45 天，所以如果距离对象上次访问超过 45 天，则无法准确计算访问的时间间隔
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;
}
如果 lfu_decay_time 为 0，则步修改 counter，否则将 counter 减少 LFUTimeElapsed(ldt)/lfu_decay_time
例如，在 lfu_decay_time 为 1 的情况下，如果有 N 分钟没有访问这个对象，那么 counter 值减 N
每次访问一个对象时都会调用 updateLFU 更新 counter 的值：
void updateLFU(robj *val) {
unsigned long counter = LFUDecrAndReturn(val);
counter = LFULogIncr(counter);
val->lru = (LFUGetTimeInMinutes()<<8) | counter;
}
执行淘汰
当 Redis 需要淘汰一批数据时，会调用 evictionPoolPopulate 获取一批待删除对象，根据设置的淘汰范围的不同，会决定传递给 evictionPoolPopulate 的 sampledict 参数是存有全部数据的 db->dict 还是只有设置了过期时间的数据的 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 淘汰
// 那么还要将对象转化成数据字典中对应的对象，然后取其值
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 的元素
// ...
}
}
完成上述操作后，pool 中剩下的就是新筛选出来的最需要淘汰的对象了。
在 freeMemoryIfNeeded 中会调用 evictionPoolPopulate 来筛选需要淘汰的对象，每次删除一个，直到释放了足够的内存。如果始终无法达到内存需求，则会报错。
到此这篇关于Redis 缓存删除机制(源码解析)的文章就介绍到这了,更多相关Redis 缓存删除内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！
相关文章
为了提高数据处理效率和降低存储成本,优化数据结构和采用高效的存储策略至关重要,使用最小存储形式、整数编码、Redis的HyperLogLog等方法可以有效减少内存占用,Redis6引入的对象压缩、设置合理的过期时间、数据分片2024-09-09
Redis提供了一种过期回调的机制,可以在某个键过期时触发一个回调函数,然而,在实际使用中,我们往往会遇到一些灾难性的问题,其中一个就是在使用过期回调的时候,我们可能会遭遇到无法预料的错误,本文就详细的介绍一下2023-09-09
本文主要介绍了redis搭建哨兵集群的实现步骤，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧2022-05-05
这篇文章主要介绍了Redis数据类型实现原理，在工作中或学习中有需要的小伙伴可以参考一下这篇文章2021-08-08
这篇文章主要介绍了Redis如何实现投票功能,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教2024-05-05
SDS是Redis中实现的一种数据结构，用来存储字符串，最近学习中正好学习到了这里，所以下面这篇文章主要给大家介绍了redis内部数据结构之SDS简单动态字符串的相关资料，文中通过示例代码介绍的非常详细，需要的朋友可以参考借鉴，下面来一起看看吧。2017-11-11
Redis报错“NOAUTH Authentication required”两种解决方案
Redis提供了一个命令行工具redis-cli,它允许你直接连接到Redis服务器,如果你知道Redis服务器的密码,你可以在连接时直接提供它,本文给大家介绍连接了Redis报错“NOAUTH Authentication required”两种解决方案2024-05-05
Redis中支持五种数据类型中有序集合Sorted Set的底层数据结构使用的跳跃表,为何不使用其他的如平衡二叉树、b+树等数据结构呢？这篇文章主要介绍了redis使用跳跃表而不是树的原因解析,需要的朋友可以参考下2024-02-02
这篇文章主要介绍了Redis集群利用Redisson实现分布式锁方式,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教2024-05-05
分布式锁为什么要选择Zookeeper而不是Redis?看完这篇你就明白了
Zookeeper的机制可以保证分布式锁实现业务代码简单，成本低，Redis如果要解决分布式锁的问题，对于一些复杂的情况，很难解决，成本较高，这篇文章重点给大家介绍分布式锁选择Zookeeper 而不是Redis的理由，一起看看吧2021-05-05
最新评论