前言
Redis 中 key 的过期删除策略
1、定时删除
2、惰性删除
3、定期删除
Redis 中过期删除策略
从库是否会脏读主库创建的过期键
内存淘汰机制
内存淘汰触发的最大内存
有哪些内存淘汰策略
内存淘汰算法
LRU
LFU
总结
参考
Redis 中 key 的过期删除策略
前言
Redis 中的 key 设置一个过期时间，在过期时间到的时候，Redis 是如何清除这个 key 的呢？
这来分析下 Redis 中的过期删除策略和内存淘汰机制
Redis 中 key 的过期删除策略
Redis 中提供了三种过期删除的策略
1、定时删除
在设置某个 key 的过期时间同时，我们创建一个定时器，让定时器在该过期时间到来时，立即执行对其进行删除的操作。
优点：
通过使用定时器，可以保证过期 key 可以被尽快的删除，并且释放过期 key 所占用的内存
缺点：
对 CPU 是不友好的，当过期键比较多的时候，删除过期 key 会占用相当一部分的 CPU 资源，对服务器的响应时间和吞吐量造成影响。
2、惰性删除
惰性删除，当一个键值对过期的时候，只有再次用到这个键值对的时候才去检查删除这个键值对，也就是如果用不着，这个键值对就会一直存在。
优点：
对 CPU 是友好的，只有在取出键值对的时候才会进行过期检查，这样就不会把 CPU 资源花费在其他无关紧要的键值对的过期删除上。
缺点：
如果一些键值对永远不会被再次用到，那么将不会被删除，最终会造成内存泄漏，无用的垃圾数据占用了大量的资源，但是服务器却不能去删除。
看下源码
// https://github.com/redis/redis/blob/6.2/src/db.c#L1541
// 当访问到 key 的时候，会调用这个函数，因为有的 key 虽然已经过期了，但是还可能存在于内存中
// key 仍然有效，函数返回值为0，否则，如果 key 过期，函数返回1。
int expireIfNeeded(redisDb *db, robj *key) {
// 没有过期
if (!keyIsExpired(db,key)) return 0;
// 从库的过期是主库控制的，是不会进行删除操作的
// 上面已经判断过是否到期了，所以这里的 key 肯定是过期的 key ,不过如果是主节点创建的 key 从节点就不删除，只会返回已经过期了
if (server.masterhost != NULL) return 1;
...
/* Delete the key */
// 删除 key
deleteExpiredKeyAndPropagate(db,key);
return 1;
}
可以看到每次操作对应的 key 是会检查 key 是否过期，如果过期则会删除对应的 key 。
如果过期键是主库创建的，那么从库进行检查是不会进行删除操作的,只是会根据 key 的过期时间返回过期或者未过期的状态。
3、定期删除
定期删除是对上面两种删除策略的一种整合和折中
每个一段时间就对一些 key 进行采样检查，检查是否过期，如果过期就进行删除
1、采样一定个数的key，采样的个数可以进行配置，并将其中过期的 key 全部删除；
2、如果过期 key 的占比超过可接受的过期 key 的百分比，则重复删除的过程，直到过期key的比例降至可接受的过期 key 的百分比以下。
优点：
定期删除，通过控制定期删除执行的时长和频率，可以减少删除操作对 CPU 的影响，同时也能较少因过期键带来的内存的浪费。
缺点：
执行的频率不太好控制
频率过快对 CPU 不友好，如果过慢了就会对内存不太友好，过期的键值对不能及时的被删除掉
同时如果一个键值对过期了，但是没有被删除，这时候业务再次获取到这个键值对，那么就会获取到被删除的数据了，这肯定是不合理的。
看下源码实现
// https://github.com/redis/redis/blob/6.2/src/server.c#L1853
// 这个函数处理我们需要在Redis数据库中增量执行的“后台”操作，例如活动键过期，调整大小，重哈希。
void databasesCron(void) {
// 通过随机抽样来过期
// 这里区分了主节点和从节点的处理
if (server.active_expire_enabled) {
if (iAmMaster()) {
activeExpireCycle(ACTIVE_EXPIRE_CYCLE_SLOW);
} else {
expireSlaveKeys();
}
}
...
}
// https://github.com/redis/redis/blob/6.2/src/expire.c#L113
void activeExpireCycle(int type) {
// 根据配置的超时工作调整运行参数。默认工作量为1，最大可配置工作量为10
unsigned long
effort = server.active_expire_effort-1, /* Rescale from 0 to 9. */
// 采样的 key 的数量
config_keys_per_loop = ACTIVE_EXPIRE_CYCLE_KEYS_PER_LOOP +
ACTIVE_EXPIRE_CYCLE_KEYS_PER_LOOP/4*effort,
// 占比CPU时间，默认是25%，最大43%，如果是100%，那除了定时删除其他的工作都做不了了，所以要做限制
config_cycle_slow_time_perc = ACTIVE_EXPIRE_CYCLE_SLOW_TIME_PERC +
2*effort,
// 可接受的过期 key 的百分比
config_cycle_acceptable_stale = ACTIVE_EXPIRE_CYCLE_ACCEPTABLE_STALE-
effort;
...
//慢速定期删除的执行时长
timelimit = config_cycle_slow_time_perc*1000000/server.hz/100;
timelimit_exit = 0;
...
// 在 key 过期时积累一些全局统计信息，以便了解逻辑上已经过期但仍存在于数据库中的 key 的数量
long total_sampled = 0;
long total_expired = 0;
for (j = 0; j < dbs_per_call && timelimit_exit == 0; j++) {
...
// 如果超过 config_cycle_acceptable_stale 的key过期了，则重复删除的过程，直到过期key的比例降至 config_cycle_acceptable_stale 以下。
// 存储在 config_cycle_acceptable_stale 中的百分比不是固定的，而是取决于Redis配置的“expire efforce”
do {
/* If there is nothing to expire try next DB ASAP. */
if ((num = dictSize(db->expires)) == 0) {
db->avg_ttl = 0;
break;
}
...
// 采样的 key 的数量
if (num > config_keys_per_loop)
num = config_keys_per_loop;
...
while (sampled < num && checked_buckets < max_buckets) {
for (int table = 0; table < 2; table++) {
...
while(de) {
/* Get the next entry now since this entry may get
* deleted. */
dictEntry *e = de;
de = de->next;
ttl = dictGetSignedIntegerVal(e)-now;
// 过期检查，并对过期键进行删除
if (activeExpireCycleTryExpire(db,e,now)) expired++;
...
}
}
db->expires_cursor++;
}
...
// 判断过期 key 的占比是否大于 config_cycle_acceptable_stale，如果大于持续进行过期 key 的删除
} while (sampled == 0 ||
(expired*100/sampled) > config_cycle_acceptable_stale);
}
...
}
// 检查删除由从节点创建的有过期的时间的 key
void expireSlaveKeys(void) {
// 从主库同步的 key，过期时间由主库维护，主库同步 DEL 操作到从库。
// 从库如果是 READ-WRITE 模式，就可以继续写入数据。从库自己写入的数据就需要自己来维护其过期操作。
if (slaveKeysWithExpire == NULL ||
dictSize(slaveKeysWithExpire) == 0) return;
...
}
惰性删除过程
1、固定的时间执行一次定期删除；
2、采样一定个数的key，采样个数可以进行配置，并将其中过期的key全部删除；
3、如果过期 key 的占比超过可接受的过期 key 的百分比，则重复删除的过程，直到过期key的比例降至可接受的过期 key 的百分比以下；
4、对于从库创建的过期 key 同样从库是不能进行删除的。
Redis 中过期删除策略
上面讨论的三种策略，都有或多或少的问题。Redis 中实际采用的策略是惰性删除加定期删除的组合方式。
组合方式的使用
定期删除，获取 CPU 和 内存的使用平衡，针对过期的 KEY 可能得不到及时的删除，当 KEY 被再次获取的时候，通过惰性删除再做一次过期检查，来避免业务获取到过期内容。
从库是否会脏读主库创建的过期键
从上面惰性删除和定期删除的源码阅读中，我们可以发现，从库对于主库的过期键是不能主动进行删除的。如果一个主库创建的过期键值对，已经过期了，主库在进行定期删除的时候，没有及时的删除掉，这时候从库请求了这个键值对，当执行惰性删除的时候，因为是主库创建的键值对，这时候是不能在从库中删除的，那么是不是就意味着从库会读取到已经过期的数据呢？
答案肯定不是的
how-redis-replication-deals-with-expires-on-keys
How Redis replication deals with expires on keys
Redis expires allow keys to have a limited time to live. Such a feature depends on the ability of an instance to count the time, however Redis slaves correctly replicate keys with expires, even when such keys are altered using Lua scripts.
To implement such a feature Redis cannot rely on the ability of the master and slave to have synchronized clocks, since this is a problem that cannot be solved and would result into race conditions and diverging data sets, so Redis uses three main techniques in order to make the replication of expired keys able to work:
1.Slaves don’t expire keys, instead they wait for masters to expire the keys. When a master expires a key (or evict it because of LRU), it synthesizes a DEL command which is transmitted to all the slaves.
2.However because of master-driven expire, sometimes slaves may still have in memory keys that are already logically expired, since the master was not able to provide the DEL command in time. In order to deal with that the slave uses its logical clock in order to report that a key does not exist only for read operations that don’t violate the consistency of the data set (as new commands from the master will arrive). In this way slaves avoid to report logically expired keys are still existing. In practical terms, an HTML fragments cache that uses slaves to scale will avoid returning items that are already older than the desired time to live.
3.During Lua scripts executions no keys expires are performed. As a Lua script runs, conceptually the time in the master is frozen, so that a given key will either exist or not for all the time the script runs. This prevents keys to expire in the middle of a script, and is needed in order to send the same script to the slave in a way that is guaranteed to have the same effects in the data set.
Once a slave is promoted to a master it will start to expire keys independently, and will not require any help from its old master.
上面是官方文档中针对这一问题的描述
大概意思就是从节点不会主动删除过期键，从节点会等待主节点触发键过期。当主节点触发键过期时，主节点会同步一个del命令给所有的从节点。
因为是主节点驱动删除的，所以从节点会获取到已经过期的键值对。从节点需要根据自己本地的逻辑时钟来判断减值是否过期，从而实现数据集合的一致性读操作。
我们知道 Redis 中的过期策略是惰性删除和定期删除，所以每个键值的操作，都会使用惰性删除来检查是否过期，然后判断是否可以进行删除
// https://github.com/redis/redis/blob/6.2/src/db.c#L1541
// 当访问到 key 的时候，会调用这个函数，因为有的 key 虽然已经过期了，但是还可能存在于内存中
// key 仍然有效，函数返回值为0，否则，如果 key 过期，函数返回1。
int expireIfNeeded(redisDb *db, robj *key) {
// 检查 key 是否过期
if (!keyIsExpired(db,key)) return 0;
// 从库的过期是主库控制的，是不会进行删除操作的
// 上面已经判断过是否到期了，所以这里的 key 肯定设计过期的 key ,不过如果是主节点创建的 key 从节点就不删除，只会返回已经过期了
if (server.masterhost != NULL) return 1;
...
/* Delete the key */
// 删除 key
deleteExpiredKeyAndPropagate(db,key);
return 1;
}
// https://github.com/redis/redis/blob/6.2/src/db.c#L1485
/* Check if the key is expired. */
int keyIsExpired(redisDb *db, robj *key) {
// 过期时间
mstime_t when = getExpire(db,key);
mstime_t now;
// 没有过期
if (when < 0) return 0; /* No expire for this key */
/* Don't expire anything while loading. It will be done later. */
if (server.loading) return 0;
// lua 脚本执行的过程中不过期
if (server.lua_caller) {
now = server.lua_time_snapshot;
}
// 如果我们正在执行一个命令，我们仍然希望使用一个不会改变的引用时间:在这种情况下，我们只使用缓存的时间，我们在每次调用call()函数之前更新。
// 这样我们就避免了RPOPLPUSH之类的命令，这些命令可能会重新打开相同的键多次，如果下次调用会看到键过期，则会使已经打开的对象在下次调用中失效，而第一次调用没有。
else if (server.fixed_time_expire > 0) {
now = server.mstime;
}
// 其他情况下，获取最新的时间
else {
now = mstime();
}
// 判断是否过期了
return now > when;
}
// 返回指定 key 的过期时间，如果没有过期则返回-1
long long getExpire(redisDb *db, robj *key) {
dictEntry *de;
/* No expire? return ASAP */
if (dictSize(db->expires) == 0 ||
(de = dictFind(db->expires,key->ptr)) == NULL) return -1;
/* The entry was found in the expire dict, this means it should also
* be present in the main dict (safety check). */
serverAssertWithInfo(NULL,key,dictFind(db->dict,key->ptr) != NULL);
return dictGetSignedIntegerVal(de);
}
上面的惰性删除，对于主节点创建的过期 key ，虽然不能进行删除的操作，但是可以进行过期时间的判断，所以如果主库创建的过期键，如果主库没有及时进行删除，这时候从库可以通过惰性删除来判断键值对的是否过期，避免读取到过期的内容。
内存淘汰机制
上面我们讨论的 Redis 过期策略指的是 Redis 使用那种策略，来删除已经过期的键值对。但是有一些 key以后永远用不到了，那么就可能一直不能被删除掉，还有就是 Redis 中的使用过程中，随着写数据的增加，Redis 中的内存不够用了，这时候就需要 Redis 的内存淘汰策略了。
Redis 过期策略指的是 Redis 使用那种策略，来删除已经过期的键值对；
Redis 内存淘汰机制指的是，当 Redis 运行内存已经超过 Redis 设置的最大内存之后，将采用什么策略来删除符合条件的键值对，以此来保障 Redis 高效的运行。
内存淘汰触发的最大内存
Redis 中的内存只有达到了阀值，才会触发内存淘汰算法,这个阀值就是我们设置的最大运行内存，在配置文件redis.conf中，通过参数 maxmemory <bytes> 来设置
查询最大运行内存
127.0.0.1:6379> config get maxmemory
1) "maxmemory"
2) "0"
在 64 位操作系统中，当 maxmemory 为 0 时，表示没有内存大小限制，32位的系统。
有哪些内存淘汰策略
当现有内存大于 maxmemory 时，便会触发redis主动淘汰内存方式，通过设置 maxmemory-policy ，有如下几种淘汰方式：
1、volatile-lru:淘汰所有设置了过期时间的键值中最久未使用的键值；
2、allkeys-lru:淘汰整个键值中最久未使用的键值；
3、volatile-random:随机淘汰设置了过期时间的任意键值；
4、allkeys-random:随机淘汰任意键值;
5、volatile-ttl:优先淘汰更早过期的键值;
6、noeviction:不淘汰任何数据，当内存不足时，新增操作会报错，Redis 默认内存淘汰策略；
其中 allkeys-xxx 表示从所有的键值中淘汰数据，而 volatile-xxx 表示从设置了过期键的键值中淘汰数据。
内存淘汰算法
除了随机删除和不删除之外，主要有两种淘汰算法：LRU 算法和 LFU 算法。
LRU
LRU 全称是Least Recently Used译为最近最少使用，是一种常用的页面置换算法，选择最近最久未使用的页面予以淘汰。
一般 LRU 算法的实现基于链表结构，链表中的元素按照操作顺序从前往后排列，最新操作的键会被移动到表头，当需要内存淘汰时，只需要删除链表尾部的元素即可。
Redis 使用的是一种近似 LRU 算法，目的是为了更好的节约内存，它的实现方式是给现有的数据结构添加一个额外的字段，用于记录此键值的最后一次访问时间，Redis 内存淘汰时，会使用随机采样的方式来淘汰数据，它是随机取 5 个值（此值可配置），然后淘汰最久没有使用的那个。
这里看下是如何实现的呢
Redis 在源码中对于每个键值对中的值，会使用一个 redisObject 结构体来保存指向值的指针，这里先来看下 redisObject 的结构
// https://github.com/redis/redis/blob/6.2/src/server.h#L673
typedef struct redisObject {
unsigned type:4;
unsigned encoding:4;
// 这里保存
// LRU时间(相对于全局LRU时钟)
// LFU数据 (低 8 bits 作为计数器,用 24 bits 中的高 16 bits，记录访问的时间戳)
unsigned lru:LRU_BITS; /* LRU time (relative to global lru_clock) or
* LFU data (least significant 8 bits frequency
* and most significant 16 bits access time). */
int refcount;
void *ptr;
} robj;
当一个键值对被创建的时候，就会记录下更新的时间
// https://github.com/redis/redis/blob/6.2/src/object.c#L41
robj *createObject(int type, void *ptr) {
robj *o = zmalloc(sizeof(*o));
o->type = type;
o->encoding = OBJ_ENCODING_RAW;
o->ptr = ptr;
o->refcount = 1;
// 如果缓存替换策略是LFU，那么将lru变量设置为LFU的计数值
if (server.maxmemory_policy & MAXMEMORY_FLAG_LFU) {
o->lru = (LFUGetTimeInMinutes()<<8) | LFU_INIT_VAL;
} else {
// 如果是 lru
// 调用LRU_CLOCK函数获取LRU时钟值
o->lru = LRU_CLOCK();
}
return o;
}
同时一个键值对被访问的时候记录的时间也会被更新，当一个键值对被访问时，访问操作最终都会调用 lookupKey 函数。
// https://github.com/redis/redis/blob/6.2/src/db.c#L63
robj *lookupKey(redisDb *db, robj *key, int flags) {
dictEntry *de = dictFind(db->dict,key->ptr);
if (de) {
robj *val = dictGetVal(de);
/* Update the access time for the ageing algorithm.
* Don't do it if we have a saving child, as this will trigger
* a copy on write madness. */
if (!hasActiveChildProcess() && !(flags & LOOKUP_NOTOUCH)){
if (server.maxmemory_policy & MAXMEMORY_FLAG_LFU) {
updateLFU(val);
} else {
// 使用 LRU 更新 lru 的时间
val->lru = LRU_CLOCK();
}
}
return val;
} else {
return NULL;
}
}
上面我们分别看了，创建和访问一个键值对的代码，每次操作，redisObject 中记录的 lru 时间就会被同步的更新
Redis 会判断当前内存的使用情况，如果超过了 maxmemory 配置的值，就会触发新的内存淘汰了
如果内存超过了 maxmemory 的值，这时候还需要去计算需要释放的内存量，这个释放的内存大小等于已使用的内存量减去 maxmemory。不过，已使用的内存量并不包括用于主从复制的复制缓冲区大小。
// https://github.com/redis/redis/blob/6.2/src/evict.c#L512
int performEvictions(void) {
...
while (mem_freed < (long long)mem_tofree) {
int j, k, i;
static unsigned int next_db = 0;
sds bestkey = NULL;
int bestdbid;
redisDb *db;
dict *dict;
dictEntry *de;
if (server.maxmemory_policy & (MAXMEMORY_FLAG_LRU|MAXMEMORY_FLAG_LFU) ||
server.maxmemory_policy == MAXMEMORY_VOLATILE_TTL)
{
struct evictionPoolEntry *pool = EvictionPoolLRU;
while(bestkey == NULL) {
unsigned long total_keys = 0, keys;
/* We don't want to make local-db choices when expiring keys,
* so to start populate the eviction pool sampling keys from
* every DB. */
// 根据淘汰策略，决定使用全局哈希表还是设置了过期时间的key的哈希表
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) {
// 将选择的哈希表dict传入evictionPoolPopulate函数，同时将全局哈希表也传给evictionPoolPopulate函数
evictionPoolPopulate(i, dict, db->dict, pool);
total_keys += keys;
}
}
...
}
}
...
}
// 用来填充evictionPool
// 按升序插入键，所以空闲时间小的键在左边，空闲时间高的键在右边。
// https://github.com/redis/redis/blob/6.2/src/evict.c#L145
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++) {
...
// 将元素插入池中。 首先，找到第一个空闲时间小于我们空闲时间的空桶或第一个填充的桶。
k = 0;
while (k < EVPOOL_SIZE &&
pool[k].key &&
pool[k].idle < idle) k++;
if (k == 0 && pool[EVPOOL_SIZE-1].key != NULL) {
/* Can't insert if the element is < the worst element we have
* and there are no empty buckets. */
continue;
} else if (k < EVPOOL_SIZE && pool[k].key == NULL) {
/* Inserting into empty position. No setup needed before insert. */
} else {
/* Inserting in the middle. Now k points to the first element
* greater than the element to insert. */
if (pool[EVPOOL_SIZE-1].key == NULL) {
/* Free space on the right? Insert at k shifting
* all the elements from k to end to the right. */
/* Save SDS before overwriting. */
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 {
/* No free space on right? Insert at k-1 */
k--;
/* Shift all elements on the left of k (included) to the
* left, so we discard the element with smaller idle time. */
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;
}
}
...
}
}
处理淘汰的数据，Redis 中提供了一个数组 EvictionPoolLRU，用来保存待淘汰的候选键值对。这个数组的元素类型是 evictionPoolEntry 结构体，该结构体保存了待淘汰键值对的空闲时间 idle、对应的 key 等信息。
可以看到上面的上面会选取一定的过期键，然后插入到 EvictionPoolLRU 中
dictGetSomeKeys 函数采样的 key 的数量，是由 redis.conf 中的配置项 maxmemory-samples 决定的，该配置项的默认值是 5
// https://github.com/redis/redis/blob/6.2/src/evict.c#L55
struct evictionPoolEntry {
// 待淘汰的键值对的空闲时间
unsigned long long idle; /* Object idle time (inverse frequency for LFU) */
// 待淘汰的键值对的key
sds key; /* Key name. */
// 缓存的SDS对象
sds cached; /* Cached SDS object for key name. */
// 待淘汰键值对的key所在的数据库ID
int dbid; /* Key DB number. */
};
static struct evictionPoolEntry *EvictionPoolLRU;
然后通过 evictionPoolPopulate 函数，进行采样，然后将采样数据写入到 EvictionPoolLRU 中，插入到 EvictionPoolLRU 中的数据是按照空闲时间从小到大进行排好序的
freeMemoryIfNeeded 函数会遍历一次 EvictionPoolLRU 数组，从数组的最后一个 key 开始选择，如果选到的 key 不是空值，那么就把它作为最终淘汰的 key。
// https://github.com/redis/redis/blob/6.2/src/evict.c#L512
int performEvictions(void) {
if (!isSafeToPerformEvictions()) return EVICT_OK;
int keys_freed = 0;
size_t mem_reported, mem_tofree;
long long mem_freed; /* May be negative */
mstime_t latency, eviction_latency;
long long delta;
int slaves = listLength(server.slaves);
int result = EVICT_FAIL;
if (getMaxmemoryState(&mem_reported,NULL,&mem_tofree,NULL) == C_OK)
return EVICT_OK;
...
while (mem_freed < (long long)mem_tofree) {
if (server.maxmemory_policy & (MAXMEMORY_FLAG_LRU|MAXMEMORY_FLAG_LFU) ||
server.maxmemory_policy == MAXMEMORY_VOLATILE_TTL)
{
struct evictionPoolEntry *pool = EvictionPoolLRU;
while(bestkey == NULL) {
unsigned long total_keys = 0, keys;
...
/* Go backward from best to worst element to evict. */
// 从数组最后一个key开始查找
for (k = EVPOOL_SIZE-1; k >= 0; k--) {
// 当前key为空值，则查找下一个key
if (pool[k].key == NULL) continue;
bestdbid = pool[k].dbid;
// 从全局哈希表或是expire哈希表中，获取当前key对应的键值对；
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);
}
/* Remove the entry from the pool. */
// 将当前key从EvictionPoolLRU数组删除
if (pool[k].key != pool[k].cached)
sdsfree(pool[k].key);
pool[k].key = NULL;
pool[k].idle = 0;
/* If the key exists, is our pick. Otherwise it is
* a ghost and we need to try the next element. */
// 如果当前key对应的键值对不为空，选择当前key为被淘汰的key
if (de) {
bestkey = dictGetKey(de);
break;
} else {
/* Ghost... Iterate again. */
}
}
}
}
...
/* Finally remove the selected key. */
if (bestkey) {
db = server.db+bestdbid;
robj *keyobj = createStringObject(bestkey,sdslen(bestkey));
propagateExpire(db,keyobj,server.lazyfree_lazy_eviction);
delta = (long long) zmalloc_used_memory();
latencyStartMonitor(eviction_latency);
// 惰性删除
if (server.lazyfree_lazy_eviction)
dbAsyncDelete(db,keyobj);
else
// 同步删除
dbSyncDelete(db,keyobj);
...
}
}
...
}
每次选中一部分过期的键值对，每次淘汰最久没有使用的那个，如果释放的内存空间还不够，就会重复的进行采样，删除的过程。
LFU
除了 LRU 算法，Redis 在 4.0 版本引入了 LFU 算法，就是最不频繁使用（Least Frequently Used，LFU）算法。
LRU 算法：淘汰最近最少使用的数据，它是根据时间维度来选择将要淘汰的元素，即删除掉最长时间没被访问的元素。
LFU 算法：淘汰最不频繁访问的数据，它是根据频率维度来选择将要淘汰的元素，即删除访问频率最低的元素。如果两个元素的访问频率相同，则淘汰最久没被访问的元素。
LFU 的基本原理
LFU（Least Frequently Used）算法，即最少访问算法，根据访问缓存的历史频率来淘汰数据，核心思想是“如果数据在过去一段时间被访问的次数很少，那么将来被访问的概率也会很低”。
它是根据频率维度来选择将要淘汰的元素，即删除访问频率最低的元素。如果两个元素的访问频率相同，则淘汰最久没被访问的元素。也就是说 LFU 淘汰的时候会选择两个维度，先比较频率，选择访问频率最小的元素；如果频率相同，则按时间维度淘汰掉最久远的那个元素。
LUF 的实现可参见LFU实现详解
这看下 Redis 中对 LFU 算法的实现
1、键值对的访问频率记录和更新
上面分析 LRU 的时候，聊到了 redisObject，Redis 在源码中对于每个键值对中的值，会使用一个 redisObject 结构体来保存指向值的指针。里面 lru:LRU_BITS 字段记录了 LRU 算法和 LFU 算法需要的时间和计数器。
// https://github.com/redis/redis/blob/6.2/src/server.h#L673
typedef struct redisObject {
unsigned type:4;
unsigned encoding:4;
// 这里保存
// LRU时间(相对于全局LRU时钟)
// LFU数据 (低 8 bits 作为计数器,用 24 bits 中的高 16 bits，记录访问的时间戳)
unsigned lru:LRU_BITS; /* LRU time (relative to global lru_clock) or
* LFU data (least significant 8 bits frequency
* and most significant 16 bits access time). */
int refcount;
void *ptr;
} robj;
当一个键值对被创建的时候，如果使用 LFU 算法，就会更新 lru 字段记录的键值对的访问时间戳和访问次数。
// https://github.com/redis/redis/blob/6.2/src/object.c#L41
robj *createObject(int type, void *ptr) {
robj *o = zmalloc(sizeof(*o));
o->type = type;
o->encoding = OBJ_ENCODING_RAW;
o->ptr = ptr;
o->refcount = 1;
// 如果缓存替换策略是LFU，lru变量包括以分钟为精度的UNIX时间戳和访问次数5
if (server.maxmemory_policy & MAXMEMORY_FLAG_LFU) {
o->lru = (LFUGetTimeInMinutes()<<8) | LFU_INIT_VAL;
} else {
// 如果是 lru
// 调用LRU_CLOCK函数获取LRU时钟值
o->lru = LRU_CLOCK();
}
return o;
}
当一个键值对被访问时，Redis 会调用 lookupKey 函数进行查找。当 maxmemory-policy 设置使用 LFU 算法时，lookupKey 函数会调用 updateLFU 函数来更新键值对的访问频率，也就是 lru 变量值，如下所示：
// https://github.com/redis/redis/blob/6.2/src/db.c#L63
robj *lookupKey(redisDb *db, robj *key, int flags) {
dictEntry *de = dictFind(db->dict,key->ptr);
if (de) {
robj *val = dictGetVal(de);
// 使用LFU算法时，调用updateLFU函数更新访问频率
if (!hasActiveChildProcess() && !(flags & LOOKUP_NOTOUCH)){
if (server.maxmemory_policy & MAXMEMORY_FLAG_LFU) {
updateLFU(val);
} else {
// 使用 LRU 更新 lru 的时间
val->lru = LRU_CLOCK();
}
}
return val;
} else {
return NULL;
}
}
// https://github.com/redis/redis/blob/6.2/src/db.c#L54
/* 访问对象时更新 LFU。
* 首先，如果达到递减时间，则递减计数器。
* 然后对计数器进行对数递增，并更新访问时间。 */
void updateLFU(robj *val) {
unsigned long counter = LFUDecrAndReturn(val);
counter = LFULogIncr(counter);
val->lru = (LFUGetTimeInMinutes()<<8) | counter;
}
// https://github.com/redis/redis/blob/6.2/src/evict.c#L318
unsigned long LFUDecrAndReturn(robj *o) {
// 获取当前键值对的上一次访问时间
unsigned long ldt = o->lru >> 8;
// 获取当前的访问次数
unsigned long counter = o->lru & 255;
unsigned long num_periods = server.lfu_decay_time ? LFUTimeElapsed(ldt) / server.lfu_decay_time : 0;
if (num_periods)
// 如果衰减大小小于当前访问次数，那么，衰减后的访问次数是当前访问次数减去衰减大小；否则，衰减后的访问次数等于0
counter = (num_periods > counter) ? 0 : counter - num_periods;
// 如果衰减大小为0，则返回原来的访问次数
return counter;
}
上面的代码可以看到，当访问一个键值对的时候，首先进行了访问次数的衰减？
LFU 算法是根据访问频率来淘汰数据的，而不只是访问次数。如果访问间隔时间越长，那么访问频率就越低。
因为 Redis 是使用 lru 变量中的访问次数来表示访问频率，所以在每次更新键值对的访问频率时，就会通过 LFUDecrAndReturn 函数对访问次数进行衰减。
LFUDecrAndReturn 函数会调用 LFUTimeElapsed 函数（在 evict.c 文件中），计算距离键值对的上一次访问已经过去的时长。这个时长也是以 1 分钟为精度来计算的。有了距离上次访问的时长后，LFUDecrAndReturn 函数会把这个时长除以 lfu_decay_time 的值，并把结果作为访问次数的衰减大小。
lfu_decay_time 变量值，是由 redis.conf 文件中的配置项 lfu-decay-time 来决定的。Redis 在初始化时，会通过 initServerConfig 函数来设置 lfu_decay_time 变量的值，默认值为 1。所以，在默认情况下，访问次数的衰减大小就是等于上一次访问距离当前的分钟数。
衰减之后，再来看下如何进行访问次数的更新
// https://github.com/redis/redis/blob/6.2/src/evict.c#L298
uint8_t LFULogIncr(uint8_t counter) {
// 等于255，不在进行次数的更新
if (counter == 255) return 255;
// 计算一个随机数
double r = (double)rand()/RAND_MAX;
// 计算当前访问次数和初始值的差值
double baseval = counter - LFU_INIT_VAL;
if (baseval < 0) baseval = 0;
// 根据baseval和lfu_log_factor计算阈值p
double p = 1.0/(baseval*server.lfu_log_factor+1);
// 概率值小于阀值
if (r < p) counter++;
return counter;
}
如果当前访问次数小于255的时候，每次 LFULogIncr 函数会计算一个阈值 p，以及一个取值为 0 到 1 之间的随机概率值 r。如果概率 r 小于阈值 p，那么 LFULogIncr 函数才会将访问次数加 1。否则的话，LFULogIncr 函数会返回当前的访问次数，不做更新。
这样按照一定的概率增加访问频率，避免了访问次数过大，8 bits 计数器对访问次数的影响。
2、使用 LFU 算法淘汰数据
LFU 处理数据淘汰和 LRU 方式差不多，这里回顾下 LRU 处理数据淘汰的过程
1、调用 getMaxmemoryState 函数计算待释放的内存空间；
2、调用 evictionPoolPopulate 函数随机采样键值对，并插入到待淘汰集合 EvictionPoolLRU 中；
3、遍历待淘汰集合 EvictionPoolLRU，选择实际被淘汰数据，并删除。
不同的是，LFU 算法在淘汰数据时，在第二步的 evictionPoolPopulate 函数中，使用了不同的方法来计算每个待淘汰键值对的空闲时间。
LRU 中 idle 记录的是它距离上次访问的空闲时间。
LFU 中 idle 记录的是用 255 减去键值对的访问次数。也就是键值对访问次数越大，它的 idle 值就越小，反之 idle 值越大。
if (server.maxmemory_policy & MAXMEMORY_FLAG_LRU) {
idle = estimateObjectIdleTime(o);
} else if (server.maxmemory_policy & MAXMEMORY_FLAG_LFU) {
idle = 255-LFUDecrAndReturn(o);
}
freeMemoryIfNeeded 函数按照 idle 值从大到小，遍历 EvictionPoolLRU 数组，选择实际被淘汰的键值对时，它就能选出访问次数小的键值对了，也就是把访问频率低的键值对淘汰出去。
具体的源码上面 LRU 已经展示了，这里不在啰嗦了。
总结
1、Redis 中实际采用的策略是惰性删除加定期删除的组合方式；
2、组合的删除策略，其中定期删除，获取 CPU 和 内存的使用平衡，针对过期的 KEY 可能得不到及时的删除，当 KEY 被再次获取的时候，通过惰性删除再做一次过期检查，来避免业务获取到过期内容；
3、删除的时候，如果主库创建的过期键，并且过期了没有被删除，这时候从库是会读取到内容，并且是不能进行删除操作，只能由主库操作删除，不过从库会根据自己的逻辑时间判断这个过期键是否过期，从而避免读取到过期的数据；
4、当 Redis 运行内存已经超过 Redis 设置的最大内存之后，这时候就会触发内存淘汰机制来清理内存，保证 Redis 的正常运行；
5、内存淘汰机制一共 6 种淘汰方式；
6、内存淘汰机制里面用到了 LRU 和 LFU；
7、具体的淘汰过程；
1、调用 getMaxmemoryState 函数计算待释放的内存空间；
2、调用 evictionPoolPopulate 函数随机采样键值对，并插入到待淘汰集合 EvictionPoolLRU 中；
3、遍历待淘汰集合 EvictionPoolLRU，选择实际被淘汰数据，并删除。
LRU 和 LFU 不同的是，在第二步的 evictionPoolPopulate 函数中，使用了不同的方法来计算每个待淘汰键值对的空闲时间。
LRU 中 idle 记录的是它距离上次访问的空闲时间。
LFU 中 idle 记录的是用 255 减去键值对的访问次数。也就是键值对访问次数越大，它的 idle 值就越小，反之 idle 值越大。
参考
【Redis核心技术与实战】https://time.geekbang.org/column/intro/100056701
【Redis设计与实现】https://book.douban.com/subject/25900156/
【Redis 源码剖析与实战】https://time.geekbang.org/column/intro/100084301
【Redis中过期键的删除】https://boilingfrog.github.io/2022/04/02/Redis中过期键的删除/
【Redis学习笔记】https://github.com/boilingfrog/Go-POINT/tree/master/redis
![C++动态申请内存 new T()与new T[]的区别](https://www.kunjuke.com/file/css/thumb/3.jpg/280-180.jpg)