Python使用LRU緩存策略進行緩存的方法步驟
一、Python 緩存
① 緩存作用
- 緩存是一種優化技術，可以在應用程序中使用它來將最近或經常使用的數據保存在內存中，通過這種方式來訪問數據的速度比直接讀取磁盤文件的高很多。
- 假設我們搭建瞭一個新聞聚合網站，類似於 Feedly，其獲取不同來源的新聞然後聚合展示。當用戶瀏覽新聞的時候，後臺程序會將文章下載然後顯示到用戶屏幕上。如果不使用緩存技術的話，當用戶多次切換瀏覽相同文章的時候，必須多次下載，效率低下且很不友好。
- 更好的做法就是在獲取每篇文章之後，在本地進行內容的存儲，比如存儲在數據庫中；然後，當用戶下次打開同一篇文章的時候，後臺程序可以從本地存儲打開內容，而不是再次下載源文件，即這種技術稱為緩存。
② 使用 Python 字典實現緩存
以新聞聚合網站為例，不必每次都去下載文章內容，而是先檢查緩存數據中是否存在對應的內容，隻有當沒有時，才會讓服務器下載文章。
如下的示例程序，就是使用 Python 字典實現緩存的，將文章的 URL 作為鍵，並將其內容作為值；執行之後，可以看到當第二次執行 get_article 函數的時候，直接就返回結果並沒有讓服務器下載：
import requests
cache = dict()
def get_article_from_server(url):
print("Fetching article from server...")
response = requests.get(url)
return response.text
def get_article(url):
print("Getting article...")
if url not in cache:
cache[url] = get_article_from_server(url)
return cache[url]
get_article("https://www.escapelife.site/love-python.html")
get_article("https://www.escapelife.site/love-python.html")
將此代碼保存到一個 caching.py 文件中，安裝 requests 庫，然後運行腳本：
# 安裝依賴 $ pip install requests # 執行腳本 $ python python_caching.py Getting article... Fetching article from server... Getting article...
盡管調用 get_article() 兩次（第 17 行和第 18 行）字符串“Fetching article from server…”，但仍然隻輸出一次。發生這種情況的原因是，在第一次訪問文章之後，將其 URL 和內容放入緩存字典中，第二次時代碼不需要再次從服務器獲取項目。
③ 使用字典來做緩存的弊端
上面這種緩存實現存在一個非常大的問題，那就是字典的內容將會無限增長，即大量用戶連續瀏覽文章的時候，後臺程序將不斷向字典中塞入需要存儲的內容，服務器內存被擠爆，最終導致應用程序崩潰。
- 上面這種緩存實現存在一個非常大的問題，那就是字典的內容將會無限增長，即大量用戶連續瀏覽文章的時候，後臺程序將不斷向字典中塞入需要存儲的內容，服務器內存被擠爆，最終導致應用程序崩潰。
- 要解決上述這個問題，就需要有一種策略來決定哪些文章應該留在內存中，哪些文章應該被刪除掉，這些緩存策略其實是一些算法，它們用於管理緩存的信息，並選擇丟棄哪些項以為新項騰出空間。
- 當然這裡不必去實現管理緩存的算法，隻需要使用不同的策略來從緩存中移除項，並防止其增長超過其最大大小。五種常見的緩存算法如下所示：
| 緩存策略 | 英文名稱 | 淘汰條件 | 在什麼時候最有用 |
|---|---|---|---|
| 先進先出算法(FIFO) | First-In/First-Out | 淘汰最舊的條目 | 較新的條目最有可能被重用 |
| 後進先出算法(LIFO) | Last-In/First-Out | 淘汰最新的條目 | 較舊的條目最有可能被重用 |
| 最近最少使用算法(LRU) | Least Recently Used | 淘汰最近使用最少的條目 | 最近使用的條目最有可能被重用 |
| 最近最多使用算法(MRU) | Most Recently Used | 淘汰最近使用最多的條目 | 最近不用的條目最有可能被重用 |
| 最近最少命中算法(LFU) | Least Frequently Used | 淘汰最不經常訪問的條目 | 命中率很高的條目更有可能被重用 |
看瞭上述五種緩存算法，是不是看到 LRU 和 LFU 的時候有點懵，主要是通過中文對應的解釋很難理解其真實的含義，看看英文的話就不難理解瞭。LRU 和 LFU 算法的不同之處在於：
- LRU 基於訪問時間的淘汰規則，根據數據的歷史訪問記錄來進行淘汰數據，如果數據最近被訪問過，那麼將來被訪問的幾率也更高；
- LFU 基於訪問次數的淘汰規則，根據數據的歷史訪問頻率來淘汰數據，如果數據過去被訪問多次，那麼將來被訪問的頻率也更高；
比如，以十分鐘為一個節點，每分鐘進行一次頁面調度，當所需的頁面走向為 2 1 2 4 2 3 4 時，且調頁面 4 時會發生缺頁中斷；若按 LRU 算法的話，應換頁面 1（十分鐘內頁面 1 最久未被使用），但按 LFU 算法的話，應換頁面 3（十分鐘內頁面 3 隻使用一次）。
二、深入理解 LRU 算法
① 查看 LRU 緩存的特點
使用 LRU 策略實現的緩存是按照使用順序進行排序的，每次訪問條目時，LRU 算法就會將其移到緩存的頂部。通過這種方式，算法可以通過查看列表的底部，快速識別出最長時間未使用的條目。
如下所示，用戶從網絡上請求第一篇文章的 LRU 策略存儲記錄：
在將文章提供給用戶之前，緩存如何將其存儲在最近的槽中？如下所示，用戶請求第二篇文章時發生的情況，第二篇文章存儲到最上層的位置，即第二篇文章采用瞭最近的位置，將第一篇文章推到列表下方：
LRU 策略假定使用的對象越新，將來使用該對象的可能性就越大，因此它嘗試將該對象保留在緩存中的時間最長，即如果發生條目淘汰的話，會優先淘汰第一篇文檔的緩存存儲記錄。
② 查看 LRU 緩存的結構
在 Python 中實現 LRU 緩存的一種方法就是使用雙向鏈表（doubly linked list）和哈希映射（hash map），雙向鏈表的頭元素將指向最近使用的條目，而其尾部將指向最近使用最少的條目。LRU 緩存實現邏輯結構如下：
通過使用哈希映射，可以將每個條目映射到雙鏈表中的特定位置，從而確保對緩存中的每個項的訪問。這個策略非常快，訪問最近最少使用的項和更新緩存的復雜度均為 O(1) 操作。
而從 Python3.2 版本開始，Python 新增 @lru_cache 這個裝飾器用於實現 LRU 策略，從此可以使用這個裝飾器來裝飾函數並緩存其計算結果。
三、使用 lru_cache 裝飾器
① @lru_cache 裝飾器的實現原理
有很多方法可以實現應用程序的快速響應，而使用緩存就是一種非常常見的方法。如果能夠正確使用緩存的話，可以使響應變得更快且減少計算資源的額外負載。
在 Python 中 functools 模塊自帶瞭 @lru_cache 這個裝飾器來做緩存，其能夠使用最近最少使用（LRU）策略來緩存函數的計算結果，這是一種簡單但功能強大的技術：
- 實現 @lru_cache 裝飾器；
- 瞭解 LRU 策略的工作運作原理；
- 使用 @lru_cache 裝飾器來提高性能；
- 擴展 @lru_cache 裝飾器的功能並使其在特定時間後過期。
就像先前實現的緩存方案一樣，Python 中的 @lru_cache 裝飾器存儲也是使用字典來做為存儲對象的，它將函數的執行結果緩存在字典的 key 裡面，該 key 由對該函數的調用（包括函數的參數）組成，這就意味著這些函數的參數必須是可哈希的，裝飾器才能正常工作。
② 斐波拉契數列
我們都應該知道斐波拉契數列的計算方式，常見的解決方式就是使用遞歸的思路：
- 0、1、1、2、3、5, 8、13、21、34 ……；
- 2 是上兩項的和 ->（1+1）；
- 3 是上兩項的和 ->（1+2）；
- 5 是上兩項的和 ->（2+3）。
遞歸的計算簡潔並且直觀，但是由於存在大量重復計算，實際運行效率很低，並且會占用較多的內存。但是這裡並不是需要關註的重點，隻是來作為演示示例而已：
# 匿名函數 fib = lambda n: 1 if n <= 1 else fib(n-1) + fib(n-2) # 將時間復雜度降低到線性 fib = lambda n, a=1, b=1: a if n == 0 else fib(n-1, b, a+b) # 保證瞭匿名函數的匿名性 fib = lambda n, fib: 1 if n <= 1 else fib(n-1, fib) + fib(n-2, fib)
③ 使用 @lru_cache 緩存輸出結果
使用 @lru_cache 裝飾器來緩存的話，可以將函數調用結果存儲在內存中，以便再次請求時直接返回結果：
from functools import lru_cache
@lru_cache
def fib(n):
if n==1 or n==2:
return 1
else:
return fib(n-1) + fib(n-2)
④ 限制 @lru_cache 裝飾器大小
Python 的 @lru_cache 裝飾器提供瞭一個 maxsize 屬性，該屬性定義瞭在緩存開始淘汰舊條目之前的最大條目數，默認情況下，maxsize 設置為 128。
如果將 maxsize 設置為 None 的話，則緩存將無限期增長，並且不會驅逐任何條目。
from functools import lru_cache
@lru_cache(maxsize=16)
def fib(n):
if n==1 or n==2:
return 1
else:
return fib(n-1) + fib(n-2)
# 查看緩存列表 >>> print(steps_to.cache_info()) CacheInfo(hits=52, misses=30, maxsize=16, currsize=16)
⑤ 使用 @lru_cache 實現 LRU 緩存
就像在前面實現的緩存解決方案一樣，@lru_cache 在底層使用一個字典，它將函數的結果緩存在一個鍵下，該鍵包含對函數的調用，包括提供的參數。這意味著這些參數必須是可哈希的，才能讓 decorator 工作。
示例：玩樓梯：
想象一下，你想通過一次跳上一個、兩個或三個樓梯來確定到達樓梯中的一個特定樓梯的所有不同方式，到第四個樓梯有多少條路？所有不同的組合如下所示：
可以這樣描述，為瞭到達當前的樓梯，你可以從下面的一個、兩個或三個樓梯跳下去，將能夠到達這些點的跳躍組合的數量相加，便能夠獲得到達當前位置的所有可能方法。
例如到達第四個樓梯的組合數量將等於你到達第三、第二和第一個樓梯的不同方式的總數。如下所示，有七種不同的方法可以到達第四層樓梯：
註意給定階梯的解是如何建立在較小子問題的答案之上的，在這種情況下，為瞭確定到達第四個樓梯的不同路徑，可以將到達第三個樓梯的四種路徑、到達第二個樓梯的兩種路徑以及到達第一個樓梯的一種路徑相加。 這種方法稱為遞歸，下面是一個實現這個遞歸的函數：
def steps_to(stair):
if stair == 1:
# You can reach the first stair with only a single step
# from the floor.
return 1
elif stair == 2:
# You can reach the second stair by jumping from the
# floor with a single two-stair hop or by jumping a single
# stair a couple of times.
return 2
elif stair == 3:
# You can reach the third stair using four possible
# combinations:
# 1. Jumping all the way from the floor
# 2. Jumping two stairs, then one
# 3. Jumping one stair, then two
# 4. Jumping one stair three times
return 4
else:
# You can reach your current stair from three different places:
# 1. From three stairs down
# 2. From two stairs down
# 2. From one stair down
#
# If you add up the number of ways of getting to those
# those three positions, then you should have your solution.
return (
steps_to(stair - 3)
+ steps_to(stair - 2)
+ steps_to(stair - 1)
)
print(steps_to(4))
將此代碼保存到一個名為 stairs.py 的文件中，並使用以下命令運行它：
$ python stairs.py 7
太棒瞭，這個代碼適用於 4 個樓梯，但是數一下要走多少步才能到達樓梯上更高的地方呢？將第 33 行中的樓梯數更改為 30，並重新運行腳本：
$ python stairs.py 53798080
可以看到結果超過 5300 萬個組合，這可真的有點多。
時間代碼：
當找到第 30 個樓梯的解決方案時，腳本花瞭相當多的時間來完成。要獲得基線，可以度量代碼運行的時間，要做到這一點，可以使用 Python 的 timeit module，在第 33 行之後添加以下代碼：
setup_code = "from __main__ import steps_to"
36stmt = "steps_to(30)"
37times = repeat(setup=setup_code, stmt=stmt, repeat=3, number=10)
38print(f"Minimum execution time: {min(times)}")
還需要在代碼的頂部導入 timeit module：
from timeit import repeat
以下是對這些新增內容的逐行解釋：
- 第 35 行導入 steps_to() 的名稱，以便 time.com .repeat() 知道如何調用它；
- 第 36 行用希望到達的樓梯數（在本例中為 30）準備對函數的調用，這是將要執行和計時的語句；
- 第 37 行使用設置代碼和語句調用 time.repeat()，這將調用該函數 10 次，返回每次執行所需的秒數；
- 第 38 行標識並打印返回的最短時間。 現在再次運行腳本：
$ python stairs.py 53798080 Minimum execution time: 40.014977024000004
可以看到的秒數取決於特定硬件，在我的系統上，腳本花瞭 40 秒，這對於 30 級樓梯來說是相當慢的。
使用記憶來改進解決方案：
這種遞歸實現通過將其分解為相互構建的更小的步驟來解決這個問題，如下所示是一個樹，其中每個節點表示對 steps_to() 的特定調用：
註意需要如何使用相同的參數多次調用 steps_to()，例如 steps_to(5) 計算兩次，steps_to(4) 計算四次，steps_to(3) 計算七次，steps_to(2) 計算六次，多次調用同一個函數會增加不必要的計算周期，結果總是相同的。
為瞭解決這個問題，可以使用一種叫做記憶的技術，這種方法將函數的結果存儲在內存中，然後在需要時引用它，從而確保函數不會為相同的輸入運行多次，這個場景聽起來像是使用 Python 的 @lru_cache 裝飾器的絕佳機會。
隻要做兩個改變，就可以大大提高算法的運行時間：
- 從 functools module 導入 @lru_cache 裝飾器；
- 使用 @lru_cache 裝飾 steps_to()。
下面是兩個更新後的腳本頂部的樣子：
from functools import lru_cache from timeit import repeat @lru_cache def steps_to(stair): if stair == 1:
運行更新後的腳本產生如下結果：
$ python stairs.py 53798080 Minimum execution time: 7.999999999987184e-07
緩存函數的結果會將運行時從 40 秒降低到 0.0008 毫秒，這是一個瞭不起的進步。@lru_cache 裝飾器存儲瞭每個不同輸入的 steps_to() 的結果，每次代碼調用帶有相同參數的函數時，它都直接從內存中返回正確的結果，而不是重新計算一遍答案，這解釋瞭使用 @lru_cache 時性能的巨大提升。
⑥ 解包 @lru_cache 的功能
有瞭@lru_cache 裝飾器，就可以將每個調用和應答存儲在內存中，以便以後再次請求時進行訪問，但是在內存耗盡之前，可以節省多少次調用呢？
Python 的 @lru_cache 裝飾器提供瞭一個 maxsize 屬性，它定義瞭在緩存開始清除舊條目之前的最大條目數，缺省情況下，maxsize 設置為 128，如果將 maxsize 設置為 None，那麼緩存將無限增長，並且不會驅逐任何條目。如果在內存中存儲大量不同的調用，這可能會成為一個問題。
如下是 @lru_cache 使用 maxsize 屬性：
from functools import lru_cache
from timeit import repeat
@lru_cache(maxsize=16)
def steps_to(stair):
if stair == 1:
在本例中，將緩存限制為最多 16 個條目，當一個新調用傳入時，decorator 的實現將會從現有的 16 個條目中刪除最近最少使用的條目，為新條目騰出位置。
要查看添加到代碼中的新內容會發生什麼，可以使用 @lru_cache 裝飾器提供的 cache_info() 來檢查命中和未命中的次數以及當前緩存的大小。為瞭清晰起見，刪除乘以函數運行時的代碼，以下是修改後的最終腳本：
from functools import lru_cache
from timeit import repeat
@lru_cache(maxsize=16)
def steps_to(stair):
if stair == 1:
# You can reach the first stair with only a single step
# from the floor.
return 1
elif stair == 2:
# You can reach the second stair by jumping from the
# floor with a single two-stair hop or by jumping a single
# stair a couple of times.
return 2
elif stair == 3:
# You can reach the third stair using four possible
# combinations:
# 1. Jumping all the way from the floor
# 2. Jumping two stairs, then one
# 3. Jumping one stair, then two
# 4. Jumping one stair three times
return 4
else:
# You can reach your current stair from three different places:
# 1. From three stairs down
# 2. From two stairs down
# 2. From one stair down
#
# If you add up the number of ways of getting to those
# those three positions, then you should have your solution.
return (
steps_to(stair - 3)
+ steps_to(stair - 2)
+ steps_to(stair - 1)
)
print(steps_to(30))
print(steps_to.cache_info())
如果再次調用腳本，可以看到如下結果：
$ python stairs.py 53798080 CacheInfo(hits=52, misses=30, maxsize=16, currsize=16)
可以使用 cache_info() 返回的信息來瞭解緩存是如何執行的，並對其進行微調，以找到速度和存儲之間的適當平衡。下面是 cache_info() 提供的屬性的詳細說明：
- hits=52 是 @lru_cache 直接從內存中返回的調用數，因為它們存在於緩存中；
- misses =30 是被計算的不是來自內存的調用數，因為試圖找到到達第 30 級樓梯的臺階數，所以每次調用都在第一次調用時錯過瞭緩存是有道理的；
- maxsize =16 是用裝飾器的 maxsize 屬性定義的緩存的大小；
- currsize =16 是當前緩存的大小，在本例中它表明緩存已滿。
如果需要從緩存中刪除所有條目，那麼可以使用 @lru_cache 提供的 cache_clear()。
四、添加緩存過期
假設想要開發一個腳本來監視 Real Python 並在任何包含單詞 Python 的文章中打印字符數。真正的 Python 提供瞭一個 Atom feed，因此可以使用 feedparser 庫來解析提要，並使用請求庫來加載本文的內容。
如下是監控腳本的實現：
import feedparser
import requests
import ssl
import time
if hasattr(ssl, "_create_unverified_context"):
ssl._create_default_https_context = ssl._create_unverified_context
def get_article_from_server(url):
print("Fetching article from server...")
response = requests.get(url)
return response.text
def monitor(url):
maxlen = 45
while True:
print("\nChecking feed...")
feed = feedparser.parse(url)
for entry in feed.entries[:5]:
if "python" in entry.title.lower():
truncated_title = (
entry.title[:maxlen] + "..."
if len(entry.title) > maxlen
else entry.title
)
print(
"Match found:",
truncated_title,
len(get_article_from_server(entry.link)),
)
time.sleep(5)
monitor("https://realpython.com/atom.xml")
將此腳本保存到一個名為 monitor.py 的文件中，安裝 feedparser 和請求庫，然後運行該腳本，它將持續運行，直到在終端窗口中按 Ctrl+C 停止它：
$ pip install feedparser requests $ python monitor.py Checking feed... Fetching article from server... The Real Python Podcast – Episode #28: Using ... 29520 Fetching article from server... Python Community Interview With David Amos 54256 Fetching article from server... Working With Linked Lists in Python 37099 Fetching article from server... Python Practice Problems: Get Ready for Your ... 164888 Fetching article from server... The Real Python Podcast – Episode #27: Prepar... 30784 Checking feed... Fetching article from server... The Real Python Podcast – Episode #28: Using ... 29520 Fetching article from server... Python Community Interview With David Amos 54256 Fetching article from server... Working With Linked Lists in Python 37099 Fetching article from server... Python Practice Problems: Get Ready for Your ... 164888 Fetching article from server... The Real Python Podcast – Episode #27: Prepar... 30784
代碼解釋：
- 第 6 行和第 7 行：當 feedparser 試圖訪問通過 HTTPS 提供的內容時，這是一個解決方案；
- 第 16 行：monitor() 將無限循環；
- 第 18 行：使用 feedparser，代碼從真正的 Python 加載並解析提要；
- 第 20 行：循環遍歷列表中的前 5 個條目；
- 第 21 到 31 行：如果單詞 python 是標題的一部分，那麼代碼將連同文章的長度一起打印它；
- 第 33 行：代碼在繼續之前休眠瞭 5 秒鐘；
- 第 35 行：這一行通過將 Real Python 提要的 URL 傳遞給 monitor() 來啟動監視過程。
每當腳本加載一篇文章時，“Fetching article from server…”的消息就會打印到控制臺，如果讓腳本運行足夠長的時間，那麼將看到這條消息是如何反復顯示的，即使在加載相同的鏈接時也是如此。
這是一個很好的機會來緩存文章的內容，並避免每五秒鐘訪問一次網絡，可以使用 @lru_cache 裝飾器，但是如果文章的內容被更新，會發生什麼呢？第一次訪問文章時，裝飾器將存儲文章的內容，並在以後每次返回相同的數據；如果更新瞭帖子，那麼監視器腳本將永遠無法實現它，因為它將提取存儲在緩存中的舊副本。要解決這個問題，可以將緩存條目設置為過期。
from functools import lru_cache, wraps
from datetime import datetime, timedelta
def timed_lru_cache(seconds: int, maxsize: int = 128):
def wrapper_cache(func):
func = lru_cache(maxsize=maxsize)(func)
func.lifetime = timedelta(seconds=seconds)
func.expiration = datetime.utcnow() + func.lifetime
@wraps(func)
def wrapped_func(*args, **kwargs):
if datetime.utcnow() >= func.expiration:
func.cache_clear()
func.expiration = datetime.utcnow() + func.lifetime
return func(*args, **kwargs)
return wrapped_func
return wrapper_cache
@timed_lru_cache(10)
def get_article_from_server(url):
...
代碼解釋：
第 4 行：@timed_lru_cache 裝飾器將支持緩存中條目的生命周期(以秒為單位)和緩存的最大大小；
第 6 行：代碼用 lru_cache 裝飾器包裝瞭裝飾函數，這允許使用 lru_cache 已經提供的緩存功能；
第 7 行和第 8 行：這兩行用兩個表示緩存生命周期和它將過期的實際日期的屬性來修飾函數；
第 12 到 14 行：在訪問緩存中的條目之前，裝飾器檢查當前日期是否超過瞭過期日期，如果是這種情況，那麼它將清除緩存並重新計算生存期和過期日期。
請註意，當條目過期時，此裝飾器如何清除與該函數關聯的整個緩存，生存期適用於整個緩存，而不適用於單個項目，此策略的更復雜實現將根據條目的單個生存期將其逐出。
在程序中，如果想要實現不同緩存策略，可以查看 cachetools 這個庫，該庫提供瞭幾個集合和修飾符，涵蓋瞭一些最流行的緩存策略。
使用新裝飾器緩存文章：
現在可以將新的 @timed_lru_cache 裝飾器與監視器腳本一起使用，以防止每次訪問時獲取文章的內容。為瞭簡單起見，把代碼放在一個腳本中，可以得到以下結果：
import feedparser
import requests
import ssl
import time
from functools import lru_cache, wraps
from datetime import datetime, timedelta
if hasattr(ssl, "_create_unverified_context"):
ssl._create_default_https_context = ssl._create_unverified_context
def timed_lru_cache(seconds: int, maxsize: int = 128):
def wrapper_cache(func):
func = lru_cache(maxsize=maxsize)(func)
func.lifetime = timedelta(seconds=seconds)
func.expiration = datetime.utcnow() + func.lifetime
@wraps(func)
def wrapped_func(*args, **kwargs):
if datetime.utcnow() >= func.expiration:
func.cache_clear()
func.expiration = datetime.utcnow() + func.lifetime
return func(*args, **kwargs)
return wrapped_func
return wrapper_cache
@timed_lru_cache(10)
def get_article_from_server(url):
print("Fetching article from server...")
response = requests.get(url)
return response.text
def monitor(url):
maxlen = 45
while True:
print("\nChecking feed...")
feed = feedparser.parse(url)
for entry in feed.entries[:5]:
if "python" in entry.title.lower():
truncated_title = (
entry.title[:maxlen] + "..."
if len(entry.title) > maxlen
else entry.title
)
print(
"Match found:",
truncated_title,
len(get_article_from_server(entry.link)),
)
time.sleep(5)
monitor("https://realpython.com/atom.xml")
請註意第 30 行如何使用 @timed_lru_cache 裝飾 get_article_from_server() 並指定 10 秒的有效性。在獲取文章後的 10 秒內，任何試圖從服務器訪問同一篇文章的嘗試都將從緩存中返回內容，而不會到達網絡。
運行腳本並查看結果：
$ python monitor.py Checking feed... Fetching article from server... Match found: The Real Python Podcast – Episode #28: Using ... 29521 Fetching article from server... Match found: Python Community Interview With David Amos 54254 Fetching article from server... Match found: Working With Linked Lists in Python 37100 Fetching article from server... Match found: Python Practice Problems: Get Ready for Your ... 164887 Fetching article from server... Match found: The Real Python Podcast – Episode #27: Prepar... 30783 Checking feed... Match found: The Real Python Podcast – Episode #28: Using ... 29521 Match found: Python Community Interview With David Amos 54254 Match found: Working With Linked Lists in Python 37100 Match found: Python Practice Problems: Get Ready for Your ... 164887 Match found: The Real Python Podcast – Episode #27: Prepar... 30783 Checking feed... Match found: The Real Python Podcast – Episode #28: Using ... 29521 Match found: Python Community Interview With David Amos 54254 Match found: Working With Linked Lists in Python 37100 Match found: Python Practice Problems: Get Ready for Your ... 164887 Match found: The Real Python Podcast – Episode #27: Prepar... 30783 Checking feed... Fetching article from server... Match found: The Real Python Podcast – Episode #28: Using ... 29521 Fetching article from server... Match found: Python Community Interview With David Amos 54254 Fetching article from server... Match found: Working With Linked Lists in Python 37099 Fetching article from server... Match found: Python Practice Problems: Get Ready for Your ... 164888 Fetching article from server... Match found: The Real Python Podcast – Episode #27: Prepar... 30783
請註意，代碼在第一次訪問匹配的文章時是如何打印“Fetching article from server…”這條消息的。之後，根據網絡速度和計算能力，腳本將從緩存中檢索文章一兩次，然後再次訪問服務器。
該腳本試圖每 5 秒訪問這些文章，緩存每 10 秒過期一次。對於實際的應用程序來說，這些時間可能太短，因此可以通過調整這些配置來獲得顯著的改進。
五、@lru_cache 裝飾器的官方實現
簡單理解，其實就是一個裝飾器：
def lru_cache(maxsize=128, typed=False):
if isinstance(maxsize, int):
if maxsize < 0:
maxsize = 0
elif callable(maxsize) and isinstance(typed, bool):
user_function, maxsize = maxsize, 128
wrapper = _lru_cache_wrapper(user_function, maxsize, typed, _CacheInfo)
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)
return update_wrapper(wrapper, user_function)
return decorating_function
_CacheInfo = namedtuple("CacheInfo", ["hits", "misses", "maxsize", "currsize"])
def _lru_cache_wrapper(user_function, maxsize, typed, _CacheInfo):
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
cache_len = cache.__len__
lock = RLock() # 因為雙向鏈表的更新不是線程安全的所以需要加鎖
root = [] # 雙向鏈表
root[:] = [root, root, None, None] # 初始化雙向鏈表
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 += 1
return 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 result
misses += 1
result = user_function(*args, **kwds)
with lock:
if key in cache:
pass
elif full:
oldroot = root
oldroot[KEY] = key
oldroot[RESULT] = result
root = oldroot[NEXT]
oldkey = root[KEY]
oldresult = root[RESULT]
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)
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
到此這篇關於Python使用LRU緩存策略進行緩存的方法步驟的文章就介紹到這瞭,更多相關Python LRU緩存 內容請搜索WalkonNet以前的文章或繼續瀏覽下面的相關文章希望大傢以後多多支持WalkonNet！
推薦閱讀：
- python自帶緩存lru_cache用法及擴展的使用
- Python的functools模塊使用及說明
- python 裝飾器重要在哪
- Python 如何手動編寫一個自己的LRU緩存裝飾器的方法實現
- Python 面向切面編程 AOP 及裝飾器