前言

上篇文章从零开始实现一个Redis（二）——缓存过期策略我们实现了Redis的缓存过期策略，来保证在Redis中被设置了过期时间的key都可以被删除，那当Redis中的数据达到了我们设置的内存最大值的时候，它又是如何来选择淘汰哪些key呢？

实现

6种缓存淘汰策略

Redis一共有6种缓存淘汰策略，查看源代码可以看到它的宏定义：

#define REDIS_MAXMEMORY_VOLATILE_LRU 0  //设置了过期时间的key按最近最少使用时间淘汰
#define REDIS_MAXMEMORY_VOLATILE_TTL 1  //设置了过期时间的key按TTL淘汰
#define REDIS_MAXMEMORY_VOLATILE_RANDOM 2  //设置了过期时间的key中随机淘汰
#define REDIS_MAXMEMORY_ALLKEYS_LRU 3  //所有的key按最近最少使用时间淘汰
#define REDIS_MAXMEMORY_ALLKEYS_RANDOM 4  //所有的key随机淘汰
#define REDIS_MAXMEMORY_NO_EVICTION 5  //不淘汰任何key
#define REDIS_DEFAULT_MAXMEMORY_POLICY REDIS_MAXMEMORY_NO_EVICTION
复制代码

可以看到默认的策略是不淘汰，除此之外，主要分为随机淘汰和LRU淘汰，这两种又分别设置在所有key中执行和在设置了过期时间的key中执行。然后最后一种是按过期时间的TTL来淘汰。

淘汰时机

Redis触发key淘汰的时机是在处理命令的时候，在真正的执行命令之前，会先判断当前已使用的内存是否已经超过了我们设置的maxmemory，如果大于，则会开始执行淘汰策略。

redis.go

//处理命令
func processCommand(client *redisClient) int {
        //lookupCommand
	...

	if server.maxMemory > 0 {
		ret := freeMemoryIfNeeded()
		if ret == redisErr {
			addReply(client, shared.oomerr)
			return redisOk
		}
	}
	call(client, 0)
	return redisOk
}
复制代码

可以看到淘汰key的处理逻辑是在主线程执行命令的过程中进行的，也就意味着它会阻塞我们正常命令的执行，当我们的内存占用达到上限，并且还在不停的写入数据的时候，会严重影响Redis的性能。

NO_EVICTION

不淘汰任何key，这个实现十分简单，直接返回oomerr错误给客户端即可
redis.go

func freeMemoryIfNeeded() int {
	memUsed := usedMemory()
	//TODO 这里需要移除AOF和slave的buffer才是真实的占用内存
	//to something

	if memUsed <= server.maxMemory {
		return redisOk
	}

	//淘汰策略不允许释放内存，返回err
	if server.maxMemoryPolicy == redisMaxMemoryNoEviction {
		return redisErr
	}
        ...
}
复制代码

Redis的默认淘汰策略就是不淘汰，那么在内存达到上限时，所有的客户端请求都会异常。所以在生产环境使用Redis，建议一定要配置淘汰策略。

RANDOM

随机淘汰也很简单，随机从字典中取一个key进行淘汰。
redis.go

