快取是一種將定量資料加以保存以備迎合后續獲取需求的處理方式,旨在加快資料獲取的速度。資料的生成程序可能需要經過計算,規整,遠程獲取等操作,如果是同一份資料需要多次使用,每次都重新生成會大大浪費時間。所以,如果將計算或者遠程請求等操作獲得的資料快取下來,會加快后續的資料獲取需求。
先來一個簡單的例子以了解快取機制的概念:
# -*- coding: utf-8 -*-
import random
import datetime
class MyCache:
"""快取類"""
def __init__(self):
# 用字典結構以 kv 的形式快取資料
self.cache = {}
# 限制快取的大小,因為快取的空間有限
# 所以當快取太大時,需要將舊的快取舍棄掉
self.max_cache_size = 10
def __contains__(self, key):
"""根據該鍵是否存在于快取當中回傳 True 或者 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):
"""更新該快取字典,如果快取太大則先洗掉最早條目"""
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 = 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 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)
以上示例僅簡單的展示了快取機制的原理,通過用鍵值對的方式將資料放到字典中,如果下次需要取值時可以直接到字典中獲取。該示例在洗掉舊資料時的實作并不高效,實際應用中可以用別的方式實作。
在 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 的使用效果:
from functools import lru_cache
@lru_cache(None)
def add(x, y):
print("calculating: %s + %s" % (x, y))
return x + y
print(add(1, 2))
print(add(1, 2))
print(add(2, 3))
輸出結果:
calculating: 1 + 2
3
3
calculating: 2 + 3
5
從結果可以看出,當第二次呼叫 add(1, 2) 時,并沒有真正執行函式體,而是直接回傳快取的結果。
如果要在 Python 2 中使用 lru_cahce 需要安裝第三方模塊 functools32。還有一個用 C 語言實作的,更快的,同時兼容 Python2 和 Python3 的第三方模塊 fastcache 能夠實作同樣的功能,且其能支持 TTL。
lru_cahce 是將資料快取到記憶體中的,其實也可以將資料快取到磁盤上。以下示例嘗試實作了一個基于磁盤的快取裝飾器:
import os
import uuid
import pickle
import shutil
import tempfile
from functools import wraps as func_wraps
class DiskCache(object):
"""快取資料到磁盤
實體化引數:
-----
cache_path: 快取檔案的路徑
"""
_NAMESPACE = uuid.UUID("c875fb30-a8a8-402d-a796-225a6b065cad")
def __init__(self, cache_path=None):
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)
cache_in_disk = DiskCache()
@cache_in_disk
def add(x, y):
return x + y
此外,還有一些其他的快取模塊,如 cachelib, cacheout 等等,實際使用需要時可以按需求去選擇合適的快取實作。
轉載請註明出處,本文鏈接:https://www.uj5u.com/qita/275947.html
標籤:其他
