在 Python 中使用 LRU 缓存策略

比眉伴天荒 2023-10-04 15:32 3阅读 0赞

有很多方法可以实现快速响应的应用程序。**缓存**是一种方法，如果使用得当，它可以使事情变得更快，同时减少计算资源的负载。Python 的[`functools`模块][functools]带有[`@lru_cache`装饰器][lru_cache]，它使您能够使用**最近最少使用 (LRU) 策略**缓存函数的结果。这是一种简单而强大的技术，您可以使用它来利用代码中的缓存功能。

**在本教程中，您将学习：**

*  有哪些**缓存策略**可用以及如何使用**Python 装饰器**实现它们
 *  **LRU 策略**是什么以及它是如何工作的
 *  如何通过**`@lru_cache`****装饰器**缓存来提高性能
 *  如何扩展`@lru_cache`装饰器的功能并使其在特定时间后**过期**

在本教程结束时，您将更深入地了解缓存的工作原理以及如何在 Python 中利用它。

### **缓存及其用途** ###

**缓存**是一种优化技术，您可以在应用程序中使用它来将最近或经常使用的数据保存在比其源访问速度更快或计算成本更低的内存位置。

想象一下，您正在构建一个新闻阅读器应用程序，该应用程序从不同来源获取最新新闻。当用户浏览列表时，您的应用程序会下载文章并将它们显示在屏幕上。

如果用户决定在几篇新闻文章之间反复来回移动会发生什么？除非您正在缓存数据，否则您的应用程序每次都必须获取相同的内容！这会使您的用户系统变得迟钝，并对托管文章的服务器造成额外压力。

更好的方法是在获取每篇文章后将内容存储在本地。然后，下次用户决定打开文章时，您的应用程序可以从本地存储的副本中打开内容，而不是返回源。在计算机科学中，这种技术称为**缓存**。

#### 使用 Python 字典实现缓存 ####

您可以使用[字典][Link 1]在 Python 中实现缓存解决方案。

继续以新闻阅读器为例，无需每次需要下载文章时都直接访问服务器，您可以检查缓存中是否有内容，只有在没有时才返回服务器。您可以将文章的 URL 用作键，并将其内容用作值。

以下是这种缓存技术的外观示例：

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://realpython.com/sorting-algorithms-python/")
    get_article("https://realpython.com/sorting-algorithms-python/")

这段代码保存到一个`caching.py`文件，[安装][Link 2]的[`requests`库][requests]，然后运行该脚本：

$ pip install requests
    $ python caching.py
    Getting article...
    Fetching article from server...
    Getting article...

请注意，尽管在第 17 行和第 18 行中调用了两次`get_article()`，但字符串`"Fetching article from server..."` [打印][Link 3]一次。发生这种情况是因为在第一次访问文章后，您将其 URL 和内容放入`cache`字典中。第二次，代码不需要再次从服务器获取项目。

#### 缓存策略 ####

这种缓存实现有一个大问题：字典的内容会无限增长！随着用户下载更多文章，应用程序将不断将它们存储在内存中，最终导致应用程序崩溃。

要解决此问题，您需要一种策略来决定哪些文章应该保留在内存中，哪些应该删除。这些缓存策略是专注于管理缓存信息并选择丢弃哪些项目以为新项目腾出空间的算法。

您可以使用多种不同的策略从缓存中驱逐项目，并防止其超过最大大小。以下是最受欢迎的五个，并解释了每个最有用的时间：

<table style="width:690px;"> 
 <thead> 
 <tr> 
 <th style="vertical-align:bottom;">战略</th> 
 <th style="vertical-align:bottom;">驱逐政策</th> 
 <th style="vertical-align:bottom;">用例</th> 
 </tr> 
 </thead> 
 <tbody> 
 <tr> 
 <td style="vertical-align:top;">先进先出 (FIFO)</td> 
 <td style="vertical-align:top;">驱逐最旧的条目</td> 
 <td style="vertical-align:top;">较新的条目最有可能被重用</td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;">后进先出 (LIFO)</td> 
 <td style="vertical-align:top;">驱逐最新的条目</td> 
 <td style="vertical-align:top;">较旧的条目最有可能被重用</td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;">最近最少使用 (LRU)</td> 
 <td style="vertical-align:top;">驱逐最近最少使用的条目</td> 
 <td style="vertical-align:top;">最近使用的条目最有可能被重用</td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;">最近使用 (MRU)</td> 
 <td style="vertical-align:top;">驱逐最近使用的条目</td> 
 <td style="vertical-align:top;">最近最少使用的条目最有可能被重用</td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;">最不常用 (LFU)</td> 
 <td style="vertical-align:top;">驱逐最不常访问的条目</td> 
 <td style="vertical-align:top;">命中率高的条目更有可能被重用</td> 
 </tr> 
 </tbody> 
