Python使用LRU快取策略進行快取的方法步驟

2022-05-27 18:05:13

一、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快取內容請搜尋it145.com以前的文章或繼續瀏覽下面的相關文章希望大家以後多多支援it145.com！