缓存淘汰算法与 python 中 lru_cache 装饰器的实现

2019-04-23 14:12:47   最后更新: 2019-04-23 14:12:47   访问数量:199




此前的文章中,我们介绍过常见两种缓存架构 -- 穿透型缓存与旁路型缓存

常见缓存架构 -- 穿透型缓存与旁路型缓存

 

穿透型缓存与旁路型缓存架构的主要区别在于当缓存中不存在被访问数据时的处理方式

但是,缓存的容量是有限的,随着时间的发展,无论是使用穿透型缓存还是旁路型缓存的架构,最终都会面临缓存被占满的情况,那么此时,为了缓存数据的实时性,我们需要淘汰一些旧的、访问率低的数据来增加空闲空间以便缓存的继续使用

应该如何选择被淘汰的数据呢?这就是缓存淘汰算法的目标

 

最常见的缓存淘汰算法有以下这些:

 

先进先出 -- FIFO

FIFO 即 First In First Out 的缩写,这可以说是实现起来最简单的一种算法了

缓存维护一个队列,总是在队首插入数据,当缓存区满时,则删除队尾数据

 

 

这个算法的优点就在于实现简单,但缺点也是显而易见的:

  1. 顺序写入与读出往往不符合我们的业务场景,虽然我们可以通过叠加其他数据结构让队列可以支持随机访问来规避这个缺点
  2. 先写入的数据并不一定比后写入的数据重要性低,这是该算法的主要问题

 

最不经常使用算法 -- LFU

LFU 是 least frequently used 的缩写

算法思想是淘汰那些被访问次数最少的数据

其实现方式也非常简单,只需要维护一个队列,存储数据与被访问次数的对应关系

每次数据被访问时,增加其对应的访问次数值,并将该节点在链表中向队首移动,直到整个队列从对少到队尾仍然保持按访问次数递减存储

当需要执行淘汰算法时,只要淘汰队尾的部分数据即可

 

 

这个算法相比于先进先出算法,他兼顾了数据的访问次数,从而避免热数据先于冷数据被淘汰

但是,这个算法仍然存在一定的问题,那就是一旦某个数据在短时间被大量访问,此后即便很长时间没有任何访问,该数据仍然凭借其巨大的访问次数数值而不被淘汰

 

最近最少使用算法 -- LRU

LRU 是 Least Recently Used 的缩写

这是目前实践中最为广泛使用的一种缓存淘汰算法,他的算法思想与 LFU 非常相似,但他去除了访问次数数值,在队列中采用访问则上升的策略,从而规避了上面提到的访问次数数值过大造成的影响

由于该算法的广泛使用性,我们下文将以 python 中十分常用的方法执行参数与结果的缓存 -- functools.lru_cache,来详细介绍一下该算法

 

 

最近最常使用算法 -- MRU

与 LRU 相对,MRU 是 Most recently used 的缩写

这个算法与 LRU 恰好相反,MRU 算法优先移除最近被使用过的数据,它用来处理越久没有被使用的数据越容易被访问到的情况

 

python 标准库中的 functools.lru_cache 装饰器实现了一个 LRU 算法的缓存,用来缓存方法所有参数与返回值的对应关系,用来提升一个方法频繁用相同参数调用场景下的性能

关于 python 的闭包与装饰器,参考此前的文章:

python 中的闭包

python 中的装饰器及其原理

 

简化后源码

下面是抽取简化后的 python 标准库 functools.lru_cache 源码:

def make_key(args, kwds): """ 通过方法参数获取缓存的 key :param args: 方法原始参数 :param kwds: 方法键值对参数 :return: 计算后 key 的 hash 值 """ key = args if kwds: key += object() for item in kwds.items(): key += item elif len(key) == 1 and type(key[0]) in {int, str, frozenset, type(None)}: return key[0] return hash(key) def lru_cache(maxsize=128): def decorating_function(user_function): PREV, NEXT, KEY, RESULT = 0, 1, 2, 3 cache = {} # 缓存结构,存储 key 与缓存数据的映射 hits = misses = 0 # 缓存命中、丢失统计 full = False # 缓冲区是否已满标记 cache_len = cache.__len__ # 缓冲区大小 root = [] root[:] = [root, root, None, None] # 队列头结点为空 def lru_cache_wrapper(*args, **kwds): nonlocal root, hits, misses, full key = make_key(args, kwds) with RLock(): # 线程安全锁 link = cache.get(key) """ 缓存命中,移动命中节点,直接返回预设结果 """ if link is not None: """ 从链表中移除命中节点 """ link_prev, link_next, _key, result = link link_prev[NEXT] = link_next link_next[PREV] = link_prev """ 将命中节点移动到队尾 """ last = root[PREV] last[NEXT] = root[PREV] = link link[PREV] = last link[NEXT] = root hits += 1 return result """ 缓存未命中,调用方法获取返回,创建节点,淘汰算法 """ result = user_function(*args, **kwds) with RLock(): if key in cache: pass elif full: """ 缓冲区已满,淘汰队首元素,删除缓冲区对应元素 """ oldroot = root oldroot[KEY] = key oldroot[RESULT] = result root = oldroot[NEXT] oldkey = root[KEY] root[KEY] = root[RESULT] = None del cache[oldkey] cache[key] = oldroot else: """ 缓冲区未满,直接创建节点,插入数据 """ last = root[PREV] link = [last, root, key, result] last[NEXT] = root[PREV] = cache[key] = link full = (cache_len() >= maxsize) misses += 1 return result return lru_cache_wrapper return decorating_function

 

 

