Python

Python 缓存函数 lru_cache + 过期时间 + hash对象支持

Python | 缓存 | 函数 | cache | hash | 超时缓存

Posted by luoruiqing on December 9, 2021

functools.lru_cache

一个为函数提供缓存功能的装饰器，当下次以相同参数调用函数时直接返回上一次的结果。用以节约高开销或I/O函数的调用时间。

参数解释

  • maxsize=128 : 用于控制被装饰的方法最大可缓存结果数量，当超出数量外则按照 lru 规则剔除不常用的缓存；当设置为 None 时将取消缓存上限控制。
  • typed=False : 当设置为 True 时，同值但类型不同参数将被分别缓存。例如 f(3) 和 f(3.0) 将被视为不同的参数来分别缓存。

用法样例

1
2
3
4
5
6
7
8
9
10
11

from functools import lru_cache


@lru_cache()
def square(x):
    return x ** 2

# 查看缓存信息
square.cache_info()
# 清除所有缓存
square.cache_clear()

注意

  1. 需要 Python Version > 3.2
  2. 由于使用了字典存储缓存，所以该函数的固定参数和关键字参数必须是可哈希的（不能传入 dict/list/set类型）。
  3. 不同模式的参数可能被视为不同从而产生多个缓存项，例如, f(a=1, b=2) 和 f(b=2, a=1) 因其参数顺序不同，可能会被缓存两次。
  4. 计算类型 Decimal(42) 与 Fraction(42) 将会分别计算缓存，但是作为元素在元组内时， (‘answer’, Decimal(42)) 与 (‘answer’, Fraction(42)) 将会命中同一缓存。
  5. 缓存的数据存在于Python运行环境的内存中，请注意内存用量。

使用的方法是非常简单的，可以在很多的业务场景内生效，例如频繁的打开文件时，筛选器的值基本不变时候等等。

扩展过期时间

方法本身的功能已经很好了，但美中不足的是没有过期时间，在业务开发时经常希望让缓存过期，以便自动获取数据库的新数据。

实现

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

import time  # monotonic_ns # need py >= 3.7
import functools


def lru_cache(timeout: int, *lru_args, **lru_kwargs):

    def wrapper_cache(func):
        func = functools.lru_cache(*lru_args, **lru_kwargs)(func)
        func.delta = timeout
        func.expiration = time.monotonic() + func.delta

        @functools.wraps(func)
        def wrapped_func(*args, **kwargs):
            if time.monotonic() >= func.expiration:
                func.cache_clear()
                func.expiration = time.monotonic() + func.delta
            return func(*args, **kwargs)

        wrapped_func.cache_info = func.cache_info
        wrapped_func.cache_clear = func.cache_clear
        return wrapped_func

    return wrapper_cache

这里使用了单调时钟 time.monotonic(), 程序会更加稳定健壮。

当你使用的 Python >= 3.7 的版本时，可以使用 time.monotonic_ns() 来增加清除时间的准确性。

测试

1
2
3
4
5

# 10分钟全部清除一次
@lru_cache(60 * 10)
def square(x):
    return x ** 2
# 与 functools.lru_cache 的 API 一致。

扩展参数类型

当调用方法的参数是 list 或 dict 或 set 的时候，该方法就无法使用了，因为动态的数据类型是无法 hash 的， 试试这个例子

测试

1
2
3
4
5
6
7
8
9
10

@lru_cache()
def square(x, log=None):
    print('没命中缓存，要计算了: ', log) # 参数只用于打印
    return x ** 2

square(2)
# 没命中缓存，要计算了:  None
# Out[41]: 4
square(2, {})
# TypeError: unhashable type: 'dict'

包含这些常用的动态变化的数据结构，都会引发 TypeError: unhashable type: ‘xxx’ 的错误！

实现

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

import functools
import json


HashDict = type('HashDict', (dict,), {'__hash__': lambda self: hash(json_hash(self))})
HashList = type('HashList', (list,), {'__hash__': lambda self: hash(json_hash(self))})
HashSet = type('HashSet', (set,), {'__hash__': lambda self: hash(json_hash(self))})


def json_hash(o, *args, **kwargs):
    return json.dumps(o, *args, **dict({"sort_keys": True, "default": repr}, **kwargs))


def to_hash_object(obj, ):
    if isinstance(obj, dict):
        return HashDict(obj)
    elif isinstance(obj, list):
        return HashList(obj)
    elif isinstance(obj, set):
        return HashSet(obj)
    return obj


def lru_cache(*lru_args, **lru_kwargs):
    def wrapper_cache(func):
        func = functools.lru_cache(*lru_args, **lru_kwargs)(func)

        @functools.wraps(func)
        def wrapped_func(*args, **kwargs):
            return func(*map(to_hash_object, args), **{k: to_hash_object(v) for k, v in kwargs.items()})
        return wrapped_func
    return wrapper_cache

