Python 缓存机制之functools.lru_cache

2023-12-24 22:08

本文主要是介绍Python 缓存机制之functools.lru_cache,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

说到缓存这个概念,我想大家应该都不陌生 ,以Redis和Memcache为代表的缓存应用基本成为了现在微服务架构的标配了。

事实上,并不是说要用缓存就必须部署Redis等服务,比如在以Python为开发语言的小型单体应用中,我们可以使用functools.lru_cache来实现缓存机制,当然也可以在这个基础上二次封装来满足自己的需求,比如加上缓存过期时间等。

首先,通过一个简单的例子以了解缓存机制的概念,如下代码所示,注意这里只是说下大概的机制,不需要采用这种方法,更优雅的方法是采用functools.lru_cache。

# -*- coding: utf-8 -*-
import random
import datetimeclass MyCache:"""缓存类"""def __init__(self):# 用字典结构以 kv 的形式缓存数据self.cache = {}# 限制缓存的大小,因为缓存的空间有限# 所以当缓存太大时,需要将旧的缓存舍弃掉self.max_cache_size = 10def __contains__(self, key):"""根据该键是否存在于缓存当中返回 True 或者 False"""return key in self.cachedef get(self, key):"""从缓存中获取数据"""data = self.cache[key]data["date_accessed"] = datetime.datetime.now()return data["value"]def add(self, key, value):"""更新该缓存字典,如果缓存太大则先删除最早条目"""if key not in self.cache and len(self.cache) >= self.max_cache_size:self.remove_oldest()self.cache[key] = {'date_accessed': datetime.datetime.now(),'value': value}def remove_oldest(self):"""删除具备最早访问日期的输入数据"""oldest_entry = Nonefor key in self.cache:if oldest_entry is None:oldest_entry = keycontinuecurr_entry_date = self.cache[key]['date_accessed']oldest_entry_date = self.cache[oldest_entry]['date_accessed']if curr_entry_date < oldest_entry_date:oldest_entry = keyself.cache.pop(oldest_entry)@propertydef size(self):"""返回缓存容量大小"""return len(self.cache)if __name__ == '__main__':# 测试缓存功能cache = MyCache()cache.add("test", sum(range(100000)))assert cache.get("test") == cache.get("test")keys = ['red', 'fox', 'fence', 'junk', 'other', 'alpha', 'bravo', 'cal','devo', 'ele']s = 'abcdefghijklmnop'for i, key in enumerate(keys):if key in cache:continueelse:value = ''.join([random.choice(s) for i in range(20)])cache.add(key, value)assert "test" not in cacheprint(cache.cache)

在 Python 的 3.2 +版本中,引入了一个非常优雅的缓存机制,即 functool 模块中的 lru_cache 装饰器,可以直接将函数或类方法的结果缓存住,后续调用则直接返回缓存的结果。lru_cache 原型如下:

@functools.lru_cache(maxsize=None, typed=False)

使用 functools 模块的 lur_cache 装饰器,可以缓存最多 maxsize 个此函数的调用结果,从而提高程序执行的效率,特别适合于耗时的函数。参数 maxsize 为最多缓存的次数,如果为 None,则无限制,设置为 2 的幂时,性能最佳;如果 typed=True(注意,在 functools32 中没有此参数),则不同参数类型的调用将分别缓存,例如 f(3) 和 f(3.0)会分别缓存。

LRU (Least Recently Used,最近最少使用) 算法是一种缓存淘汰策略。其根据数据的历史访问记录来进行淘汰,核心思想是,“如果数据最近被访问过,那么将来被访问的几率也更高”。该算法最初为操作系统中一种内存管理的页面置换算法,主要用于找出内存中较久时间没有使用的内存块,将其移出内存从而为新数据提供空间。其原理就如以上的简单示例。

被 lru_cache 装饰的函数会有 cache_clear 和 cache_info 两个方法,分别用于清除缓存和查看缓存信息

以下为一个简单的 lru_cache 的使用效果,如果函数被调用则会使用print打印一条日志,如果走缓存了则不会。

