近似 LRU 算法具体如何执行？
近似 LRU 算法的执行可以分成三大步骤，分别是
- 判断当前内存使用情况
- 更新待淘汰的候选键值对集合
- 选择被淘汰的键值对并删除
下面我们就依次来看下。
判断当前内存使用情况
- 首先，freeMemoryIfNeeded 函数会调用 getMaxmemoryState 函数，评估当前的内存使用情况。getMaxmemoryState 函数是在 evict.c 文件中实现的，它会判断当前 Redis server 使用的内存容量是否超过了 maxmemory 配置的值。
- 如果当前内存使用量没有超过 maxmemory，那么，getMaxmemoryState 函数会返回 C_OK，紧接着，freeMemoryIfNeeded 函数也会直接返回了。
int freeMemoryIfNeeded(void) { ... if (getMaxmemoryState(&mem_reported,NULL,&mem_tofree,NULL) == C_OK) return C_OK; ... }
这里，你需要注意的是，getMaxmemoryState 函数在评估当前内存使用情况的时候，如果发现已用内存超出了 maxmemory，它就会计算需要释放的内存量。这个释放的内存大小等于已使用的内存量减去 maxmemory。不过，已使用的内存量并不包括用于主从复制的复制缓冲区大小，这是 getMaxmemoryState 函数，通过调用 freeMemoryGetNotCountedMemory 函数来计算的。
我把 getMaxmemoryState 函数的基本执行逻辑代码放在这里，你可以看下。
/* Get the memory status from the point of view of the maxmemory directive: * if the memory used is under the maxmemory setting then C_OK is returned. * Otherwise, if we are over the memory limit, the function returns * C_ERR. * * The function may return additional info via reference, only if the * pointers to the respective arguments is not NULL. Certain fields are * populated only when C_ERR is returned: * * 'total' total amount of bytes used. * (Populated both for C_ERR and C_OK) * * 'logical' the amount of memory used minus the slaves/AOF buffers. * (Populated when C_ERR is returned) * * 'tofree' the amount of memory that should be released * in order to return back into the memory limits. * (Populated when C_ERR is returned) * * 'level' this usually ranges from 0 to 1, and reports the amount of * memory currently used. May be > 1 if we are over the memory * limit. * (Populated both for C_ERR and C_OK) */ int getMaxmemoryState(size_t *total, size_t *logical, size_t *tofree, float *level) { size_t mem_reported, mem_used, mem_tofree; /* Check if we are over the memory usage limit. If we are not, no need * to subtract the slaves output buffers. We can just return ASAP. */ // 计算已使用的内存量 mem_reported = zmalloc_used_memory(); if (total) *total = mem_reported; /* We may return ASAP if there is no need to compute the level. */ int return_ok_asap = !server.maxmemory || mem_reported <= server.maxmemory; if (return_ok_asap && !level) return C_OK; /* Remove the size of slaves output buffers and AOF buffer from the * count of used memory. */ // 将用于主从复制的复制缓冲区大小和AOF缓冲区大小从已使用内存量中扣除 mem_used = mem_reported; size_t overhead = freeMemoryGetNotCountedMemory(); mem_used = (mem_used > overhead) ? mem_used-overhead : 0; /* Compute the ratio of memory usage. */ // 计算内存使用率。 if (level) { if (!server.maxmemory) { *level = 0; } else { *level = (float)mem_used / (float)server.maxmemory; } } if (return_ok_asap) return C_OK; /* Check if we are still over the memory limit. */ // 检查我们是否仍然超过内存限制。 if (mem_used <= server.maxmemory) return C_OK; // 计算需要释放的内存量 /* Compute how much memory we need to free. */ mem_tofree = mem_used - server.maxmemory; if (logical) *logical = mem_used; if (tofree) *tofree = mem_tofree; return C_ERR; }
而如果当前 server 使用的内存量，的确已经超出 maxmemory 的上限了，那么 freeMemoryIfNeeded 函数就会执行一个 while 循环，来淘汰数据释放内存。
其实，为了淘汰数据，Redis 定义了一个数组 EvictionPoolLRU，用来保存待淘汰的候选键值对。这个数组的元素类型是 evictionPoolEntry 结构体，该结构体保存了待淘汰键值对的空闲时间 idle、对应的 key 等信息。以下代码展示了 EvictionPoolLRU 数组和 evictionPoolEntry 结构体，它们都是在 evict.c 文件中定义的。
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;
这样，Redis server 在执行 initSever 函数进行初始化时，会调用 evictionPoolAlloc 函数（在 evict.c 文件中）为 EvictionPoolLRU 数组分配内存空间，该数组的大小由宏定义 EVPOOL_SIZE（在 evict.c 文件中）决定，默认是 16 个元素，也就是可以保存 16 个待淘汰的候选键值对。
#define EVPOOL_SIZE 16 /* Create a new eviction pool. */ void evictionPoolAlloc(void) { struct evictionPoolEntry *ep; int j; ep = zmalloc(sizeof(*ep)*EVPOOL_SIZE); for (j = 0; j < EVPOOL_SIZE; j++) { ep[j].idle = 0; ep[j].key = NULL; ep[j].cached = sdsnewlen(NULL,EVPOOL_CACHED_SDS_SIZE); ep[j].dbid = 0; } EvictionPoolLRU = ep; }
那么，freeMemoryIfNeeded 函数在淘汰数据的循环流程中，就会更新这个待淘汰的候选键值对集合，也就是 EvictionPoolLRU 数组。下面我就来给你具体介绍一下。
更新待淘汰的候选键值对集合
首先，freeMemoryIfNeeded 函数会调用 evictionPoolPopulate 函数（在 evict.c 文件中），而 evictionPoolPopulate 函数会先调用 dictGetSomeKeys 函数（在 dict.c 文件中），从待采样的哈希表中随机获取一定数量的 key。不过，这里还有两个地方你需要注意下。
第一点，dictGetSomeKeys 函数采样的哈希表，是由 maxmemory_policy 配置项来决定的。如果 maxmemory_policy 配置的是 allkeys_lru，那么待采样哈希表就是 Redis server 的全局哈希表，也就是在所有键值对中进行采样；否则，待采样哈希表就是保存着设置了过期时间的 key 的哈希表。
以下代码是 freeMemoryIfNeeded 函数中对 evictionPoolPopulate 函数的调用过程，你可以看下。
/* We don't want to make local-db choices when expiring keys, * so to start populate the eviction pool sampling keys from * every DB. */ for (i = 0; i < server.dbnum; i++) { // 对Redis server上的每一个数据库都执行 db = server.db+i; // 根据淘汰策略，决定使用全局哈希表还是设置了过期时间的key的哈希表 dict = (server.maxmemory_policy & MAXMEMORY_FLAG_ALLKEYS) ? db->dict : db->expires; // 将选择的哈希表dict传入evictionPoolPopulate函数，同时将全局哈希表也传给evictionPoolPopulate函数 if ((keys = dictSize(dict)) != 0) { evictionPoolPopulate(i, dict, db->dict, pool); total_keys += keys; } }
第二点，dictGetSomeKeys 函数采样的 key 的数量，是由 redis.conf 中的配置项 maxmemory-samples 决定的，该配置项的默认值是 5。下面代码就展示了 evictionPoolPopulate 函数对 dictGetSomeKeys 函数的调用：
void evictionPoolPopulate(int dbid, dict *sampledict, dict *keydict, struct evictionPoolEntry *pool) { ... dictEntry *samples[server.maxmemory_samples]; //采样后的集合，大小为maxmemory_samples //将待采样的哈希表sampledict、采样后的集合samples、以及采样数量maxmemory_samples，作为参数传给dictGetSomeKeys count = dictGetSomeKeys(sampledict,samples,server.maxmemory_samples); ... }
如此一来，dictGetSomeKeys 函数就能返回采样的键值对集合了。然后，evictionPoolPopulate 函数会根据实际采样到的键值对数量 count，执行一个循环。
for (j = 0; j < count; j++) { ... if (server.maxmemory_policy & MAXMEMORY_FLAG_LRU) { idle = estimateObjectIdleTime(o); } ...
紧接着，evictionPoolPopulate 函数会遍历待淘汰的候选键值对集合，也就是 EvictionPoolLRU 数组。在遍历过程中，它会尝试把采样的每一个键值对插入 EvictionPoolLRU 数组，这主要取决于以下两个条件之一：
- 一是，它能在数组中找到一个尚未插入键值对的空位；
- 二是，它能在数组中找到一个空闲时间小于采样键值对空闲时间的键值对
这两个条件有一个成立的话，evictionPoolPopulate 函数就可以把采样键值对插入 EvictionPoolLRU 数组。等所有采样键值对都处理完后，evictionPoolPopulate 函数就完成对待淘汰候选键值对集合的更新了。
接下来，freeMemoryIfNeeded 函数，就可以开始选择最终被淘汰的键值对了。
选择被淘汰的键值对并删除
因为 evictionPoolPopulate 函数已经更新了 EvictionPoolLRU 数组，而且这个数组里面的 key，是按照空闲时间从小到大排好序了。所以，freeMemoryIfNeeded 函数会遍历一次 EvictionPoolLRU 数组，从数组的最后一个 key 开始选择，如果选到的 key 不是空值，那么就把它作为最终淘汰的 key。
这个过程的基本执行逻辑如下所示：
// 从数组最后一个key开始查找 /* Go backward from best to worst element to evict. */ for (k = EVPOOL_SIZE-1; k >= 0; k--) { // 当前key为空值，则查找下一个key if (pool[k].key == NULL) continue; bestdbid = pool[k].dbid; // 从全局哈希表或是expire哈希表中，获取当前key对应的键值对；并将当前key从EvictionPoolLRU数组删除 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. */ 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 { //否则，继续查找下个key /* Ghost... Iterate again. */ } }
最后，一旦选到了被淘汰的 key，freeMemoryIfNeeded 函数就会根据 Redis server 的惰性删除配置，来执行同步删除或异步删除，如下所示：
if (bestkey) { db = server.db+bestdbid; robj *keyobj = createStringObject(bestkey,sdslen(bestkey)); //将删除key的信息传递给从库和AOF文件 propagateExpire(db,keyobj,server.lazyfree_lazy_eviction); //如果配置了惰性删除，则进行异步删除 if (server.lazyfree_lazy_eviction) dbAsyncDelete(db,keyobj); else //否则进行同步删除 dbSyncDelete(db,keyobj); }
好了，到这里，freeMemoryIfNeeded 函数就淘汰了一个 key。而如果此时，释放的内存空间还不够，也就是说没有达到我前面介绍的待释放空间，那么 freeMemoryIfNeeded 函数还会重复执行前面所说的更新待淘汰候选键值对集合、选择最终淘汰 key 的过程，直到满足待释放空间的大小要求。
下图就展示了 freeMemoryIfNeeded 函数涉及的基本流程，你可以再来整体回顾下。
其实，从刚才介绍的内容中，你就可以看到，近似 LRU 算法并没有使用耗时耗空间的链表，而是使用了固定大小的待淘汰数据集合，每次随机选择一些 key 加入待淘汰数据集合中。最后，再按照待淘汰集合中 key 的空闲时间长度，删除空闲时间最长的 key。这样一来，Redis 就近似实现了 LRU 算法的效果了。
小结
今天这节课我给你介绍了 Redis 中，是如何实现 LRU 算法来进行缓存数据替换的。其中，我们根据 LRU 算法的基本原理，可以发现如果严格按照原理来实现 LRU 算法，那么开发的系统就需要用额外的内存空间来保存 LRU 链表，而且系统运行时也会受到 LRU 链表操作的开销影响。
而对于 Redis 来说，内存资源和性能都很重要，所以 Redis 实现了近似 LRU 算法。而为了实现近似 LRU 算法，Redis 首先是设置了全局 LRU 时钟，并在键值对创建时获取全局 LRU 时钟值作为访问时间戳，以及在每次访问时获取全局 LRU 时钟值，更新访问时间戳。
然后，当 Redis 每处理一个命令时，都会调用 freeMemoryIfNeeded 函数来判断是否需要释放内存。如果已使用内存超出了 maxmemory，那么，近似 LRU 算法就会随机选择一些键值对，组成待淘汰候选集合，并根据它们的访问时间戳，选出最旧的数据，将其淘汰。
Redis 计算实例内存时，不会把「主从复制」的缓冲区计算在内，也就是说不管一个实例后面挂了多少个从库，主库不会把主从复制所需的「缓冲区」内存，计算到实例内存中，即这部分内存增加，不会对数据淘汰产生影响。
每日一问：为什么键值对的 LRU 时钟值，不是直接通过调用 getLRUClock 函数来获取？
本质上是为了性能。Redis 这种对性能要求极高的数据库，在系统调用上的优化也做到了极致。
获取机器时钟本质上也是一个「系统调用」，对于 Redis 这种动不动每秒上万的 QPS，如果每次都触发一次系统调用，这么频繁的操作也是一笔不小的开销。
所以，Redis 用一个定时任务（serverCron 函数），以固定频率触发系统调用获取机器时钟，然后把机器时钟挂到 server 的全局变量下，这相当于维护了一个「本地缓存」，当需要获取时钟时，直接从全局变量获取即可，节省了大量的系统调用开销。
版本变动
6.2的版本中，增加了maxmemory-eviction-tenacity配置，目的是控制大量淘汰内存空间阻塞客户端的时间。
6.0版本evict.c的结构
6.2版本evict.c的结构
6.2的版本中没有了freeMemoryIfNeeded 和 freeMemoryIfNeededAndSafe函数，而是被performEvictions替代，然后在执行内存释放期间，耗时统计，超过限制时间，新增时间事件，然后结束循环，流程继续往下执行。
/* After some time, exit the loop early - even if memory limit * hasn't been reached. If we suddenly need to free a lot of * memory, don't want to spend too much time here. */ // 如果执行释放空间超过限制时间，则添加一个时间事件，时间事件中继续释放内存 if (elapsedUs(evictionTimer) > eviction_time_limit_us) { // We still need to free memory - start eviction timer proc if (!isEvictionProcRunning) { isEvictionProcRunning = 1; // 新增时间事件 aeCreateTimeEvent(server.el, 0, evictionTimeProc, NULL, NULL); } break; }
// 如果一次时间事件没结束，则该时间事件不结束，等待一段时间，继续执行 // 如果返回值不是AE_NOMORE，则继续把当前事件间隔retval毫秒后继续执行 if (retval != AE_NOMORE) { te->when = now + retval * 1000; } else { te->id = AE_DELETED_EVENT_ID; }
