周日, 1月 31 2016, 10:43:16 上午

python缓存模块的一些用法

python

Caches are bugs waiting to happen.- Rob Pike.

最近接触了几个关于python缓存模块的一些用法，记录一下。

###python3.2的@functools.lru_cache
functools 模块在python3.2以后新加入了这个装饰器，用来缓存一些耗时的IO请求结果或者周期调用的函数结果，如果缓存数据超出maxsize的值，就是用LRU（最近最少使用算法）清除一些缓存结果。
官方文档给出了几个示例:

@lru_cache(maxsize=32)
def get_pep(num):
    'Retrieve text of a Python Enhancement Proposal'
    resource = 'http://www.python.org/dev/peps/pep-%04d/' % num
    try:
        with urllib.request.urlopen(resource) as s:
            return s.read()
    except urllib.error.HTTPError:
        return 'Not Found'

>>> for n in 8, 290, 308, 320, 8, 218, 320, 279, 289, 320, 9991:
...     pep = get_pep(n)
...     print(n, len(pep))

>>> get_pep.cache_info()
CacheInfo(hits=3, misses=8, maxsize=32, currsize=8)

还有最常见的斐波那契数列，没有优化过的递归算法时间复杂度是O(1.618 ^ n)，如果用数组记录之前计算过的值或者用缓存就是O(n)的时间复杂度。

@lru_cache(maxsize=None)
def fib(n):
    if n < 2:
        return n
    return fib(n-1) + fib(n-2)

>>> [fib(n) for n in range(16)]
[0, 1, 1, 2, 3, 5, 8, 13, 21, 34, 55, 89, 144, 233, 377, 610]

>>> fib.cache_info()
CacheInfo(hits=28, misses=16, maxsize=None, currsize=16)

###cachetools模块
 cachetools模块帮我们实现了很多缓存失效方式，包括:

LFUCache(Least Frequently Used (LFU) cache implementation.)
LRUCache(Least Recently Used (LRU) cache implementation.)
RRCache(Random Replacement (RR) cache implementation.)
TTLCAche(LRU Cache implementation with per-item time-to-live (TTL) value.)

# pip install cachetools
import time
from cachetools import ttl_cache, cached


@ttl_cache(ttl=5)
def get(n):
    print('from function')    # 标记结果是从函数得到，而不是从缓存
    return n


for i in range(3):
    print(get(10))

time.sleep(6)

for i in range(3):
    print(get(10))

可以看见结果输出如下:

from function
10
10
10
from function
10
10
10

###redis，memcached等内存数据库缓存

这个就不细说了，网上很多例子。可以使用数据库提供的python api操作数据库实现缓存，不过一般需要自己实现失效策略，也可以直接给redis设置超时过期参数使缓存失效。一般数据量比较大的时候考虑用内存数据库。
最近学习flask的时候发现werkzeug.contrib.cache模块实现了MemcachedCache和RedisCache，可以直接拿来用，实现参考源码。

###实现原理
一般就是设置一个字典，第一次计算的时候结果存入字典，以后每次就可以从字典里取得计算结果，如果需要超时就可以再设置一个字典保存对应结果计算的时期，每次计算比对这个时间，超时就重新计算。注意参数要是可以hash的，也就是可当做字典的key的。

def cache(func):
    memo = {}

    @wraps(func)
    def _wrapper(*args):
        res = memo.get(args, None)
        if res is not None:
            return res
        else:
            res = func(*args)
            memo[args] = res
        return res
    return _wrapper


@cache
def fib(n):
    if n <= 1:
        return n
    else:
        return fib(n-1) + fib(n-2)

print([fib(i) for i in list(range(31))])

另外也可以用class来实现类似缓存，比如实现一个能够缓存网页爬虫结果的类：

import time
import requests

class CacheFetch:
    __slots__ = ('url', 'timeout', 'last_time', '_response')

    def __init__(self, url, timeout=None):
        self.url = url
        self.timeout = timeout or 3
        self.last_time = time.time()
        self._response = None

    @property
    def response(self):
        if not self._response or (time.time()-self.last_time > self.timeout):
            print('fetch page')
            self._response = requests.get(self.url)
            self.last_time = time.time()
        return self._response


url = 'http://www.baidu.com'
fetcher = CacheFetch(url, 1)

for i in range(3):
    st = time.time()
    print(fetcher.response.status_code)
    print(time.time()-st)

time.sleep(5)

for i in range(3):
    st = time.time()
    print(fetcher.response.status_code)
    print(time.time()-st)

执行可以看到一开始的for循环第一次是请求的网站数据，后面两次是直接从缓存取得的。之间sleep一下，然后再一次的for循环第一次还是要请求原来的网页，因为已经超时了，后两次循环就是直接从缓存取了。
最近看别人的代码还见到了这个cache_property，从flask里边抠出来如下，思想也差不多:

class _Missing(object):
    def __repr__(self):
        return 'no value'

    def __reduce__(self):
        return '_missing'


_missing = _Missing()    # 伪单例


class cached_property(object):
    def __init__(self, func, name=None, doc=None):
        self.__name__ = name or func.__name__
        self.__module__ = func.__module__
        self.__doc__ = doc or func.__doc__
        self.func = func

    def __get__(self, obj, type=None):
        if obj is None:
            return self
        value = obj.__dict__.get(self.__name__, _missing)
        if value is _missing:
            value = self.func(obj)
            obj.__dict__[self.__name__] = value
        print(value)
        return value


class TestCache():
    @cached_property
    def get(self):
        print('from function')
        return time.time()


def test():
    t = TestCache()
    print(t.get)
    print(t.get)

test()

执行后可以看到get只计算了一次，后两个的time.time()函数并没有执行。所以缓存不适用这种每次都返回不同值的函数。

###Ref
https://wiki.python.org/moin/PythonDecoratorLibrary#Memoize - 总结了几乎所有python装饰器的用法
缓存算法
 缓存更新的 Design Pattern，让我们多一点套路！

留言