在計(jì)算機(jī)軟件領(lǐng)域，緩存（Cache）指的是將部分?jǐn)?shù)據(jù)存儲(chǔ)在內(nèi)存中，以便下次能夠更快地訪問(wèn)這些數(shù)據(jù)，這也是一個(gè)典型的用空間換時(shí)間的例子。一般用于緩存的內(nèi)存空間是固定的，當(dāng)有更多的數(shù)據(jù)需要緩存的時(shí)候，需要將已緩存的部分?jǐn)?shù)據(jù)清除后再將新的緩存數(shù)據(jù)放進(jìn)去。需要清除哪些數(shù)據(jù)，就涉及到了緩存置換的策略，LRU（Least Recently Used，最近最少使用）是很常見(jiàn)的一個(gè)，也是 Python 中提供的緩存置換策略。

下面我們通過(guò)一個(gè)簡(jiǎn)單的示例來(lái)看 Python 中的 lru_cache 是如何使用的。

def factorial(n): print(f'計(jì)算 {n} 的階乘') return 1 if n <= 1 else n * factorial(n - 1)a = factorial(5)print(f’5! = {a}’)b = factorial(3)print(f’3! = {b}’)

上面的代碼中定義了函數(shù) factorial，通過(guò)遞歸的方式計(jì)算 n 的階乘，并且在函數(shù)調(diào)用的時(shí)候打印出 n 的值。然后分別計(jì)算 5 和 3 的階乘，并打印結(jié)果。運(yùn)行上面的代碼，輸出如下

計(jì)算 5 的階乘計(jì)算 4 的階乘計(jì)算 3 的階乘計(jì)算 2 的階乘計(jì)算 1 的階乘5! = 120計(jì)算 3 的階乘計(jì)算 2 的階乘計(jì)算 1 的階乘3! = 6

可以看到， factorial(3) 的結(jié)果在計(jì)算 factorial(5) 的時(shí)候已經(jīng)被計(jì)算過(guò)了，但是后面又被重復(fù)計(jì)算了。為了避免這種重復(fù)計(jì)算，我們可以在定義函數(shù) factorial 的時(shí)候加上 lru_cache 裝飾器，如下所示

import functools# 注意 lru_cache 后的一對(duì)括號(hào)，證明這是帶參數(shù)的裝飾器@functools.lru_cache()def factorial(n): print(f'計(jì)算 {n} 的階乘') return 1 if n <= 1 else n * factorial(n - 1)

重新運(yùn)行代碼，輸入如下

計(jì)算 5 的階乘計(jì)算 4 的階乘計(jì)算 3 的階乘計(jì)算 2 的階乘計(jì)算 1 的階乘5! = 1203! = 6

可以看到，這次在調(diào)用 factorial(3) 的時(shí)候沒(méi)有打印相應(yīng)的輸出，也就是說(shuō) factorial(3) 是直接從緩存讀取的結(jié)果，證明緩存生效了。

被 lru_cache 修飾的函數(shù)在被相同參數(shù)調(diào)用的時(shí)候，后續(xù)的調(diào)用都是直接從緩存讀結(jié)果，而不用真正執(zhí)行函數(shù)。下面我們深入源碼，看看 Python 內(nèi)部是怎么實(shí)現(xiàn) lru_cache 的。寫作時(shí) Python 最新發(fā)行版是 3.9，所以這里使用的是Python 3.9 的源碼 ，并且保留了源碼中的注釋。

def lru_cache(maxsize=128, typed=False): '''Least-recently-used cache decorator. If *maxsize* is set to None, the LRU features are disabled and the cache can grow without bound. If *typed* is True, arguments of different types will be cached separately. For example, f(3.0) and f(3) will be treated as distinct calls with distinct results. Arguments to the cached function must be hashable. View the cache statistics named tuple (hits, misses, maxsize, currsize) with f.cache_info(). Clear the cache and statistics with f.cache_clear(). Access the underlying function with f.__wrapped__. See: http://en.wikipedia.org/wiki/Cache_replacement_policies#Least_recently_used_(LRU) ''' # Users should only access the lru_cache through its public API: # cache_info, cache_clear, and f.__wrapped__ # The internals of the lru_cache are encapsulated for thread safety and # to allow the implementation to change (including a possible C version). if isinstance(maxsize, int): # Negative maxsize is treated as 0 if maxsize < 0: maxsize = 0 elif callable(maxsize) and isinstance(typed, bool): # The user_function was passed in directly via the maxsize argument user_function, maxsize = maxsize, 128 wrapper = _lru_cache_wrapper(user_function, maxsize, typed, _CacheInfo) wrapper.cache_parameters = lambda : {’maxsize’: maxsize, ’typed’: typed} return update_wrapper(wrapper, user_function) elif maxsize is not None: raise TypeError( ’Expected first argument to be an integer, a callable, or None’) def decorating_function(user_function): wrapper = _lru_cache_wrapper(user_function, maxsize, typed, _CacheInfo) wrapper.cache_parameters = lambda : {’maxsize’: maxsize, ’typed’: typed} return update_wrapper(wrapper, user_function) return decorating_function

