pythonキャッシュメカニズムとfunctools.lru_cache
概要
この間djangoソースを見てget_を見ましたcommads()メソッドを使用するとfunctools.lru_が表示されます.Cache装飾器でキャッシュを実現するには、簡単に説明します.キャッシュは、後続の取得ニーズに対応するために定量的データを保存する処理であり、データ取得の速度を速めることを目的としています.データの生成プロセスには、計算、整列、リモート取得などの操作が必要になる場合があります.同じデータを複数回使用する必要がある場合は、毎回再生成するのに時間がかかります.したがって,計算やリモートリクエストなどの操作で得られたデータをキャッシュすると,後続のデータ取得要件が加速する.
簡単な例
まず簡単な例を見てfunctools.lru_を体験しましょうCache装飾器
実行結果:
戻ってきた現象から分かるようにtest_lru_Cacheは1回しか実行されず、現象から見るとlru_Cache装飾器のキャッシュ機能が機能します.
次にdjangoがどのように実現されているかを見てみましょう
ソース解析
入ってlru_Cacheメソッドは、まずlru_を見ることができます.Cacheは閉パッケージです
次は_lru_cache_wrapperメソッドを見てみましょう
これまで,このキャッシュ方式はスレッドが安全であることが知られており,
次の結論がわかります. lru_Cacheは閉パケットを用いて関数結果のキャッシュ を実現する.は、 と見なす.は、結果を辞書でキャッシュするため、関数の位置およびキーワードパラメータがハッシュ可能である必要がある. は、異なるパラメータ・モードを、個別のキャッシュ・アイテムを有する異なる呼び出しと見なすことができる.例えば、 をクリアする.
またもっと自分でdjangoに行ってみたいと思います.
シミュレーション実装キャッシュ
キャッシュ・プロシージャの動的表示
上のコードをpythontutorに置いて実行するプロセスを示します
ハードディスク(HDD)へのキャッシュの例
おすすめ読書
Python functools.lru_Cacheキャッシュとその原理ソース解析を実現
キャッシュ置換ポリシー
django起動プロセスソース解読
知る者は言わず,言う者は知らない
この間djangoソースを見てget_を見ましたcommads()メソッドを使用するとfunctools.lru_が表示されます.Cache装飾器でキャッシュを実現するには、簡単に説明します.キャッシュは、後続の取得ニーズに対応するために定量的データを保存する処理であり、データ取得の速度を速めることを目的としています.データの生成プロセスには、計算、整列、リモート取得などの操作が必要になる場合があります.同じデータを複数回使用する必要がある場合は、毎回再生成するのに時間がかかります.したがって,計算やリモートリクエストなどの操作で得られたデータをキャッシュすると,後続のデータ取得要件が加速する.
簡単な例
まず簡単な例を見てfunctools.lru_を体験しましょうCache装飾器
import time
from functools import lru_cache
@lru_cache()
def test_lru_cache(x, y):
time.sleep(1)
print('i am test')
return x * 10, y * 10
print(" ")
test_lru_cache(1, 2)
print(" ")
test_lru_cache(1, 2)
実行結果:
i am test
戻ってきた現象から分かるようにtest_lru_Cacheは1回しか実行されず、現象から見るとlru_Cache装飾器のキャッシュ機能が機能します.
次にdjangoがどのように実現されているかを見てみましょう
ソース解析
入ってlru_Cacheメソッドは、まずlru_を見ることができます.Cacheは閉パッケージです
def lru_cache(maxsize=128, typed=False):
"""
。
:param maxsize: （ 128）， maxsize=None， LRU ，
:param typed: typed true，
:return:
"""
# API lru_cache：cache_info，cache_clear f .__ wrapped__
# lru_cache （ C ）。
# @lru_cache ， maxsize None。
if maxsize is not None and not isinstance(maxsize, int):
raise TypeError('Expected maxsize to be an integer or None')
def decorating_function(user_function):
wrapper = _lru_cache_wrapper(user_function, maxsize, typed, _CacheInfo)
return update_wrapper(wrapper, user_function)
return decorating_function次は_lru_cache_wrapperメソッドを見てみましょう
def _lru_cache_wrapper(user_function, maxsize, typed, _CacheInfo):
"""
:param user_function:
:param maxsize:
:param typed: typed true，
:param _CacheInfo:
:return:
"""
# Constants shared by all lru cache instances:
# lru ：
#
sentinel = object()
# _make_key key hash
make_key = _make_key
# link
PREV, NEXT, KEY, RESULT = 0, 1, 2, 3
#
cache = {}
#
hits = misses = 0
#cache
full = False
#
cache_get = cache.get
#
cache_len = cache.__len__
# ( )
lock = RLock()
#
root = []
#
root[:] = [root, root, None, None]
if maxsize == 0:
# -
def wrapper(*args, **kwds):
nonlocal misses
result = user_function(*args, **kwds)
misses += 1
return result
elif maxsize is None:
# ，
def wrapper(*args, **kwds):
nonlocal hits, misses
# key hash
key = make_key(args, kwds, typed)
#
result = cache_get(key, sentinel)
#
if result is not sentinel:
hits += 1
return result
result = user_function(*args, **kwds)
cache[key] = result
misses += 1
return result
else:
# ，
def wrapper(*args, **kwds):
nonlocal root, hits, misses, full
key = make_key(args, kwds, typed)
with lock:
# value
link = cache_get(key)
if link is not None:
#
link_prev, link_next, _key, result = link
link_prev[NEXT] = link_next
link_next[PREV] = link_prev
last = root[PREV]
last[NEXT] = root[PREV] = link
link[PREV] = last
link[NEXT] = root
hits += 1
return result
result = user_function(*args, **kwds)
with lock:
if key in cache:
# key 。 Link ，
pass
elif full:
# root key result。
oldroot = root
oldroot[KEY] = key
oldroot[RESULT] = result
# Link， root。 key result ， 。
# Link ， （ __del__）。
root = oldroot[NEXT]
oldkey = root[KEY]
oldresult = root[RESULT]
root[KEY] = root[RESULT] = None
# cache
del cache[oldkey]
# root Link ， cache[key] 。
cache[key] = oldroot
else:
# 。
last = root[PREV]
link = [last, root, key, result]
last[NEXT] = root[PREV] = cache[key] = link
# cache_len len()
# lru_cache 。
full = (cache_len() >= maxsize)
misses += 1
return result
def cache_info():
"""cache """
with lock:
return _CacheInfo(hits, misses, maxsize, cache_len())
def cache_clear():
""" cache """
with lock:
cache.clear()
pass
wrapper.cache_info = cache_info
wrapper.cache_clear = cache_clear
return wrapperこれまで,このキャッシュ方式はスレッドが安全であることが知られており,
LRUアルゴリズムを用いて,そのライフサイクルはプロセス創設後の装飾関数の最初の実行からプロセス終了まで開始した.次の結論がわかります.
LRUアルゴリズム(最近最も使用されていない)を用いて、関数結果のキャッシュ更新を実現する.maxsize=Noneの場合、LRUの機能は無効になり、キャッシュは無限に増加することができる.typedをtrueに設定すると、異なるタイプの関数パラメータがそれぞれキャッシュされます.例えば、f(3)およびf(3.0)は、異なる結果を有する異なる呼び出しf(a=1,b=2)およびf(b=2,a=1) のキーワードパラメータの順序が異なり、2つの個別のキャッシュエントリがある場合があります.func.cache_info()：キャッシュ情報を表示func.cache_clear()：キャッシュ情報またもっと自分でdjangoに行ってみたいと思います.
シミュレーション実装キャッシュ
import random
import datetime
class MyCache:
def __init__(self):
# kv
self.cache = {}
#
self.max_cache_size = 10
def __contains__(self, key):
"""
:param key:
:return: True or False
"""
return key in self.cache
def get(self, key):
""" """
data = self.cache[key]
data["date_accessed"] = datetime.datetime.now()
return data["value"]
def add(self, key, value):
"""
（ ）
:param key:
:param value:
:return:
"""
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):
"""
:return:
"""
oldest_entry = None
for key in self.cache:
if oldest_entry is None:
oldest_entry = key
continue
curr_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 = key
self.cache.pop(oldest_entry)
@property
def size(self):
"""
:return:
"""
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:
continue
else:
value = ''.join([random.choice(s) for i in range(20)])
cache.add(key, value)
assert "test" not in cache
print(cache.cache)キャッシュ・プロシージャの動的表示
#
PREV, NEXT, KEY, RESULT = 0, 1, 2, 3
cache = {}
root = []
root[:] = [root, root, None, None]
#
key= 1
result = 10
last = root[PREV]
link = [last, root, key, result]
last[NEXT] = root[PREV] = cache[key] = link
#
key1 = 2
result1 = 20
last1 = root[PREV]
link1 = [last, root, key1, result1]
last1[NEXT] = root[PREV] = cache[key1] = link1
#
link_prev, link_next, _key, result2 = link
link_prev[NEXT] = link_next
link_next[PREV] = link_prev
last = root[PREV]
last[NEXT] = root[PREV] = link
link[PREV] = last
link[NEXT] = root
# ，
key2 = 3
result2 = 30
oldroot = root
oldroot[KEY] = key2
oldroot[RESULT] = result2
root = oldroot[NEXT]
oldkey = root[KEY]
oldresult = root[RESULT]
root[KEY] = root[RESULT] = None
del cache[oldkey]
cache[key2] = oldroot
上のコードをpythontutorに置いて実行するプロセスを示します
ハードディスク(HDD)へのキャッシュの例
import os
import uuid
import pickle
import shutil
import tempfile
from functools import wraps as func_wraps
class DiskCache(object):
""" """
_NAMESPACE = uuid.UUID("xxxxxxx-xxxxx-xxxxxxxxxxxx")
def __init__(self, cache_path=None):
"""
:param cache_path:
"""
if cache_path:
self.cache_path = os.path.abspath(cache_path)
else:
self.cache_path = os.path.join(tempfile.gettempdir(), ".diskcache")
def __call__(self, func):
"""
，
，
"""
@func_wraps(func)
def wrapper(*args, **kw):
params_uuid = uuid.uuid5(self._NAMESPACE, "-".join(map(str, (args, kw))))
key = '{}-{}.cache'.format(func.__name__, str(params_uuid))
cache_file = os.path.join(self.cache_path, key)
if not os.path.exists(self.cache_path):
os.makedirs(self.cache_path)
try:
with open(cache_file, 'rb') as f:
val = pickle.load(f)
except Exception:
val = func(*args, **kw)
try:
with open(cache_file, 'wb') as f:
pickle.dump(val, f)
except Exception:
pass
return val
return wrapper
def clear(self, func_name):
""" """
for cache_file in os.listdir(self.cache_path):
if cache_file.startswith(func_name + "-"):
os.remove(os.path.join(self.cache_path, cache_file))
def clear_all(self):
""" """
if os.path.exists(self.cache_path):
shutil.rmtree(self.cache_path)おすすめ読書
Python functools.lru_Cacheキャッシュとその原理ソース解析を実現
キャッシュ置換ポリシー
django起動プロセスソース解読
知る者は言わず,言う者は知らない