from functools import lru_cache@lru_cache(None)
def add(x, y):print("calculating: %s + %s" % (x, y))return x + yprint(add(1, 2))
print(add(1, 2))
print(add(2, 3))

运行结果如下:

calculating: 1 + 2
3
3
calculating: 2 + 3
5

从结果可以看出,当第二次调用 add(1, 2) 时,并没有真正执行函数体,而是直接返回缓存的结果。

我们接着来看下maxsize,typed参数的设置和使用,如下代码,maxsize=1代表只缓存一个结果,type=True代表严格区分类型,a=3与a=3.0是不同的场景,会被当做2种结果缓存。

from functools import lru_cache@lru_cache(maxsize=1, typed=True)
def add(x, y):print("calculating: %s + %s" % (x, y))return x + yprint(add(1, 2))
print(add(1, 2))
print(add(2, 3))
print(add(2, 3.0))
print(add(1, 2))
print(add(2, 3))
print(add(2, 3.0))
print(add(2, 3.0))

输出结果:

calculating: 1 + 2
3
3
calculating: 2 + 3
5
calculating: 2 + 3.0
5.0
calculating: 1 + 2
3
calculating: 2 + 3
5
calculating: 2 + 3.0
5.0
5.0

查看函数当前的缓存信息可以使用如下方法,比如查看add函数 :

# 查看函数缓存信息
cache_info = add.cache_info()
print(cache_info)

输出结果类似:hits代表命中缓存次数,misses代表未命中缓存次数,maxsize代表允许的最大存储结果数目,currsize表示当前存储的结果数据。

CacheInfo(hits=3, misses=2, maxsize=1, currsize=1)

如果需要考虑过期时间以及线程安全,可以使用下面的方法,基于collections.OrderedDict()实现的示例。

# -*- coding: UTF-8 -*-
"""
# rs
"""
import collections
import threading
import time# 线程非安全类,未加锁
class LRUCacheNotThreadSafe(object):"""# LRU cache"""def __init__(self, capacity):"""# cache"""self.capacity = capacityself.cache = collections.OrderedDict()def get_and_clear_expired(self, key, current_timestamp):"""# get value and clean expired valued."""try:(value, expire_time) = self.cache.pop(key)if expire_time > current_timestamp:# only when don't expire, we keep this keyself.cache[key] = (value, expire_time)return True, valueexcept KeyError:return False, Nonedef set(self, key, value, expire_time):"""# set value"""try:self.cache.pop(key)except KeyError:if len(self.cache) >= self.capacity:self.cache.popitem(last=False)self.cache[key] = (value, expire_time)# 线程安全类,加锁
class LRUCacheThreadSafe(object):"""LRU cache,clean expired value only when get."""def __init__(self, capacity):"""cache"""self.capacity = capacityself.cache = collections.OrderedDict()self._lock = threading.Lock()def get_and_clear_expired(self, key, current_timestamp):"""# get value and clean expired valued.:param key::param current_timestamp::return:"""with self._lock:try:# 取出key对应的数据(value, expire_time) = self.cache.pop(key)# 如果没有过期,再次写入if expire_time > current_timestamp:self.cache[key] = (value, expire_time)return True, valueexcept KeyError:return False, Nonedef set(self, key, value, expire_time):"""# set value"""with self._lock:try:self.cache.pop(key)except KeyError:# 超出允许缓存的最大结果数目,先删除再写入if len(self.cache) >= self.capacity:self.cache.popitem(last=False)self.cache[key] = (value, expire_time)# 缓存实例
lru_ins = LRUCacheThreadSafe(capacity=50)
# 写入1
lru_ins.set("key1", "value1", int(time.time()))
# 查询1
s, v = lru_ins.get_and_clear_expired("key1", int(time.time()))
print(s, v)# 写入2
lru_ins.set("key1", "value2", int(time.time()))
# 查询2
s, v = lru_ins.get_and_clear_expired("key1", int(time.time()))
print(s, v)

输出结果:

True value1
True value2

参考:

Python 缓存机制与 functools.lru_cache | Huoty's Blog (konghy.cn)

