Python 缓存机制之functools.lru

本文主要是介绍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)

输出结果：