</table>

在下面的部分中，您将仔细研究 LRU 策略以及如何使用`@lru_cache`Python`functools`模块中的装饰器来实现它。

#### 深入研究最近最少使用 (LRU) 缓存策略 ####

使用 LRU 策略实现的缓存按使用顺序组织其项目。每次访问一个条目时，LRU 算法都会将它移动到缓存的顶部。这样，算法可以通过查看列表底部来快速识别未使用时间最长的条目。

下图显示了用户从网络请求文章后的假设缓存表示：

[![从网络访问时，项如何插入 LRU 缓存中][LRU]][LRU _LRU]

请注意缓存如何在将文章提供给用户之前将文章存储在最近的插槽中。下图显示了当用户请求第二篇文章时发生的情况：

[![从网络访问时，项如何插入 LRU 缓存中][LRU 1]][LRU _LRU 1]

第二篇文章占据最近的位置，将第一篇文章推到列表中。

LRU 策略假设对象使用得越近，将来就越有可能需要它，因此它会尝试将该对象保留在缓存中的时间最长。

#### 窥视 LRU 缓存的幕后 ####

在 Python 中实现 LRU 缓存的一种方法是使用[双向链表][Link 4]和[哈希映射的组合][Link 5]。双向链表的**头部元素**将指向最近使用的条目，**尾部**将指向最近最少使用的条目。

下图展示了 LRU 缓存实现背后的潜在结构：

[![LRU缓存背后的数据结构][LRU 2]][LRU_LRU 2]

使用哈希映射，您可以通过将每个条目映射到双向链表中的特定位置来确保访问缓存中的每个项目。

这个策略非常快。访问最近最少使用的项目和更新缓存是运行时间为*O* (1) 的操作。

**注意：**要更深入地了解[Big O 表示法][Big O]，以及 Python 中的几个实际示例，请查看[Big O 表示法和 Python 示例的算法分析][Big O _ Python]。

从 3.2 版本开始，Python 包含了`@lru_cache`用于实现 LRU 策略的装饰器。您可以使用此装饰器来包装函数并将其结果缓存到最大条目数。

### **在 Python 中使用 @lru\_cache 实现 LRU 缓存** ###

就像您之前实现的缓存解决方案一样，`@lru_cache`在幕后使用字典。它将函数的结果缓存在一个键下，该键由对函数的调用组成，包括提供的参数。这很重要，因为这意味着这些参数必须是可**散列**的，装饰器才能工作。

#### 玩楼梯 ####

想象一下，您想通过一次跳一个、两个或三个楼梯来确定到达楼梯中特定楼梯的所有不同方式。到第四层楼梯有多少条路？以下是所有不同的组合：

[![组合跳跃到达第 7 层][7]][7 _7]

您可以通过说明，要到达当前的楼梯，您可以从下面的一个楼梯、两个楼梯或三个楼梯跳下，来构建此问题的解决方案。将您可以用来到达每个点的跳跃组合的数量相加，应该可以为您提供到达当前位置的可能方法总数。

例如，到达第四个楼梯的组合数将等于您到达第三个、第二个和第一个楼梯的不同方式的总数：

[![到达第 4 级楼梯的步骤][4]][4 _4]

如图所示，有七种不同的方式可以到达第四层。请注意给定楼梯的解决方案如何建立在较小子问题的答案上。在这种情况下，要确定到第四个楼梯的不同路径，您可以将到达第三个楼梯的四种方式、两种到达第二个楼梯的方式和一种到达第一个楼梯的方式相加。

这种方法称为**递归**。如果您想了解更多信息，请查看[Python][]中的[递归思考][Python]以了解该主题的介绍。

这是一个实现此递归的函数：

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][timeit_module]。在最后后添加以下几行：

