当前位置:

首页 > temp > python入门教程 >

python基础（补充）：关于递归的优化（使用缓存）

在上一节中，我们一起探讨了递归的深度
递归深度有了底，你可以大胆使用递归了，然而问题又来了，python的递归和蜗牛一样慢，那么有没有优化的余地呢？因为我也是菜鸟，所以简单提供几种优化方案供大家学习交流。

1|0优化思路

第一角度优化算法，根据递归的计算过程计算过程中实例化了大量重复的函数计算，第一角度尝试优化计算逻辑，但是怎么优化算法说实话心里没谱。

既然优化算法没法实现，那么我们分析一下耗时的原因，其实在递归过程中自身调用自身不断实例化自身，计算机堆内存溢出导致递归有深度一次，在运算结果时候也是不断去计算每个实例化返回值，是否可以将计算过程中实例化返回值保存在一个缓存中或者一个IO中，计算结果时候每次从缓存或者IO中读取是不是能简化计算量从而提高效率呢？尝试去寻找一下缓存解决方案。找到了以下几种缓存优化方案，下面来共同学习下：

我们还是以斐波那契函数为例，先来看一下没有使用缓存的斐波那契数列：

copy

import sys from timeit import Timer sys.setrecursionlimit(3000) # 没有使用缓存的斐波那契数列 def fib(n): if n <= 2: return 1 else: return fib(n - 1) + fib(n - 2) t1 = Timer("fib(100)", "from __main__ import fib") # 斐波那契数列递归深度100，计算1000次时间 print("fib--100", t1.timeit(number=1000), "seconds") # 结果 # 太伤机器，放弃了

2|0优化方案一：使用计算缓存

copy

import sys from timeit import Timer sys.setrecursionlimit(3000) # 使用计算缓存 def fib(n, _cache={}): if n in _cache: return _cache[n] elif n > 1: return _cache.setdefault(n, fib(n - 1) + fib(n - 2)) return n t1 = Timer("fib(100)", "from __main__ import fib") print("fib--100", t1.timeit(number=1000), "seconds") # 运行结果 # fib--100 0.000353 seconds

3|0方案二：使用functools 中装饰器

copy

import sys from functools import lru_cache from timeit import Timer sys.setrecursionlimit(3000) # 使用functools装饰器 @lru_cache(maxsize=None) def fib(n): if n <= 2: return 1 else: return fib(n - 1) + fib(n - 2) t1 = Timer("fib(100)", "from __main__ import fib") print("fib--100", t1.timeit(number=1000), "seconds") # 运行结果 # fib--100 0.0002997 seconds

4|0方案三：使用github上的cache方案

copy

import sys from timeit import Timer import cache sys.setrecursionlimit(3000) @cache.cache(timeout=20, fname="my_cache.pkl") def fib(n): if n <= 2: return 1 else: return fib(n - 1) + fib(n - 2) t1 = Timer("fib(100)", "from __main__ import fib") print("fib--100", t1.timeit(number=1000), "seconds") # 运行结果 # fib--100 0.7063512 seconds

chahe.py代码

copy

import base64 import hashlib import inspect import pickle import time debug = False def log(s): if debug: print(s) caches = dict() updated_caches = [] def get_cache(fname): if fname in caches: return caches[fname] try: with open(fname, "rb") as f: c = pickle.load(f) except: c = dict() caches[fname] = c return c def write_to_cache(fname, obj): updated_caches.append(fname) caches[fname] = obj def cleanup(): for fname in updated_caches: with open(fname, "wb") as f: pickle.dump(caches[fname], f) def get_fn_hash(f): return base64.b64encode(hashlib.sha1(inspect.getsource(f).encode("utf-8")).digest()) NONE = 0 ARGS = 1 KWARGS = 2 def cache(fname=".cache.pkl", timeout=-1, key=ARGS | KWARGS): def impl(fn): load_t = time.time() c = get_cache(fname) log("loaded cache in {:.2f}s".format(time.time() - load_t)) def d(*args, **kwargs): log("checking cache on {}".format(fn.__name__)) if key == ARGS | KWARGS: k = pickle.dumps((fn.__name__, args, kwargs)) if key == ARGS: k = pickle.dumps((fn.__name__, args)) if key == KWARGS: k = pickle.dumps((fn.__name__, kwargs)) if key == NONE: k = pickle.dumps((fn.__name__)) if k in c: h, t, to, res = c[k] if get_fn_hash(fn) == h and (to < 0 or (time.time() - t) < to): log("cache hit.") return res log("cache miss.") res = fn(*args, **kwargs) c[k] = (get_fn_hash(fn), time.time(), timeout, res) save_t = time.time() write_to_cache(fname, c) log("saved cache in {:.2f}s".format(time.time() - save_t)) return res return d return impl @cache(timeout=0.2) def expensive(k): time.sleep(0.2) return k @cache(key=KWARGS) def expensive2(k, kwarg1=None): time.sleep(0.2) return k def test(): # Test timeout t = time.time() v = expensive(1) assert v == 1 assert time.time() - t > 0.1 t = time.time() expensive(1) assert time.time() - t < 0.1 time.sleep(0.3) t = time.time() expensive(1) assert time.time() - t > 0.1 t = time.time() v = expensive(2) assert v == 2 assert time.time() - t > 0.1 # Test key=_ annotation t = time.time() v = expensive2(2, kwarg1="test") assert v == 2 assert time.time() - t > 0.1 t = time.time() v = expensive2(1, kwarg1="test") assert v == 2 assert time.time() - t < 0.1 t = time.time() v = expensive2(1, kwarg1="test2") assert v == 1 assert time.time() - t > 0.1 cleanup() print("pass") if __name__ == "__main__": test()

5|0方案四：使用diskcache专业的缓存方案

copy

import sys from timeit import Timer from diskcache import FanoutCache #如果没有diskcache包需要先安装  sys.setrecursionlimit(3000) # 缓存临时文件位置 cache = FanoutCache('tmp/diskcache/fanoutcache') @cache.memoize(typed=True, expire=None, tag='fib') def fib(n): if n <= 2: return 1 else: return fib(n - 1) + fib(n - 2) t1 = Timer("fib(100)", "from __main__ import fib") print("fib--100", t1.timeit(number=1000), "seconds") # 运行结果 # fib--100 0.1423292 seconds