這段代碼中有如下幾個(gè)關(guān)鍵點(diǎn)

關(guān)鍵字參數(shù)

maxsize 表示緩存容量，如果為 None 表示容量不設(shè)限， typed 表示是否區(qū)分參數(shù)類型，注釋中也給出了解釋，如果 typed == True ，那么 f(3) 和 f(3.0) 會(huì)被認(rèn)為是不同的函數(shù)調(diào)用。

第 24 行的條件分支

如果 lru_cache 的第一個(gè)參數(shù)是可調(diào)用的，直接返回 wrapper，也就是把 lru_cache 當(dāng)做不帶參數(shù)的裝飾器，這是 Python 3.8 才有的特性，也就是說(shuō)在 Python 3.8 及之后的版本中我們可以用下面的方式使用 lru_cache，可能是為了防止程序員在使用 lru_cache 的時(shí)候忘記加括號(hào)。

import functools# 注意 lru_cache 后面沒(méi)有括號(hào)，# 證明這是將其當(dāng)做不帶參數(shù)的裝飾器@functools.lru_cachedef factorial(n): print(f'計(jì)算 {n} 的階乘') return 1 if n <= 1 else n * factorial(n - 1)

注意，Python 3.8 之前的版本運(yùn)行上面代碼會(huì)報(bào)錯(cuò)：TypeError: Expected maxsize to be an integer or None。

lru_cache 的具體邏輯是在 _lru_cache_wrapper 函數(shù)中實(shí)現(xiàn)的，還是一樣，列出源碼，保留注釋。

def _lru_cache_wrapper(user_function, maxsize, typed, _CacheInfo): # Constants shared by all lru cache instances: sentinel = object() # unique object used to signal cache misses make_key = _make_key # build a key from the function arguments PREV, NEXT, KEY, RESULT = 0, 1, 2, 3 # names for the link fields cache = {} hits = misses = 0 full = False cache_get = cache.get # bound method to lookup a key or return None cache_len = cache.__len__ # get cache size without calling len() lock = RLock() # because linkedlist updates aren’t threadsafe root = []# root of the circular doubly linked list root[:] = [root, root, None, None] # initialize by pointing to self if maxsize == 0: def wrapper(*args, **kwds): # No caching -- just a statistics update nonlocal misses misses += 1 result = user_function(*args, **kwds) return result elif maxsize is None: def wrapper(*args, **kwds): # Simple caching without ordering or size limit nonlocal hits, misses key = make_key(args, kwds, typed) result = cache_get(key, sentinel) if result is not sentinel:hits += 1return result misses += 1 result = user_function(*args, **kwds) cache[key] = result return result else: def wrapper(*args, **kwds): # Size limited caching that tracks accesses by recency nonlocal root, hits, misses, full key = make_key(args, kwds, typed) with lock:link = cache_get(key)if link is not None: # Move the link to the front of the circular queue 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 resultmisses += 1 result = user_function(*args, **kwds) with lock:if key in cache: # Getting here means that this same key was added to the # cache while the lock was released. Since the link # update is already done, we need only return the # computed result and update the count of misses. passelif full: # Use the old root to store the new key and result. oldroot = root oldroot[KEY] = key oldroot[RESULT] = result # Empty the oldest link and make it the new root. # Keep a reference to the old key and old result to # prevent their ref counts from going to zero during the # update. That will prevent potentially arbitrary object # clean-up code (i.e. __del__) from running while we’re # still adjusting the links. root = oldroot[NEXT] oldkey = root[KEY] oldresult = root[RESULT] root[KEY] = root[RESULT] = None # Now update the cache dictionary. del cache[oldkey] # Save the potentially reentrant cache[key] assignment # for last, after the root and links have been put in # a consistent state. cache[key] = oldrootelse: # Put result in a new link at the front of the queue. last = root[PREV] link = [last, root, key, result] last[NEXT] = root[PREV] = cache[key] = link # Use the cache_len bound method instead of the len() function # which could potentially be wrapped in an lru_cache itself. full = (cache_len() >= maxsize) return result def cache_info(): '''Report cache statistics''' with lock: return _CacheInfo(hits, misses, maxsize, cache_len()) def cache_clear(): '''Clear the cache and cache statistics''' nonlocal hits, misses, full with lock: cache.clear() root[:] = [root, root, None, None] hits = misses = 0 full = False wrapper.cache_info = cache_info wrapper.cache_clear = cache_clear return wrapper

函數(shù)開(kāi)始的地方 2~14 行定義了一些關(guān)鍵變量，