setup_code = "from __main__ import steps_to"
    stmt = "steps_to(30)"
    times = repeat(setup=setup_code, stmt=stmt, repeat=3, number=10)
    print(f"Minimum execution time: {
            min(times)}")

您还需要[导入][Link 6]的`timeit`在代码的顶部模块：

from timeit import repeat

以下是修改后的完整代码：

from timeit import repeat
    
    
    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))
    setup_code = "from __main__ import steps_to"
    stmt = "steps_to(30)"
    times = repeat(setup=setup_code, stmt=stmt, repeat=3, number=10)
    print(f"Minimum execution time: {min(times)}")

以下是对这些添加内容的逐行解释：

*  **第 35 行**导入了 的名称，`steps_to()`以便`timeit.repeat()`知道如何调用它。
 *  **第 36 行**准备调用具有您想要到达的楼梯编号的函数，在本例中为`30`。这是将要执行和计时的语句。
 *  **第 37 行**调用`timeit.repeat()`设置代码和语句。这将调用函数`10`时间，返回每次执行所用的秒数。
 *  **第 38 行**标识并打印返回的最短时间。

**注意：**一个常见的误解是您应该找到函数每次运行的平均时间，而不是选择最短时间。

时间测量是嘈杂的，因为系统同时运行其他进程。最短的时间总是最少的噪音，使其成为函数运行时的最佳表示。

现在再次运行脚本：

$ python stairs.py
    53798080
    Minimum execution time: 40.014977024000004

您将看到的秒数取决于您的特定硬件。在我的系统上，脚本需要四十秒，这对于三十个楼梯来说是相当慢的！

**注意：**您可以`timeit`在[官方 Python 文档中][timeit_module]了解有关该模块的更多信息。

需要这么长时间的解决方案是一个问题，但您可以使用记忆来改进它。

#### 使用记忆来改进解决方案 ####

这种递归实现通过将问题分解为相互建立的更小的步骤来解决问题。下图显示了一个树，其中每个节点代表对 的特定调用`steps_to()`：

[![我们有多少种方式可以到达第七层？][345e1ac64341f6b45b9f90f5655f559b.png]][345e1ac64341f6b45b9f90f5655f559b.png 1]

请注意您如何需要`steps_to()`多次使用相同的参数进行调用。例如，`steps_to(5)`计算了两次，`steps_to(4)`计算了四次、`steps_to(3)`七次和`steps_to(2)`六次。多次调用同一个函数会增加不必要的计算周期——结果总是相同的。

要解决此问题，您可以使用一种称为**memoization**的技术。这种方法通过将其结果存储在内存中，然后在需要时稍后引用它来确保函数不会针对相同的输入多次运行。这个场景听起来像是使用 Python`@lru_cache`装饰器的绝佳机会！

**注意：**有关**memoization**和`@lru_cache`用于实现它的更多信息，请查看[Python 中的 Memoization][Python _ Memoization]。

只需进行两项更改，您就可以显着改善算法的运行时间：

1.  从`functools`模块导入`@lru_cache`装饰器。
2.  使用`@lru_cache`装饰`steps_to()`。

以下是两个更新后脚本顶部的样子：

from functools import lru_cache
     from timeit import repeat
     
     @lru_cache(maxsize=None)
     def steps_to(stair):
         if stair == 1:

以下是修改后的完整代码：

from functools import lru_cache
    from timeit import repeat
    
    
    @lru_cache(maxsize=None)
    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))
    setup_code = "from __main__ import steps_to"
    stmt = "steps_to(30)"
    times = repeat(setup=setup_code, stmt=stmt, repeat=3, number=10)
    print(f"Minimum execution time: {min(times)}")

运行更新后的脚本会产生以下结果：

$ python stairs.py
    53798080
    Minimum execution time: 7.999999999987184e-07

缓存函数的结果将运行时间从 40 秒缩短到 0.0008 毫秒！这是一个了不起的改进！

**注意：**在[Python 3.8][]及更高版本中，`@lru_cache`如果您未指定任何参数，则可以使用不带括号的装饰器。在以前的版本中，您可能需要包含括号：`@lru_cache()`.

请记住，在幕后，`@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`条目。当新的调用进来时，装饰器的实现将驱逐现有`16`条目中最近最少使用的条目，为新项目腾出空间。

