在Python中进行实现LRU缓存策略

6个月前 471次点击 来自 其他

收录专题: Python边学边译

标签: Python

原文链接:Caching in Python Using the LRU Cache Strategy

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

在本教程中,您将学习:

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

缓存及其用途

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

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

如果用户决定在几则新闻文章之间来回反复阅读,将会发生什么?除非您缓存数据,否则您的应用每次都必须获取相同内容!那会使您的用户系统呆滞,并给托管文章的服务器带来额外压力。

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

使用Python字典实现缓存

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

这是此缓存技术的示例:

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文件,安装requests,然后运行该脚本:

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

请注意,尽管在第17行和第18行中调用了两次,但是第一次访问该文章之后,您将其URL和内容放入了cache中。第二次访问则直接从cache中读取数据。

缓存策略

此缓存实现存在一个大问题:字典的内容将无限增长!当用户下载更多文章时,应用程序将继续将它们存储在内存中,最终导致应用程序崩溃。

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

您可以使用几种不同的策略从缓存中删除项目,并防止其超出其最大大小。以下是五种最受欢迎的方法,并说明每种方法何时最有用:

策略名 删除策略 用例
先进先出(FIFO) 删除最旧的条目 较新的条目最有可能被重用
后进先出(LIFO) 删除最新条目 较旧的条目最有可能被重用
最近最少使用(LRU) 删除最近最少使用的条目 最近使用的条目最有可能被重用
最近使用(MRU) 删除最近使用的条目 最近使用最少的条目最有可能被重用
最少使用(LFU) 删除访问最少的条目 命中率很高的条目更有可能被重用

深入了解最近最少使用(LRU)缓存策略

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

下图显示了用户从网络请求文章后的假想缓存表示形式:

How are items inserted in the LRU Cache as they are accessed from the network

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

How are items inserted in the LRU Cache as they are accessed from the network

第二篇文章占据最新位置,将第一篇文章推入列表。

LRU策略假定使用的对象越新,将来使用它的可能性就越大,因此它尝试将该对象保留在高速缓存中的时间最长。

LRU Cache技术一窥

在Python中实现LRU缓存的一种方法是使用双链表和哈希图的组合。双向链表的头元素将指向最近使用的条目,尾部将指向最近使用最少的条目。

下图显示了幕后LRU缓存实现的潜在结构:

Data structure behind the LRU cache

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

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

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

使用@lru_cache在Python中实现LRU缓存

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

爬楼梯

想象一下,您想通过一次跳一个,两个或三个楼梯来确定可以到达楼梯中特定楼梯的所有不同方式。那请问,到第四层有几种组合方式?以下是所有不同的组合:

Combination of jumps to get to the 7th stair

您可以声明该问题的解决方案,指出要到达当前的楼梯,可以从下面的一个楼梯,两个楼梯或三个楼梯上跳下来。累加可用于到达每个点的跳跃组合的数量,应为您提供达到当前位置的可能方法的总数。

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

Steps to reach the 4th stair

如图所示,有七种方法可以到达第四阶梯。请注意,给定楼梯的解决方案如何以较小子问题的答案为基础。在这种情况下,要确定通向第四层楼梯的不同路径,可以将到达第三层的四种方式,到达第二层的两种方式以及到达第一层的一种方式加起来。

这种方法称为递归,这是实现此递归的函数:

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

使用备忘录(Memoization)来改善解决方案

递归实现通过将问题分解为相互重叠的较小步骤来解决该问题。

假设,我们现在开始爬10层楼梯。下图显示了一个树,其中每个节点代表对以下对象的特定调用steps_to()

How many ways we can reach the seventh stair?

注意,您需要多次使用相同的参数进行调用steps_to()。例如,steps_to(5)被计算两次,steps_to(4)被计算四次,steps_to(3)七次和steps_to(2)六次。多次调用同一个函数会增加不必要的计算周期,结果将始终相同。

要解决此问题,可以使用一种称为备忘录的技术。这种方法通过将函数的结果存储在内存中,然后在必要时稍后对其进行引用,从而确保函数不会针对相同的输入多次运行。这种情况听起来像是使用@lru_cache装饰器的绝好机会!

只需进行两项更改,就可以大大改善算法的运行时间:

  1. functools模块导入装饰器@lru_cache
  2. 使用@lru_cache装饰steps_to()
from functools import lru_cache

@lru_cache
def steps_to(stair):
    if stair == 1:
...

@lru_cache装饰器提供了一个maxsize属性,该属性定义了在缓存开始删除旧项目之前的最大条目数。默认情况下,maxsize设置为128。如果设置maxsize为None,则缓存将无限期增长,并且不会删除任何条目。如果您要在内存中存储大量不同的调用,这可能会成为问题。

这是@lru_cache使用maxsize属性的示例:

from functools import lru_cache

@lru_cache(maxsize=16)
def steps_to(stair):
    if stair == 1:

要查看新添加的代码会发生什么,可以使用装饰器cache_info()提供的@lru_cache来检查命中和未命中的数量以及当前的缓存大小。

from functools import lru_cache

@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(10))
print(steps_to.cache_info())

如果再次调用该脚本,则会看到以下结果:

$ python stairs.py
274
CacheInfo(hits=12, misses=10, maxsize=16, currsize=10)

添加缓存过期

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

Real Python提供了Atom提要,因此您可以像以前一样使用该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

每次脚本加载文章时,该消息"Fetching article from server..."都会打印到控制台。如果让脚本运行足够长的时间,那么即使加载相同的链接,屏幕也会重复显示此消息,这意味着每次程序都尝试了联网获取文章数据。

接下来,我们将考虑使用@lru_cache装饰器缓存文章,并基于一定规则更新或删除文章。

基于时间和空间删除缓存条目

即在一定时间内,例如我们认为在10秒钟内,文章都是最新的,如果缓存中保存的文章时间大于10秒,我们则联网更新数据并同时更新本地缓存。

我们将设计一个@timed_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装饰器与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")

get_article_from_server()使用@timed_lru_cache修饰器,其指定10秒数的有效期。程序尝试从服务器中获取同一文章的任何操作,若在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

结论

高速缓存是提高任何软件系统性能的基本优化技术,了解缓存的工作原理,至关重要。

若您的程序中需要使用其他几种缓存策略,请参考使用 cachetools ,该库提供了几个集合和修饰符,涵盖了一些最流行的缓存策略。

Card image cap
开发者雷

尘世间一个小小的开发者,每天增加一些无聊的知识,就不会无聊了

要加油~~~

技术文档 >> 系列应用 >>
热推应用
Let'sLearnSwift
学习Swift的入门教程
PyPie
Python is as good as Pie
标签