hits 和 misses 分別表示緩存命中和沒(méi)有命中的次數(shù) root 雙向循環(huán)鏈表的頭結(jié)點(diǎn)，每個(gè)節(jié)點(diǎn)保存前向指針、后向指針、key 和 key 對(duì)應(yīng)的 result，其中 key 為 _make_key 函數(shù)根據(jù)參數(shù)結(jié)算出來(lái)的字符串，result 為被修飾的函數(shù)在給定的參數(shù)下返回的結(jié)果。 注意 ，root 是不保存數(shù)據(jù) key 和 result 的。 cache 是真正保存緩存數(shù)據(jù)的地方，類型為 dict。 cache 中的 key 也是 _make_key 函數(shù)根據(jù)參數(shù)結(jié)算出來(lái)的字符串，value 保存的是 key 對(duì)應(yīng)的雙向循環(huán)鏈表中的節(jié)點(diǎn)。

接下來(lái)根據(jù) maxsize 不同，定義不同的 wrapper 。

maxsize == 0 ，其實(shí)也就是沒(méi)有緩存，那么每次函數(shù)調(diào)用都不會(huì)命中，并且沒(méi)有命中的次數(shù) misses 加 1。 maxsize is None ，不限制緩存大小，如果函數(shù)調(diào)用不命中，將沒(méi)有命中次數(shù) misses 加 1，否則將命中次數(shù) hits 加 1。 限制緩存的大小，那么需要根據(jù) LRU 算法來(lái)更新 cache ，也就是 42~97 行的代碼。 如果緩存命中 key，那么將命中節(jié)點(diǎn)移到雙向循環(huán)鏈表的結(jié)尾，并且返回結(jié)果（47~58 行） 這里通過(guò)字典加雙向循環(huán)鏈表的組合數(shù)據(jù)結(jié)構(gòu)，實(shí)現(xiàn)了用 O(1) 的時(shí)間復(fù)雜度刪除給定的節(jié)點(diǎn)。 如果沒(méi)有命中，并且緩存滿了，那么需要將最久沒(méi)有使用的節(jié)點(diǎn)（root 的下一個(gè)節(jié)點(diǎn)）刪除，并且將新的節(jié)點(diǎn)添加到鏈表結(jié)尾。在實(shí)現(xiàn)中有一個(gè)優(yōu)化，直接將當(dāng)前的 root 的 key 和 result 替換成新的值，將 root 的下一個(gè)節(jié)點(diǎn)置為新的 root，這樣得到的雙向循環(huán)鏈表結(jié)構(gòu)跟刪除 root 的下一個(gè)節(jié)點(diǎn)并且將新節(jié)點(diǎn)加到鏈表結(jié)尾是一樣的，但是避免了刪除和添加節(jié)點(diǎn)的操作（68~88 行） 如果沒(méi)有命中，并且緩存沒(méi)滿，那么直接將新節(jié)點(diǎn)添加到雙向循環(huán)鏈表的結(jié)尾（ root[PREV] ，這里我認(rèn)為是結(jié)尾，但是代碼注釋中寫的是開(kāi)頭）（89~96 行）

最后給 wrapper 添加兩個(gè)屬性函數(shù) cache_info 和 cache_clear ， cache_info 顯示當(dāng)前緩存的命中情況的統(tǒng)計(jì)數(shù)據(jù)， cache_clear 用于清空緩存。對(duì)于上面階乘相關(guān)的代碼，如果在最后執(zhí)行 factorial.cache_info() ，會(huì)輸出

CacheInfo(hits=1, misses=5, maxsize=128, currsize=5)

第一次執(zhí)行 factorial(5) 的時(shí)候都沒(méi)命中，所以 misses = 5，第二次執(zhí)行 factorial(3) 的時(shí)候，緩存命中，所以 hits = 1。

最后需要說(shuō)明的是， 對(duì)于有多個(gè)關(guān)鍵字參數(shù)的函數(shù)，如果兩次調(diào)用函數(shù)關(guān)鍵字參數(shù)傳入的順序不同，會(huì)被認(rèn)為是不同的調(diào)用，不會(huì)命中緩存。另外，被 lru_cache 裝飾的函數(shù)不能包含可變類型參數(shù)如 list，因?yàn)樗鼈儾恢С?hash。

總結(jié)一下，這篇文章首先簡(jiǎn)介了一下緩存的概念，然后展示了在 Python 中 lru_cache 的使用方法，最后通過(guò)源碼分析了 Python 中 lru_cache 的實(shí)現(xiàn)細(xì)節(jié)。

到此這篇關(guān)于Python中l(wèi)ru_cache的使用和實(shí)現(xiàn)詳解的文章就介紹到這了,更多相關(guān)Python lru_cache 內(nèi)容請(qǐng)搜索好吧啦網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持好吧啦網(wǎng)！