python自带缓存lru_cache用法及扩展(详细)

这篇关于Python 缓存机制之functools.lru_cache的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/533305

相关文章

Python实现终端清屏的几种方式详解

《Python实现终端清屏的几种方式详解》在使用Python进行终端交互式编程时,我们经常需要清空当前终端屏幕的内容,本文为大家整理了几种常见的实现方法,有需要的小伙伴可以参考下... 目录方法一:使用 `os` 模块调用系统命令方法二:使用 `subprocess` 模块执行命令方法三:打印多个换行符模拟

Python实现MQTT通信的示例代码

《Python实现MQTT通信的示例代码》本文主要介绍了Python实现MQTT通信的示例代码,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一... 目录1. 安装paho-mqtt库‌2. 搭建MQTT代理服务器(Broker)‌‌3. pytho

基于Python开发一个图像水印批量添加工具

《基于Python开发一个图像水印批量添加工具》在当今数字化内容爆炸式增长的时代,图像版权保护已成为创作者和企业的核心需求,本方案将详细介绍一个基于PythonPIL库的工业级图像水印解决方案,有需要... 目录一、系统架构设计1.1 整体处理流程1.2 类结构设计(扩展版本)二、核心算法深入解析2.1 自

从入门到进阶讲解Python自动化Playwright实战指南

《从入门到进阶讲解Python自动化Playwright实战指南》Playwright是针对Python语言的纯自动化工具,它可以通过单个API自动执行Chromium,Firefox和WebKit... 目录Playwright 简介核心优势安装步骤观点与案例结合Playwright 核心功能从零开始学习

Python 字典 (Dictionary)使用详解

《Python字典(Dictionary)使用详解》字典是python中最重要,最常用的数据结构之一,它提供了高效的键值对存储和查找能力,:本文主要介绍Python字典(Dictionary)... 目录字典1.基本特性2.创建字典3.访问元素4.修改字典5.删除元素6.字典遍历7.字典的高级特性默认字典

Python自动化批量重命名与整理文件系统

《Python自动化批量重命名与整理文件系统》这篇文章主要为大家详细介绍了如何使用Python实现一个强大的文件批量重命名与整理工具,帮助开发者自动化这一繁琐过程,有需要的小伙伴可以了解下... 目录简介环境准备项目功能概述代码详细解析1. 导入必要的库2. 配置参数设置3. 创建日志系统4. 安全文件名处

使用Python构建一个高效的日志处理系统

《使用Python构建一个高效的日志处理系统》这篇文章主要为大家详细讲解了如何使用Python开发一个专业的日志分析工具,能够自动化处理、分析和可视化各类日志文件,大幅提升运维效率,需要的可以了解下... 目录环境准备工具功能概述完整代码实现代码深度解析1. 类设计与初始化2. 日志解析核心逻辑3. 文件处

python生成随机唯一id的几种实现方法

《python生成随机唯一id的几种实现方法》在Python中生成随机唯一ID有多种方法,根据不同的需求场景可以选择最适合的方案,文中通过示例代码介绍的非常详细,需要的朋友们下面随着小编来一起学习学习... 目录方法 1:使用 UUID 模块(推荐)方法 2:使用 Secrets 模块(安全敏感场景)方法

使用Python删除Excel中的行列和单元格示例详解

《使用Python删除Excel中的行列和单元格示例详解》在处理Excel数据时,删除不需要的行、列或单元格是一项常见且必要的操作,本文将使用Python脚本实现对Excel表格的高效自动化处理,感兴... 目录开发环境准备使用 python 删除 Excphpel 表格中的行删除特定行删除空白行删除含指定

Python通用唯一标识符模块uuid使用案例详解

《Python通用唯一标识符模块uuid使用案例详解》Pythonuuid模块用于生成128位全局唯一标识符,支持UUID1-5版本,适用于分布式系统、数据库主键等场景,需注意隐私、碰撞概率及存储优... 目录简介核心功能1. UUID版本2. UUID属性3. 命名空间使用场景1. 生成唯一标识符2. 数