Python中lru_cache的使用和实现详解
在计算机软件领域，缓存（Cache）指的是将部分数据存储在内存中，以便下次能够更快地访问这些数据，这也是一个典型的用空间换时间的例子。一般用于缓存的内存空间是固定的，当有更多的数据需要缓存的时候，需要将已缓存的部分数据清除后再将新的缓存数据放进去。需要清除哪些数据，就涉及到了缓存置换的策略，LRU（LeastRecentlyUsed，最近最少使用）是很常见的一个，也是Python中提供的缓存置换策略。
下面我们通过一个简单的示例来看Python中的lru_cache是如何使用的。
deffactorial(n):
print(f"计算{n}的阶乘")
return1ifn<=1elsen*factorial(n-1)
a=factorial(5)
print(f'5!={a}')
b=factorial(3)
print(f'3!={b}')
上面的代码中定义了函数factorial，通过递归的方式计算n的阶乘，并且在函数调用的时候打印出n的值。然后分别计算5和3的阶乘，并打印结果。运行上面的代码，输出如下
计算5的阶乘
计算4的阶乘
计算3的阶乘
计算2的阶乘
计算1的阶乘
5!=120
计算3的阶乘
计算2的阶乘
计算1的阶乘
3!=6
可以看到，factorial(3)的结果在计算factorial(5)的时候已经被计算过了，但是后面又被重复计算了。为了避免这种重复计算，我们可以在定义函数factorial的时候加上lru_cache装饰器，如下所示
importfunctools
#注意lru_cache后的一对括号，证明这是带参数的装饰器
@functools.lru_cache()
deffactorial(n):
print(f"计算{n}的阶乘")
return1ifn<=1elsen*factorial(n-1)
重新运行代码，输入如下
计算5的阶乘
计算4的阶乘
计算3的阶乘
计算2的阶乘
计算1的阶乘
5!=120
3!=6
可以看到，这次在调用factorial(3)的时候没有打印相应的输出，也就是说factorial(3)是直接从缓存读取的结果，证明缓存生效了。
被lru_cache修饰的函数在被相同参数调用的时候，后续的调用都是直接从缓存读结果，而不用真正执行函数。下面我们深入源码，看看Python内部是怎么实现lru_cache的。写作时Python最新发行版是3.9，所以这里使用的是Python3.9的源码，并且保留了源码中的注释。
deflru_cache(maxsize=128,typed=False):
"""Least-recently-usedcachedecorator.
If*maxsize*issettoNone,theLRUfeaturesaredisabledandthecache
cangrowwithoutbound.
If*typed*isTrue,argumentsofdifferenttypeswillbecachedseparately.
Forexample,f(3.0)andf(3)willbetreatedasdistinctcallswith
distinctresults.
Argumentstothecachedfunctionmustbehashable.
Viewthecachestatisticsnamedtuple(hits,misses,maxsize,currsize)
withf.cache_info().Clearthecacheandstatisticswithf.cache_clear().
Accesstheunderlyingfunctionwithf.__wrapped__.
See:http://en.wikipedia.org/wiki/Cache_replacement_policies#Least_recently_used_(LRU)
"""
#Usersshouldonlyaccessthelru_cachethroughitspublicAPI:
#cache_info,cache_clear,andf.__wrapped__
#Theinternalsofthelru_cacheareencapsulatedforthreadsafetyand
#toallowtheimplementationtochange(includingapossibleCversion).
ifisinstance(maxsize,int):
#Negativemaxsizeistreatedas0
ifmaxsize<0:
maxsize=0
elifcallable(maxsize)andisinstance(typed,bool):
#Theuser_functionwaspassedindirectlyviathemaxsizeargument
user_function,maxsize=maxsize,128
wrapper=_lru_cache_wrapper(user_function,maxsize,typed,_CacheInfo)
wrapper.cache_parameters=lambda:{'maxsize':maxsize,'typed':typed}
returnupdate_wrapper(wrapper,user_function)
elifmaxsizeisnotNone:
raiseTypeError(
'Expectedfirstargumenttobeaninteger,acallable,orNone')
defdecorating_function(user_function):
wrapper=_lru_cache_wrapper(user_function,maxsize,typed,_CacheInfo)
wrapper.cache_parameters=lambda:{'maxsize':maxsize,'typed':typed}
returnupdate_wrapper(wrapper,user_function)
returndecorating_function
这段代码中有如下几个关键点
关键字参数
maxsize表示缓存容量，如果为None表示容量不设限，typed表示是否区分参数类型，注释中也给出了解释，如果typed==True，那么f(3)和f(3.0)会被认为是不同的函数调用。
第24行的条件分支
如果lru_cache的第一个参数是可调用的，直接返回wrapper，也就是把lru_cache当做不带参数的装饰器，这是Python3.8才有的特性，也就是说在Python3.8及之后的版本中我们可以用下面的方式使用lru_cache，可能是为了防止程序员在使用lru_cache的时候忘记加括号。
importfunctools
#注意lru_cache后面没有括号，
#证明这是将其当做不带参数的装饰器
@functools.lru_cache
deffactorial(n):
print(f"计算{n}的阶乘")
return1ifn<=1elsen*factorial(n-1)
注意，Python3.8之前的版本运行上面代码会报错：TypeError:ExpectedmaxsizetobeanintegerorNone。
lru_cache的具体逻辑是在_lru_cache_wrapper函数中实现的，还是一样，列出源码，保留注释。
def_lru_cache_wrapper(user_function,maxsize,typed,_CacheInfo):
#Constantssharedbyalllrucacheinstances:
sentinel=object()#uniqueobjectusedtosignalcachemisses
make_key=_make_key#buildakeyfromthefunctionarguments
PREV,NEXT,KEY,RESULT=0,1,2,3#namesforthelinkfields
cache={}
hits=misses=0
full=False
cache_get=cache.get#boundmethodtolookupakeyorreturnNone
cache_len=cache.__len__#getcachesizewithoutcallinglen()
lock=RLock()#becauselinkedlistupdatesaren'tthreadsafe
root=[]#rootofthecirculardoublylinkedlist
root[:]=[root,root,None,None]#initializebypointingtoself
ifmaxsize==0:
defwrapper(*args,**kwds):
#Nocaching--justastatisticsupdate
nonlocalmisses
misses+=1
result=user_function(*args,**kwds)
returnresult
elifmaxsizeisNone:
defwrapper(*args,**kwds):
#Simplecachingwithoutorderingorsizelimit
nonlocalhits,misses
key=make_key(args,kwds,typed)
result=cache_get(key,sentinel)
ifresultisnotsentinel:
hits+=1
returnresult
misses+=1
result=user_function(*args,**kwds)
cache[key]=result
returnresult
else:
defwrapper(*args,**kwds):
#Sizelimitedcachingthattracksaccessesbyrecency
nonlocalroot,hits,misses,full
key=make_key(args,kwds,typed)
withlock:
link=cache_get(key)
iflinkisnotNone:
#Movethelinktothefrontofthecircularqueue
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
returnresult
misses+=1
result=user_function(*args,**kwds)
withlock:
ifkeyincache:
#Gettingheremeansthatthissamekeywasaddedtothe
#cachewhilethelockwasreleased.Sincethelink
#updateisalreadydone,weneedonlyreturnthe
#computedresultandupdatethecountofmisses.
pass
eliffull:
#Usetheoldroottostorethenewkeyandresult.
oldroot=root
oldroot[KEY]=key
oldroot[RESULT]=result
#Emptytheoldestlinkandmakeitthenewroot.
#Keepareferencetotheoldkeyandoldresultto
#preventtheirrefcountsfromgoingtozeroduringthe
#update.Thatwillpreventpotentiallyarbitraryobject
#clean-upcode(i.e.__del__)fromrunningwhilewe're
#stilladjustingthelinks.
root=oldroot[NEXT]
oldkey=root[KEY]
oldresult=root[RESULT]
root[KEY]=root[RESULT]=None
#Nowupdatethecachedictionary.
delcache[oldkey]
#Savethepotentiallyreentrantcache[key]assignment
#forlast,aftertherootandlinkshavebeenputin
#aconsistentstate.
cache[key]=oldroot
else:
#Putresultinanewlinkatthefrontofthequeue.
last=root[PREV]
link=[last,root,key,result]
last[NEXT]=root[PREV]=cache[key]=link
#Usethecache_lenboundmethodinsteadofthelen()function
#whichcouldpotentiallybewrappedinanlru_cacheitself.
full=(cache_len()>=maxsize)
returnresult
defcache_info():
"""Reportcachestatistics"""
withlock:
return_CacheInfo(hits,misses,maxsize,cache_len())
defcache_clear():
"""Clearthecacheandcachestatistics"""
nonlocalhits,misses,full
withlock:
cache.clear()
root[:]=[root,root,None,None]
hits=misses=0
full=False
wrapper.cache_info=cache_info
wrapper.cache_clear=cache_clear
returnwrapper
函数开始的地方2~14行定义了一些关键变量，
- hits和misses分别表示缓存命中和没有命中的次数
- root双向循环链表的头结点，每个节点保存前向指针、后向指针、key和key对应的result，其中key为_make_key函数根据参数结算出来的字符串，result为被修饰的函数在给定的参数下返回的结果。注意，root是不保存数据key和result的。
- cache是真正保存缓存数据的地方，类型为dict。cache中的key也是_make_key函数根据参数结算出来的字符串，value保存的是key对应的双向循环链表中的节点。
接下来根据maxsize不同，定义不同的wrapper。
- maxsize==0，其实也就是没有缓存，那么每次函数调用都不会命中，并且没有命中的次数misses加1。
- maxsizeisNone，不限制缓存大小，如果函数调用不命中，将没有命中次数misses加1，否则将命中次数hits加1。
- 限制缓存的大小，那么需要根据LRU算法来更新cache，也就是42~97行的代码。
- 如果缓存命中key，那么将命中节点移到双向循环链表的结尾，并且返回结果（47~58行）
- 这里通过字典加双向循环链表的组合数据结构，实现了用O(1)的时间复杂度删除给定的节点。
- 如果没有命中，并且缓存满了，那么需要将最久没有使用的节点（root的下一个节点）删除，并且将新的节点添加到链表结尾。在实现中有一个优化，直接将当前的root的key和result替换成新的值，将root的下一个节点置为新的root，这样得到的双向循环链表结构跟删除root的下一个节点并且将新节点加到链表结尾是一样的，但是避免了删除和添加节点的操作（68~88行）
- 如果没有命中，并且缓存没满，那么直接将新节点添加到双向循环链表的结尾（root[PREV]，这里我认为是结尾，但是代码注释中写的是开头）（89~96行）
最后给wrapper添加两个属性函数cache_info和cache_clear，cache_info显示当前缓存的命中情况的统计数据，cache_clear用于清空缓存。对于上面阶乘相关的代码，如果在最后执行factorial.cache_info()，会输出
CacheInfo(hits=1,misses=5,maxsize=128,currsize=5)
第一次执行factorial(5)的时候都没命中，所以misses=5，第二次执行factorial(3)的时候，缓存命中，所以hits=1。
最后需要说明的是，对于有多个关键字参数的函数，如果两次调用函数关键字参数传入的顺序不同，会被认为是不同的调用，不会命中缓存。另外，被lru_cache装饰的函数不能包含可变类型参数如list，因为它们不支持hash。
总结一下，这篇文章首先简介了一下缓存的概念，然后展示了在Python中lru_cache的使用方法，最后通过源码分析了Python中lru_cache的实现细节。
到此这篇关于Python中lru_cache的使用和实现详解的文章就介绍到这了,更多相关Pythonlru_cache内容请搜索毛票票以前的文章或继续浏览下面的相关文章希望大家以后多多支持毛票票！