算法流程

上述代码的实现逻辑非常清晰,实现方式也非常巧妙

他利用字典实现了一个缓冲区,同时创建了一个环形双向链表,而链表中每个节点都是一个 list,list 中的四个元素分别代表前驱的引用、后继引用、key、函数返回值

通过缓冲区与环形双向链表的同步操作完成 LRU 算法的实现

  1. 【初始状态】 初始状态下,cache 字典为空,环形双向链表中只有 key、result 均为 None 的 root 节点
  2. 【缓存命中】 当插入元素命中缓存,则在链表中移除该节点,并将该节点插入 root 之前,实现最近使用数据在链表中位置的提升
  3. 【缓存未命中且队列未满】 当插入元素未命中缓存,则创建该元素的节点,并直接在环形双线链表的 root 之前插入节点,cache[key] 赋值为插入节点
  4. 【缓存未命中且队列已满】 此时需要触发 LRU 缓存淘汰算法,此时将 root 的 key 与 result 分别赋值为待插入节点对应的值,向后移动 root,将 root 的 key、result 分别赋值为 None,从而实现 root 后相邻节点的清除,cache[key] 赋值为插入节点,删除 cache 中被移除节点

 

下图展示了缓冲命中与缓存淘汰两种场景下的算法执行过程:

 

 

此前我们曾经提到,由于 python 没有尾递归优化,递归执行算法效率是很低的

在此前的文章中,针对这一情况,我们自行实现了简易的尾递归优化

经典动态规划问题 -- 青蛙上台阶与 python 的递归优化

 

斐波那契数列的递归生成

让我们加上此前文章中的 clock 装饰器,再次看看递归生成斐波那契数列的程序

def clock(func): @functools.wraps(func) def clocked(*args, **kwargs): t0 = time.time() result = func(*args, **kwargs) elapsed = time.time() - t0 name = func.__name__ arg_lst = [] if args: arg_lst.append(', '.join(repr(arg) for arg in args)) if kwargs: pairs = ['%s=%r' % (k, w) for k, w in sorted(kwargs.items())] arg_lst.append(', '.join(pairs)) arg_str = ', '.join(arg_lst) print('[%0.8fs] %s(%s) -> %r ' % (elapsed, name, arg_str, result)) return result return clocked @lru_cache() @clock def fibonacci(n): if n < 2: return n return fibonacci(n - 2) + fibonacci(n - 1) if __name__ == '__main__': print(fibonacci(6))

 

 

执行程序,打印出了:

[0.00000000s] fibonacci(0) -> 0 

[0.00000000s] fibonacci(1) -> 1 

[0.00000000s] fibonacci(2) -> 1 

[0.00000000s] fibonacci(1) -> 1 

[0.00000000s] fibonacci(0) -> 0 

[0.00000000s] fibonacci(1) -> 1 

[0.00000000s] fibonacci(2) -> 1 

[0.00000000s] fibonacci(3) -> 2 

[0.00050116s] fibonacci(4) -> 3 

[0.00000000s] fibonacci(1) -> 1 

[0.00000000s] fibonacci(0) -> 0 

[0.00000000s] fibonacci(1) -> 1 

[0.00000000s] fibonacci(2) -> 1 

[0.00000000s] fibonacci(3) -> 2 

[0.00000000s] fibonacci(0) -> 0 

[0.00000000s] fibonacci(1) -> 1 

[0.00000000s] fibonacci(2) -> 1 

[0.00000000s] fibonacci(1) -> 1 