func freeMemoryIfNeeded() int {

        ...
        
	memToFree := memUsed - server.maxMemory
	var memFreed uint64 = 0
	for memFreed < memToFree {

		var dt *dict
		var bestKey *robj
		var bestVal int64

		if server.maxMemoryPolicy == redisMaxMemoryAllKeysLru ||
			server.maxMemoryPolicy == redisMaxMemoryAllKeysRandom {
			dt = server.db.dict
		}else {
			dt = server.db.expires
		}
		if server.maxMemoryPolicy == redisMaxMemoryAllKeysRandom ||
			server.maxMemoryPolicy == redisMaxMemoryVolatileRandom {
			//随机淘汰
			val := dt.getRandomKey()
			if val != nil {
				bestKey = val
			}
		}
        ...
 }

复制代码

随机淘汰比较适用于在我们Redis中的数据访问比较平均，没有特别的热点数据的时候，可以使用随机的淘汰策略。

LRU（最近最少使用）

LRU淘汰就稍微复杂一点了，Redis维护了一个全局的lru时钟，在每次触发定时事件时，都会更新这个时钟：

redis.go

//redis服务端结构
type redisServer struct {
	...

	lruclock uint64
        ...
}

//上篇的redis的过期策略中的主动过期也是通过这里来触发
func serverCron() time.Duration {
	server.lruclock = getLruClock()  //更新时钟
	databasesCron()
	return time.Millisecond * time.Duration(1000/server.hz)
}
复制代码

还记得我们之前实现redisObj时，它也维护了一个lru的字段:

object.go

type robj struct {
	rtype    uint8
	encoding uint8
	lru      uint64
	refcount int
	ptr      interface{}
}

复制代码

我们在每次访问到某个key时，都会将这个lru字段更新为系统的当前时钟：

db.go

func (r *redisDb) doLookupKey(key *robj) *robj {
	entry := r.dict.dictFind(key.ptr)
	if entry != nil {
		val := entry.(*robj)
		val.lru = lruClock()
		return val
	}
	return nil
}
复制代码

也就意味着，如果某个key越长时间没有被访问到，那么他的lru值和当前系统的时钟差值会越大：

redis.go

func evictionPoolPopulate(sampleDict *dict, keyDict *dict, pool []*evictionPoolEntry) {
	//sampleCount := 16
	//if server.maxMemorySamples <= sampleCount {
	//}
	samples := sampleDict.getSomeKeys(server.maxMemorySamples)
	var o *robj
	for k,v := range samples {
		o = v.(*robj)
		key := k.(sds)
		if sampleDict != keyDict {
			//sampledict是过期字典，要重新从数据字典中拿到value
			o = keyDict.dictFind(k).(*robj)
		}
		idle := estimateObjectIdleTime(o)

		k := 0
		for k < redisEvictionPoolSize && len(pool[k].key) == 0 && pool[k].idle < idle {
			k++
		}

		if k == 0 && len(pool[redisEvictionPoolSize - 1].key) != 0 {
			//key的空闲时间比淘汰池中最短的还要短，并且淘汰池已满，跳过这个key
			continue
		}else if k < redisEvictionPoolSize && len(pool[k].key) == 0 {
			//不处理，k还没有值，可以直接插入
		}else {
			//需要在数组中插入k
			if len(pool[redisEvictionPoolSize - 1].key) == 0 {
				//未满，后移
				copy(pool[k+1:], pool[k:])
			}else {
				//满了，前移
				//free(pool[0])
				pool[0].key = ""
				k--
				copy(pool[:k-1], pool[1:k])
			}
		}
		pool[k].key = key
		pool[k].idle = idle
	}
	//free samples
	samples = nil
}
复制代码

可以看到LRU的实现并不是从所有的key中进行排序的，也是通过抽样的方式，默认的抽样个数是5个（通过maxmemory_samples控制），也就是每次会随机从字典中抽取5个键，然后按key的空闲时间在淘汰池中进行排序（数组）。将抽样数据进行LRU排序后，再将空闲时间最长的key删除并释放内存。

func freeMemoryIfNeeded() int {
	...
	memToFree := memUsed - server.maxMemory
	var memFreed uint64 = 0
	for memFreed < memToFree {
                ...
		else if server.maxMemoryPolicy == redisMaxMemoryAllKeysLru ||
			server.maxMemoryPolicy == redisMaxMemoryVolatileLru {
			//LRU淘汰
			for bestKey != nil {
				pool := server.db.evictionPool
				evictionPoolPopulate(dt, server.db.dict, pool)

				for k:=redisEvictionPoolSize - 1; k >= 0; k-- {
					//从后往前遍历，空间时间长的往短的
					if len(pool[k].key) == 0 {
						continue
					}
					val := dt.dictFind(pool[k].key).(*robj)
					//释放pool对象
					//sdsfree(pool[k].key)
					//模拟
					pool[k].key = ""
					//链表元素的右边全部左移，如果是最后一个元素不需要移动
					if k < redisEvictionPoolSize - 1 {
						copy(pool[k:], pool[k+1:])
					}
					//左移后最后一个元素重新初始化
					pool[redisEvictionPoolSize - 1].key = ""
					pool[redisEvictionPoolSize - 1].idle = 0

					if val != nil {
						bestKey = val
						break
					}
				}
			}
		}

复制代码

LRU淘汰适合我们有明显的热点数据，大量的请求是访问少量的热点数据的情况下，适合使用这种策略

TTL

按照过期时间淘汰，它也是进行抽样淘汰，从过期字典中随机抽取maxmemory_samples个key，淘汰最快要过期的数据。

redis.go

else if server.maxMemoryPolicy == redisMaxMemoryVolatileTtl {
			//TTL淘汰
			for k:=0; k < server.maxMemorySamples; k++ {
				key := dt.getRandomKey()
				val := dt.dictFind(key.ptr).(int64)

				if bestKey != nil || val < bestVal {
					bestKey = key
					bestVal = val
				}
			}
		}
复制代码

在根据淘汰策略选择出key之后，则会进行删除操作，并释放内存。如此循环处理，直到可用内存小于最大内存为止：

redis.go

func freeMemoryIfNeeded() int {
	...
	for memFreed < memToFree {
                ...
		if bestKey != nil {
			//free memory
			delta := usedMemory()
			server.db.dbDelete(bestKey)
			delta -= usedMemory()
			memFreed += delta
		}
	}
	return redisOk
}
复制代码