要查看代码中添加的这个新内容会发生什么，您可以使用`cache_info()`由`@lru_cache`装饰器提供的来检查**命中**和**未****命中**的数量以及缓存的当前大小。为清楚起见，删除对函数运行时间进行计时的代码。以下是所有修改后最终脚本的外观：

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`**是缓存的当前大小。在这种情况下，它表明您的缓存已满。

如果您需要从缓存中删除所有条目，那么您可以使用`cache_clear()`提供的`@lru_cache`.

### **添加缓存过期** ###

假设您要开发一个脚本来监视*Real Python*并打印包含单词 的任何文章中的字符数`python`。

*Real Python*提供了一个[Atom feed][]，因此您可以像以前一样使用该`feedparser`库来解析提要并使用该`requests`库来加载文章的内容。

下面是监控脚本的一个实现：

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`和`requests`库，然后运行该脚本。它将持续运行，直到您通过在终端窗口中按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`，代码加载并解析来自*Real Python*的提要。
 *  **第 20 行**：循环遍历`5`列表中的第一个条目。
 *  **第 21 行到第 31 行**：如果这个词`python`是标题的一部分，那么代码会将它与文章的长度一起打印出来。
 *  **第33行**：代码[睡][Link 7]了`5`秒钟，然后再继续。
 *  **第 35**行：该行通过将*Real Python*提要的 URL 传递到 来启动监视过程`monitor()`。

每次脚本加载一篇文章时，都会将消息`"Fetching article from server..."`打印到控制台。如果让脚本运行足够长的时间，那么即使加载相同的链接，您也会看到此消息如何重复显示。

**注意：**有关`feedparser`访问通过 HTTPS提供的内容的问题的更多信息，请查看存储库上的[问题 84][84]`feedparser`。[PEP 476][]描述了 Python 如何在默认情况下为`stdlib`HTTP 客户端启用证书验证，这是导致此错误的根本原因。

这是缓存文章内容并避免每五秒访问一次网络的绝佳机会。您可以使用`@lru_cache`装饰器，但是如果文章的内容更新了会发生什么？

第一次访问文章时，装饰器将存储其内容并在每次之后返回相同的数据。如果帖子被更新，那么监视器脚本将永远不会意识到它，因为它将拉取存储在缓存中的旧副本。要解决此问题，您可以将缓存条目设置为过期。

#### 基于时间和空间驱逐缓存条目 ####

`@lru_cache`只有当没有更多空间来存储新列表时，装饰器才会驱逐现有条目。如果有足够的空间，缓存中的条目将永远存在并且永远不会被刷新。

这会给您的监控脚本带来问题，因为您永远不会获取为以前缓存的文章发布的更新。要解决此问题，您可以更新缓存实现，使其在特定时间后过期。

您可以将这个想法实现到一个扩展的新装饰器中`@lru_cache`。如果调用者试图访问超过其生命周期的项目，则缓存将不会返回其内容，从而迫使调用者从网络中获取文章。

**注意：**有关 Python 装饰器的更多信息，请查看[Python Decorators][]和[Python Decorators 101][][上的 Primer][Python Decorators]。

这是这个新装饰器的可能实现：

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

这是此实现的细分：

*  **第 4 行**：`@timed_lru_cache`装饰器将支持缓存中条目的生命周期（以秒为单位）和缓存的最大大小。
 *  **第 6 行**：代码用`lru_cache`装饰器包装了被装饰的函数。这允许您使用已经提供的缓存功能`lru_cache`。
 *  **第 7 行和第 8 行**：这两行用两个属性来检测装饰函数，它们代表缓存的生命周期和实际到期日期。
 *  **第 12 到 14 行**：在访问缓存中的条目之前，装饰器检查当前日期是否已超过到期日期。如果是这种情况，则它会清除缓存并重新计算生存期和到期日期。

请注意，当条目过期时，此装饰器如何清除与该函数关联的整个缓存。生命周期适用于整个缓存，而不是单个文章。此策略的更复杂实现将根据条目的个人生命周期驱逐条目。

#### 使用新装饰器缓存文章 ####

您现在可以将新的`@timed_lru_cache`装饰器与`monitor`脚本一起使用，以防止每次访问文章时都获取它的内容。

为简单起见，将代码放在一个脚本中，最终得到以下结果：

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 行如何`get_article_from_server()`用`@timed_lru_cache`和 指定`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`。这些时间对于实际应用程序来说可能太短了，因此您可以通过调整这些配置来获得显着的改进。

### **结论** ###

缓存是提高任何软件系统性能的基本优化技术。了解缓存的工作原理是将其有效地整合到您的应用程序中的基本步骤。

**在本教程中，您学习了：**

*  不同的**缓存策略**是什么以及它们是如何工作的
 *  如何使用 Python 的**`@lru_cache`**装饰器
 *  如何**创建一个新的装饰器**来扩展功能`@lru_cache`
 *  如何使用模块测量代码的**运行时间**`timeit`
 *  什么是**递归**以及如何使用它解决问题
 *  **记忆化**如何通过将中间结果存储在内存中来改善运行时间

在应用程序中实现不同缓存策略的下一步是查看[`cachetools`][cachetools]模块。这个库提供了几个集合和装饰器，涵盖了一些你可以立即开始使用的最流行的缓存策略。

[functools]: https://docs.python.org/3/library/functools.html
[lru_cache]: https://docs.python.org/3/library/functools.html#functools.lru_cache
[Link 1]: https://realpython.com/python-dicts/
[Link 2]: https://realpython.com/what-is-pip/#installing-packages-with-pip
[requests]: https://realpython.com/python-requests/
[Link 3]: https://realpython.com/python-print/
[LRU]: https://img-blog.csdnimg.cn/img_convert/23667bc4de25f875c598c3f14b999cc8.png
[LRU _LRU]: https://files.realpython.com/media/lru_cache_1_1.2eb80a8b24a3.png
[LRU 1]: https://img-blog.csdnimg.cn/img_convert/bb273f1238e7709bb1368c5748afb807.png
[LRU _LRU 1]: https://files.realpython.com/media/lru_cache_2_1.8c4f225e79d0.png
[Link 4]: https://realpython.com/linked-lists-python/#how-to-use-doubly-linked-lists
[Link 5]: https://realpython.com/python-data-structures/#dictionaries-maps-and-hash-tables
[LRU 2]: https://img-blog.csdnimg.cn/img_convert/9fa5fb584be378062669960dd9878198.png
[LRU_LRU 2]: https://files.realpython.com/media/cache_internal_representation_1.6fdd3a39fa28.png
[Big O]: https://realpython.com/sorting-algorithms-python/#measuring-efficiency-with-big-o-notation
[Big O _ Python]: https://stackabuse.com/big-o-notation-and-algorithm-analysis-with-python-examples/
[7]: https://img-blog.csdnimg.cn/img_convert/8eb90c1f4bde35058d2130ca9b140e61.png
[7 _7]: https://files.realpython.com/media/7_ways_to_make_it_to_4th_stair.2dea6527a4b7.png
[4]: https://img-blog.csdnimg.cn/img_convert/8c1cb3cb78f8d13933de5ee79264e84d.png
[4 _4]: https://files.realpython.com/media/combination-jumps-to-7th-stair-lru-cache_1.385e09435b81.png
[Python]: https://realpython.com/python-thinking-recursively/
[timeit_module]: https://docs.python.org/3/library/timeit.html
[Link 6]: https://realpython.com/python-import/
[345e1ac64341f6b45b9f90f5655f559b.png]: https://img-blog.csdnimg.cn/img_convert/345e1ac64341f6b45b9f90f5655f559b.png
[345e1ac64341f6b45b9f90f5655f559b.png 1]: https://files.realpython.com/media/tree.3ebda400089a.png
[Python _ Memoization]: https://towardsdatascience.com/memoization-in-python-57c0a738179a
[Python 3.8]: https://realpython.com/python38-new-features/
[Atom feed]: https://en.wikipedia.org/wiki/Atom_%28Web_standard%29
[HTTPS]: https://realpython.com/python-https/
[Link 7]: https://realpython.com/python-sleep/
[84]: https://github.com/kurtmckee/feedparser/issues/84
[PEP 476]: https://www.python.org/dev/peps/pep-0476/
[Python Decorators]: https://realpython.com/primer-on-python-decorators/
[Python Decorators 101]: https://realpython.com/courses/python-decorators-101/
[cachetools]: https://github.com/tkem/cachetools/