注意

  • 如果参数的结构非常复杂，可能依然会发生错误.
  • 如果关心dict的顺序，需要去掉 sort_keys 参数
  • json_hash 中依赖 repr, 如果传入特殊的对象，注意 repr 的返回值来避免命中错误的缓存
  • 注意返回的对象尽量 clone 后使用，小心对象引用的错误！

测试2

1
2
3
4
5
6
7
8
9
10

In [289]: square(2, [{1: [1, 2, 3]}])
没命中缓存，要计算了:  [{1: [1, 2, 3]}]
Out[289]: 4

In [290]: square(2, [{1: [1, 2, 2, 3]}])
没命中缓存，要计算了:  [{1: [1, 2, 2, 3]}]
Out[290]: 4

In [291]: square(2, [{1: [1, 2, 2, 3]}])
Out[291]: 4

综合扩展

这个例子既包含超时过期， 也包含动态数据类型的兼容

实现

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41

import functools
import json
import time  # monotonic_ns # need py >= 3.7


HashDict = type('HashDict', (dict,), {'__hash__': lambda self: hash(json_hash(self))})
HashList = type('HashList', (list,), {'__hash__': lambda self: hash(json_hash(self))})
HashSet = type('HashSet', (set,), {'__hash__': lambda self: hash(json_hash(self))})


def json_hash(o, *args, **kwargs):
    return json.dumps(o, *args, **dict({"sort_keys": True, "default": repr}, **kwargs))


def to_hash_object(obj, ):
    if isinstance(obj, dict):
        return HashDict(obj)
    elif isinstance(obj, list):
        return HashList(obj)
    elif isinstance(obj, set):
        return HashSet(obj)
    return obj


def lru_cache(timeout: int, *lru_args, **lru_kwargs):
    def wrapper_cache(func):
        func = functools.lru_cache(*lru_args, **lru_kwargs)(func)
        func.delta = timeout
        func.expiration = time.monotonic() + func.delta

        @functools.wraps(func)
        def wrapped_func(*args, **kwargs):
            if time.monotonic() >= func.expiration:
                func.cache_clear()
                func.expiration = time.monotonic() + func.delta
            return func(*map(to_hash_object, args), **{k: to_hash_object(v) for k, v in kwargs.items()})

        wrapped_func.cache_info = func.cache_info
        wrapped_func.cache_clear = func.cache_clear
        return wrapped_func
    return wrapper_cache

关于hash

这部分内容是在扩展参数类型时候，尝试各种方式的过程：

lru_cache 方法会 hash 每个传入的位置参数和命名参数。用来命中缓存， 那动态数据类型则无法 hash 了吗？ 扩展考虑到下面的：

  1. 不可变的集合才能拥有 hash， 可以使用 frozenset 来固定参数并获取 hash
  2. hash 方法本身调用的都是 Python 的内置方法 __hash__

frozenset

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

In [162]: hash(frozenset({'1': 1, '2': 2}.items()))
Out[162]: -4104304231929626832

In [163]: hash(frozenset({'2': 2, '1': 1}.items()))
Out[163]: -4104304231929626832

In [164]: hash(frozenset([1, 2, 3]))
Out[164]: -7699079583225461316

In [165]: hash(frozenset([1, 2, 2, 3]))
Out[165]: -7699079583225461316

In [167]: hash(frozenset([{'1': 1}, ]))
---------------------------------------------------------------------------
Traceback (most recent call last)
....
TypeError: unhashable type: 'dict'

查看结果得到的状况：

  • 字典的顺序不影响 hash 值，这个很好，符合传参时的乱序
  • 列表有重复的元素时 hash 值相同，严重的错误
  • 嵌套的结构还是会报相同的错误，无法 hash， 直接错误

pass这个方法

json

如果能保证字典的 key 都是字符的情况下， 完全可以使用 json.dumps 模块进行深层数据的转换，同时拿到对应 JSON 字符的 hash，这种方式在简单的同时性能也很高。

sort_keys 用来保证 hash 的顺序，尝试一下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14

In [173]: hash(json.dumps({'1': 1, '2': 2}))
Out[173]: 7455153485734856158

In [174]: hash(json.dumps({'2': 2, '1': 1}))
Out[174]: 1617658018128597682

In [175]: hash(json.dumps({'2': 2, '1': 1}, sort_keys=True))
Out[175]: 7455153485734856158

In [176]: hash(json.dumps([{'1': 1}], sort_keys=True))
Out[176]: -8429833462434112332

In [177]: hash(json.dumps([{ 1 : 1}], sort_keys=True))
Out[177]: -8429833462434112332

这次便可以 hash 所有的数据结构了，前提是对象可以 dumps.

CATALOG
    FEATURED TAGS
    CSS Vue Django JavaScript Python Tools Docker Mysql VSCode Link Linux 爬虫 Chrome Webpack
    FRIENDS