[0.00000000s] fibonacci(0) -> 0 

[0.00000000s] fibonacci(1) -> 1 

[0.00000000s] fibonacci(2) -> 1 

[0.00000000s] fibonacci(3) -> 2 

[0.00000000s] fibonacci(4) -> 3 

[0.00000000s] fibonacci(5) -> 5 

[0.00050116s] fibonacci(6) -> 8 

8

 

可以看到,有很多重复参数的调用在消耗着程序性能,随着数列规模的增长,重复调用的次数将成倍增长

一个有效的优化条件就是将这些重复调用的结果缓存起来,再次调用时直接返回即可,这正是 lru_cache 的用途

 

使用 lru_cache 优化斐波那契数列的生成

import functools import time from _thread import RLock def make_key(args, kwds): """ 通过方法参数获取缓存的 key :param args: 方法原始参数 :param kwds: 方法键值对参数 :return: 计算后 key 的 hash 值 """ key = args if kwds: key += object() for item in kwds.items(): key += item elif len(key) == 1 and type(key[0]) in {int, str, frozenset, type(None)}: return key[0] return hash(key) def lru_cache(maxsize=128): def decorating_function(user_function): PREV, NEXT, KEY, RESULT = 0, 1, 2, 3 cache = {} # 缓存结构,存储 key 与缓存数据的映射 hits = misses = 0 # 缓存命中、丢失统计 full = False # 缓冲区是否已满标记 cache_len = cache.__len__ # 缓冲区大小 root = [] root[:] = [root, root, None, None] # 队列头结点为空 def lru_cache_wrapper(*args, **kwds): nonlocal root, hits, misses, full key = make_key(args, kwds) with RLock(): # 线程安全锁 link = cache.get(key) """ 缓存命中,移动命中节点,直接返回预设结果 """ if link is not None: """ 从链表中移除命中节点 """ link_prev, link_next, _key, result = link link_prev[NEXT] = link_next link_next[PREV] = link_prev """ 将命中节点移动到队尾 """ last = root[PREV] last[NEXT] = root[PREV] = link link[PREV] = last link[NEXT] = root hits += 1 return result """ 缓存未命中,调用方法获取返回,创建节点,淘汰算法 """ result = user_function(*args, **kwds) with RLock(): if key in cache: pass elif full: """ 缓冲区已满,淘汰队首元素,删除缓冲区对应元素 """ oldroot = root oldroot[KEY] = key oldroot[RESULT] = result root = oldroot[NEXT] oldkey = root[KEY] root[KEY] = root[RESULT] = None del cache[oldkey] cache[key] = oldroot else: """ 缓冲区未满,直接创建节点,插入数据 """ last = root[PREV] link = [last, root, key, result] last[NEXT] = root[PREV] = cache[key] = link full = (cache_len() >= maxsize) misses += 1 return result return lru_cache_wrapper return decorating_function def clock(func): @functools.wraps(func) def clocked(*args, **kwargs): t0 = time.time() result = func(*args, **kwargs) elapsed = time.time() - t0 name = func.__name__ arg_lst = [] if args: arg_lst.append(', '.join(repr(arg) for arg in args)) if kwargs: pairs = ['%s=%r' % (k, w) for k, w in sorted(kwargs.items())] arg_lst.append(', '.join(pairs)) arg_str = ', '.join(arg_lst) print('[%0.8fs] %s(%s) -> %r ' % (elapsed, name, arg_str, result)) return result return clocked @lru_cache() @clock def fibonacci(n): if n < 2: return n return fibonacci(n - 2) + fibonacci(n - 1) if __name__ == '__main__': print(fibonacci(6))

 

 

打印出了:

[0.00000000s] fibonacci(0) -> 0 

[0.00000000s] fibonacci(1) -> 1 

[0.00000000s] fibonacci(2) -> 1 

[0.00000000s] fibonacci(3) -> 2 

[0.00000000s] fibonacci(4) -> 3 

[0.00000000s] fibonacci(5) -> 5 

[0.00000000s] fibonacci(6) -> 8 

8

 

我们看到,方法从原来的 25 次递归调用变成了只有 7 次递归调用,执行时间上的优化效果也是相当明显

 

欢迎关注微信公众号,以技术为主,涉及历史、人文等多领域的学习与感悟,每周三到七篇推文,全部原创,只有干货没有鸡汤

 

 






技术帖      算法      斐波那契      python      缓存      分布式      递归      装饰器      缓存淘汰      lru_cache      functools     


京ICP备15018585号