Redis资料淘汰演算法
众所周知，Redis的所有资料都储存在内存中，但是内存是一种有限的资源，所以为了防止Redis无限制的使用内存，在启动Redis时可以通过配置项 maxmemory 来指定其最大能使用的内存容量。例如可以通过以下配置来设定Redis最大能使用 1G 内存：
maxmemory 1G
当Redis使用的内存超过配置的 maxmemory 时，便会触发资料淘汰策略。Redis提供了多种资料淘汰的策略，如下：
volatile-lru: 最近最少使用算法，从设定了过期时间的键中选择空转时间最长的键值对清除掉volatile-lfu: 最近最不经常使用算法，从设定了过期时间的键中选择某段时间之内使用频次最小的键值对清除掉volatile-ttl: 从设定了过期时间的键中选择过期时间最早的键值对清除volatile-random: 从设定了过期时间的键中，随机选择键进行清除allkeys-lru: 最近最少使用算法，从所有的键中选择空转时间最长的键值对清除allkeys-lfu: 最近最不经常使用算法，从所有的键中选择某段时间之内使用频次最少的键值对清除allkeys-random: 所有的键中，随机选择键进行删除noeviction: 不做任何的清理工作，在redis的内存超过限制之后，所有的写入操作都会返回错误；但是读操作都能正常的进行可以在启动Redis时，通过配置项 maxmemory_policy 来指定要使用的资料淘汰策略。例如要使用 volatile-lru 策略可以通过以下配置来指定：
maxmemory_policy volatile-lru
LRU算法
LRU是 Least Recently Used 的缩写，即最近最少使用，很多快取系统都使用此算法作为淘汰策略。最简单的实现方式就是把所有快取通过一个连结串列连线起来，新建立的快取新增到连结串列的头部，如果有快取被访问了，就把快取移动到连结串列的头部。由于被访问的快取会移动到连结串列的头部，所以没有被访问的快取会随着时间的推移移动的连结串列的尾部，淘汰资料时只需要从连结串列的尾部开始即可。下图展示了这个过程：
Redis的LRU算法
Redis使用了结构体 robj 来储存快取物件，而 robj 结构有个名为 lru 的字段，用于记录快取物件最后被访问的时间，Redis就是以 lru 字段的值作为淘汰依据。robj 结构如下：typedef struct redisObject {
...
unsigned lru:24;
...
} robj;
当快取物件被访问时，便会更新此字段的值。程式码如下：
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 (server.rdb_child_pid == -1 &&
server.aof_child_pid == -1 &&
!(flags & LOOKUP_NOTOUCH))
{
if (server.maxmemory_policy & MAXMEMORY_FLAG_LFU) {
updateLFU(val);
} else {
val->lru = LRU_CLOCK(); // 更新lru字段的值
}
}
return val;
} else {
return NULL;
}
}
lookupKey() 函式用于查询key对应的快取物件，所以当快取物件被访问时便会呼叫此函式。
Redis资料淘汰
接下来我们分析一下当Redis内存使用超过配置的最大内存使用限制时的处理方式。Redis在处理每一个命令时都会检查内存的使用是否超过了限制的最大值，处理命令是通过 processCommand() 函式进行的，检查内存使用情况的程式码如下：
int processCommand(client *c) {
...
if (server.maxmemory && !server.lua_timedout) {
int out_of_memory = freeMemoryIfNeededAndSafe() == C_ERR;
if (server.current_client == NULL) return C_ERR;
if (out_of_memory &&
(c->cmd->flags & CMD_DENYOOM ||
(c->flags & CLIENT_MULTI && c->cmd->proc != execCommand))) {
flagTransaction(c);
addReply(c, shared.oomerr);
return C_OK;
}
}
...
}
检查内存的使用情况主要通过 freeMemoryIfNeededAndSafe() 函式进行，而 freeMemoryIfNeededAndSafe() 函式最终会呼叫 freeMemoryIfNeeded() 函式进行处理，由于 freeMemoryIfNeeded() 函式比较庞大，所以我们分段来进行分析：
int freeMemoryIfNeeded(void) {
...
size_t mem_reported, mem_tofree, mem_freed;
mstime_t latency, eviction_latency;
long long delta;
int slaves = listLength(server.slaves);
...
if (getMaxmemoryState(&mem_reported,NULL,&mem_tofree,NULL) == C_OK)
return C_OK;
mem_freed = 0;
if (server.maxmemory_policy == MAXMEMORY_NO_EVICTION)
goto cant_free;
freeMemoryIfNeeded() 函式首先会呼叫 getMaxmemoryState() 函式来获取Redis的内存使用情况，如果 getMaxmemoryState() 函式返回 C_OK，表示内存使用总量还没有超出限制，直接返回 C_OK 就可以了。如果 getMaxmemoryState() 函式不是返回 C_OK，表示内存使用总量已经超出限制，需要进行资料淘汰，需要淘汰资料的大小通过 mem_tofree 引数返回。
当然，如果配置的淘汰策略为 noeviction，表示不能进行资料淘汰，所以需要返回 C_ERR 表示有错误。
接着分析剩余的程式码片段：
latencyStartMonitor(latency);
while (mem_freed int j, k, i, keys_freed = 0;
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;
for (i = 0; i db = server.db+i;
dict = (server.maxmemory_policy & MAXMEMORY_FLAG_ALLKEYS) ?
db->dict : db->expires;
if ((keys = dictSize(dict)) != 0) {
evictionPoolPopulate(i, dict, db->dict, pool);
total_keys += keys;
}
}
if (!total_keys) break; /* No keys to evict. */
for (k = EVPOOL_SIZE-1; k >= 0; k--) {
if (pool[k].key == NULL) continue;
bestdbid = pool[k].dbid;
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);
}
if (pool[k].key != pool[k].cached)
sdsfree(pool[k].key);
pool[k].key = NULL;
pool[k].idle = 0;
if (de) {
bestkey = dictGetKey(de);
break;
} else {
/* Ghost... Iterate again. */
}
}
}
}
如果内存使用总量超出限制，并且配置了淘汰策略，那么就开始资料淘汰过程。在上面的程式码中，mem_tofree 变量表示要淘汰的资料总量，而 mem_freed 变量表示已经淘汰的资料总量。所以在 while 循环中的条件是 mem_freed
前面介绍过，Redis的淘汰策略有很多中，所以进行资料淘汰时需要根据配置的策略进行。如果配置的淘汰策略是 LRU/LFU/TTL 的话，那么就进入 if 程式码块。在 if 程式码块里，首先呼叫 evictionPoolPopulate() 函式选择一些快取物件样本放置到 EvictionPoolLRU 阵列中。evictionPoolPopulate() 函式后面会进行分析，现在只需要知道 evictionPoolPopulate() 函式是选取一些快取物件样本就可以了。
获取到快取物件样本后，还需要从样本中获取最合适的快取物件进行淘汰，因为在选择样本时会把最合适的快取物件放置在 EvictionPoolLRU 阵列的尾部，所以只需要从 EvictionPoolLRU 阵列的尾部开始查询一个不为空的快取物件即可。
else if (server.maxmemory_policy == MAXMEMORY_ALLKEYS_RANDOM ||
server.maxmemory_policy == MAXMEMORY_VOLATILE_RANDOM)
{
for (i = 0; i j = (++next_db) % server.dbnum;
db = server.db+j;
dict = (server.maxmemory_policy == MAXMEMORY_ALLKEYS_RANDOM) ?
db->dict : db->expires;
if (dictSize(dict) != 0) {
de = dictGetRandomKey(dict);
bestkey = dictGetKey(de);
bestdbid = j;
break;
}
}
}
如果使用随机淘汰策略，那么就进入 else if 程式码块，这部分程式码的逻辑很简单，如果配置的淘汰策略是 volatile-random，那么就从有过期时间的快取物件中随机获取，否则就从所有的快取物件中随机获取。
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);
latencyEndMonitor(eviction_latency);
latencyAddSampleIfNeeded("eviction-del",eviction_latency);
latencyRemoveNestedEvent(latency,eviction_latency);
delta -= (long long) zmalloc_used_memory();
mem_freed += delta;
server.stat_evictedkeys++;
notifyKeyspaceEvent(NOTIFY_EVICTED, "evicted",
keyobj, db->id);
decrRefCount(keyobj);
keys_freed++;
if (slaves) flushSlavesOutputBuffers();
if (server.lazyfree_lazy_eviction && !(keys_freed % 16)) {
if (getMaxmemoryState(NULL,NULL,NULL,NULL) == C_OK) {
mem_freed = mem_tofree;
}
}
}
如果找到要淘汰的快取物件，那么就开始释放快取物件所占用的内存空间。除了需要释放快取物件占用的内存空间外，还需要进行一些其他的操作，比如把淘汰的快取物件同步到从服务器和把淘汰的快取物件追加到 AOF档案 中等。
当条件 mem_freed
淘汰资料样本采集
前面说了，当使用非随机淘汰策略时需要进行资料取样（volatile-lru/volatile-lfu/volatile-ttl/allkeys-lru/allkeys-lfu），资料取样通过 evictionPoolPopulate() 函式进行，由于此函式比较庞大，所以对程式码分段分析：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);
evictionPoolPopulate() 函式首先呼叫 dictGetSomeKeys() 函式从快取物件集合中获取一些样本，并储存在 samples 阵列中。
for (j = 0; j unsigned long long idle;
sds key;
robj *o;
dictEntry *de;
de = samples[j];
key = dictGetKey(de);
if (server.maxmemory_policy != MAXMEMORY_VOLATILE_TTL) {
if (sampledict != keydict) de = dictFind(keydict, key);
o = dictGetVal(de);
}
if (server.maxmemory_policy & MAXMEMORY_FLAG_LRU) {
idle = estimateObjectIdleTime(o);
} else if (server.maxmemory_policy & MAXMEMORY_FLAG_LFU) {
idle = 255-LFUDecrAndReturn(o);
} else if (server.maxmemory_policy == MAXMEMORY_VOLATILE_TTL) {
idle = ULLONG_MAX - (long)dictGetVal(de);
} else {
serverPanic("Unknown eviction policy in evictionPoolPopulate()");
}
上面的程式码主要是获取样本快取物件的排序权值 idel，如果使用 LRU淘汰算法，那么就呼叫 estimateObjectIdleTime() 函式获取排序权值，estimateObjectIdleTime() 函式用于获取快取物件有多长时间没有被访问。排序按照 idle 的值升序排序，就是说 idle 的值越大，就排到越后。
k = 0;
while (k pool[k].key &&
pool[k].idle if (k == 0 && pool[EVPOOL_SIZE-1].key != NULL) {
continue;
} else if (k } else {
if (pool[EVPOOL_SIZE-1].key == NULL) {
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 {
k--;
sds cached = pool[0].cached;
if (pool[0].key != pool[0].cached) sdsfree(pool[0].key);
memmove(pool,pool+1,sizeof(pool[0])*k);
pool[k].cached = cached;
}
}
int klen = sdslen(key);
if (klen > EVPOOL_CACHED_SDS_SIZE) {
pool[k].key = sdsdup(key);
} else {
memcpy(pool[k].cached,key,klen+1);
sdssetlen(pool[k].cached,klen);
pool[k].key = pool[k].cached;
}
pool[k].idle = idle;
pool[k].dbid = dbid;
}
}
上面这段程式码的作用是：根据 idle 的值找到当前快取物件所在 EvictionPoolLRU 阵列的位置，然后把快取物件储存到 EvictionPoolLRU 阵列中。以下插图解释了资料取样的过程：
所以 EvictionPoolLRU 阵列的最后一个元素便是最优的淘汰快取物件。
从上面的分析可知，淘汰资料时只是从样本中找到最优的淘汰快取物件，并不是从所有快取物件集合中查询。由于前面介绍的 LRU算法 需要维护一个LRU连结串列，而维护一个LRU连结串列的成本比较大，所以Redis才出此下策。
