装饰器功能概述
本装饰器为纯函数与 I/O 密集函数提供可配置缓存，面向数据分析计算与三方接口聚合等高频重复计算/请求场景，支持同步与异步函数两种形态。核心特性：
- 缓存后端可选：内存（LRU）或 Redis
- TTL 过期控制
- 自定义键生成（key）与条件跳过缓存（unless）
- 命名空间（namespace）隔离与失效策略（按参失效/按命名空间清空）
- 自定义序列化（serializer）
- 防缓存击穿的同参并发锁（同步/异步分别处理）
- 完整保留原函数元信息（name、docstring、annotations）
适用方向：
- 数据分析中昂贵的统计/聚合计算
- 对三方 API 的聚合/拼装（避免重复请求）
- 复杂查询或跨系统接口的结果缓存
完整代码实现
from __future__ import annotations
import asyncio
import functools
import hashlib
import inspect
import pickle
import threading
import time
from collections import OrderedDict
from dataclasses import dataclass
from typing import Any, Callable, Dict, Optional, Protocol, Tuple, TypeVar, ParamSpec, Union
P = ParamSpec("P")
R = TypeVar("R")
# -------------------------
# Serializer abstraction
# -------------------------
class Serializer(Protocol):
def dumps(self, obj: Any) -> bytes: ...
def loads(self, data: bytes) -> Any: ...
class PickleSerializer:
"""默认序列化实现，使用 pickle 的最高协议。"""
def dumps(self, obj: Any) -> bytes:
return pickle.dumps(obj, protocol=pickle.HIGHEST_PROTOCOL)
def loads(self, data: bytes) -> Any:
return pickle.loads(data)
# -------------------------
# Cache Backend abstraction
# -------------------------
class BaseCacheBackend:
"""缓存后端抽象，提供同步与异步两套 API（异步默认调用同步实现）。"""
def get(self, full_key: str) -> Any:
raise NotImplementedError
def set(self, full_key: str, value: Any, ttl: Optional[int]) -> None:
raise NotImplementedError
def delete(self, full_key: str) -> None:
raise NotImplementedError
def clear_namespace(self, namespace: str) -> None:
raise NotImplementedError
async def aget(self, full_key: str) -> Any:
# 默认异步转同步
return self.get(full_key)
async def aset(self, full_key: str, value: Any, ttl: Optional[int]) -> None:
self.set(full_key, value, ttl)
async def adelete(self, full_key: str) -> None:
self.delete(full_key)
async def aclear_namespace(self, namespace: str) -> None:
self.clear_namespace(namespace)
# -------------------------
# Memory LRU cache backend
# -------------------------
class MemoryCache(BaseCacheBackend):
"""线程安全的内存 LRU 缓存，带 TTL 与命名空间前缀。"""
def __init__(self, maxsize: int = 500, namespace: str = "default") -> None:
self.maxsize = maxsize
self.namespace = namespace
self._store: "OrderedDict[str, Tuple[Optional[float], Any]]" = OrderedDict()
self._lock = threading.RLock()
def _now(self) -> float:
return time.time()
def _is_expired(self, exp: Optional[float]) -> bool:
return exp is not None and self._now() >= exp
def _compose(self, key: str) -> str:
return f"{self.namespace}:{key}"
def _evict_if_needed(self) -> None:
# 优先清理过期；否则按 LRU 淘汰
if self.maxsize <= 0:
self._store.clear()
return
# 先尽量清掉过期项
keys_to_delete = [k for k, (exp, _) in self._store.items() if self._is_expired(exp)]
for k in keys_to_delete:
self._store.pop(k, None)
while len(self._store) > self.maxsize:
self._store.popitem(last=False)
def get(self, full_key: str) -> Any:
with self._lock:
item = self._store.get(full_key)
if not item:
return None
exp, value = item
if self._is_expired(exp):
self._store.pop(full_key, None)
return None
# 命中则移动到 LRU 末尾
self._store.move_to_end(full_key, last=True)
return value
def set(self, full_key: str, value: Any, ttl: Optional[int]) -> None:
with self._lock:
exp = None if not ttl or ttl <= 0 else (self._now() + ttl)
self._store[full_key] = (exp, value)
self._store.move_to_end(full_key, last=True)
self._evict_if_needed()
def delete(self, full_key: str) -> None:
with self._lock:
self._store.pop(full_key, None)
def clear_namespace(self, namespace: str) -> None:
prefix = f"{namespace}:"
with self._lock:
keys = [k for k in self._store.keys() if k.startswith(prefix)]
for k in keys:
self._store.pop(k, None)
# 异步 API 直接复用同步（内存操作非阻塞）
# aget/aset/adelete/aclear_namespace 继承自父类的默认实现
# -------------------------
# Redis cache backend
# -------------------------
class RedisCache(BaseCacheBackend):
"""
Redis 后端。
- 支持 redis.Redis（同步）或 redis.asyncio.Redis（异步）
- 若仅提供同步 client 且在异步调用中使用，会通过 asyncio.to_thread 转线程池避免阻塞事件循环
"""
def __init__(self, client: Any, serializer: Serializer, namespace: str = "default") -> None:
self.client = client
self.serializer = serializer
self.namespace = namespace
# 检测是否是异步客户端（redis.asyncio）
self._is_async_client = any(
inspect.iscoroutinefunction(getattr(client, name, None))
for name in ("get", "set", "delete", "scan", "scan_iter")
)
def _compose(self, key: str) -> str:
return f"{self.namespace}:{key}"
# ---- Sync API ----
def get(self, full_key: str) -> Any:
if self._is_async_client:
# 避免 event loop 中误用同步 API
raise RuntimeError("Async Redis client in sync context; use aget() wrapper.")
data = self.client.get(full_key)
if data is None:
return None
return self.serializer.loads(data)
def set(self, full_key: str, value: Any, ttl: Optional[int]) -> None:
if self._is_async_client:
raise RuntimeError("Async Redis client in sync context; use aset() wrapper.")
data = self.serializer.dumps(value)
ex = ttl if ttl and ttl > 0 else None
# ex=TTL 秒
self.client.set(full_key, data, ex=ex)
def delete(self, full_key: str) -> None:
if self._is_async_client:
raise RuntimeError("Async Redis client in sync context; use adelete() wrapper.")
self.client.delete(full_key)
def clear_namespace(self, namespace: str) -> None:
if self._is_async_client:
raise RuntimeError("Async Redis client in sync context; use aclear_namespace() wrapper.")
prefix = f"{namespace}:"
# 使用 scan_iter 避免大键空间阻塞
if hasattr(self.client, "scan_iter"):
for k in self.client.scan_iter(match=f"{prefix}*"):
self.client.delete(k)
else:
# 回退到 scan 循环
cursor = 0
while True:
cursor, keys = self.client.scan(cursor=cursor, match=f"{prefix}*", count=1000)
if keys:
self.client.delete(*keys)
if cursor == 0:
break
# ---- Async API ----
async def aget(self, full_key: str) -> Any:
if self._is_async_client:
data = await self.client.get(full_key)
else:
data = await asyncio.to_thread(self.client.get, full_key)
if data is None:
return None
return self.serializer.loads(data)
async def aset(self, full_key: str, value: Any, ttl: Optional[int]) -> None:
data = self.serializer.dumps(value)
if self._is_async_client:
ex = ttl if ttl and ttl > 0 else None
await self.client.set(full_key, data, ex=ex)
else:
await asyncio.to_thread(self.client.set, full_key, data, None if not ttl or ttl <= 0 else ttl)
async def adelete(self, full_key: str) -> None:
if self._is_async_client:
await self.client.delete(full_key)
else:
await asyncio.to_thread(self.client.delete, full_key)
async def aclear_namespace(self, namespace: str) -> None:
prefix = f"{namespace}:"
if self._is_async_client:
# 优先使用异步 scan_iter
if hasattr(self.client, "scan_iter"):
async for k in self.client.scan_iter(match=f"{prefix}*"):
await self.client.delete(k)
else:
cursor = 0
while True:
cursor, keys = await self.client.scan(cursor=cursor, match=f"{prefix}*", count=1000)
if keys:
await self.client.delete(*keys)
if cursor == 0:
break
else:
# 同步客户端转线程
def _clear():
if hasattr(self.client, "scan_iter"):
for k in self.client.scan_iter(match=f"{prefix}*"):
self.client.delete(k)
else:
cursor = 0
while True:
cursor, keys = self.client.scan(cursor=cursor, match=f"{prefix}*", count=1000)
if keys:
self.client.delete(*keys)
if cursor == 0:
break
await asyncio.to_thread(_clear)
# -------------------------
# Key builder
# -------------------------
def _stable_hash_bytes(obj: Any) -> bytes:
"""尽量稳定、可跨进程的参数摘要：优先 pickle；失败则退回 repr()."""
try:
return pickle.dumps(obj, protocol=pickle.HIGHEST_PROTOCOL)
except Exception:
return repr(obj).encode("utf-8", errors="ignore")
def default_key_builder(func: Callable[..., Any], namespace: str, args: Tuple[Any, ...], kwargs: Dict[str, Any]) -> str:
"""
产生稳定且短小的 key：namespace + 函数标识 + 参数摘要（blake2b）。
"""
hasher = hashlib.blake2b(digest_size=16)
hasher.update(_stable_hash_bytes(args))
# kwargs 按 key 排序
if kwargs:
items = tuple(sorted(kwargs.items(), key=lambda kv: kv[0]))
hasher.update(_stable_hash_bytes(items))
digest = hasher.hexdigest()
fn_id = f"{func.__module__}.{getattr(func, '__qualname__', func.__name__)}"
return f"{fn_id}:{digest}"
# -------------------------
# Decorator factory
# -------------------------
@dataclass
class CacheConfig:
backend: str = "memory" # "memory" | "redis"
ttl: int = 60
key: Optional[Callable[[Callable[..., Any], str, Tuple[Any, ...], Dict[str, Any]], str]] = None
unless: Optional[Callable[..., bool]] = None # True => 跳过缓存
maxsize: int = 500 # 仅对 memory 生效
namespace: str = "default"
serializer: Optional[Serializer] = None # Redis 使用；若不传用 Pickle
backend_options: Optional[Dict[str, Any]] = None # e.g. {'client': redis_client}
def cacheable(
*,
backend: str = "memory",
ttl: int = 60,
key: Optional[Callable[[Callable[..., Any], str, Tuple[Any, ...], Dict[str, Any]], str]] = None,
unless: Optional[Callable[..., bool]] = None,
maxsize: int = 500,
namespace: str = "default",
serializer: Optional[Serializer] = None,
backend_options: Optional[Dict[str, Any]] = None,
) -> Callable[[Callable[P, R]], Callable[P, R]]:
"""
可配置缓存装饰器：
- backend: "memory" | "redis"
- ttl: 秒；<=0 表示不过期（但仍可手动失效）
- key: 自定义键函数 (func, namespace, args, kwargs) -> str
- unless: 返回 True 则跳过缓存（不读不写）
- maxsize: 内存 LRU 容量上限
- namespace: 业务隔离用前缀
- serializer: 序列化器（默认 PickleSerializer）；Redis 必需/默认
- backend_options: 后端额外配置，Redis 需提供 {'client': redis_client}
"""
cfg = CacheConfig(
backend=backend,
ttl=ttl,
key=key or default_key_builder,
unless=unless,
maxsize=maxsize,
namespace=namespace,
serializer=serializer or PickleSerializer(),
backend_options=backend_options or {},
)
# 构造后端
if cfg.backend not in ("memory", "redis"):
raise ValueError("backend must be 'memory' or 'redis'.")
if cfg.backend == "memory":
cache_backend: BaseCacheBackend = MemoryCache(maxsize=cfg.maxsize, namespace=cfg.namespace)
else:
client = cfg.backend_options.get("client")
if client is None:
raise ValueError("For backend='redis', backend_options['client'] is required.")
cache_backend = RedisCache(client=client, serializer=cfg.serializer, namespace=cfg.namespace)
# 并发锁：按 key 细粒度
_thread_locks: Dict[str, threading.Lock] = {}
_async_locks: Dict[str, asyncio.Lock] = {}
def _get_thread_lock(k: str) -> threading.Lock:
# 轻量线程安全：双重检查
lock = _thread_locks.get(k)
if lock:
return lock
with threading.Lock():
return _thread_locks.setdefault(k, threading.Lock())
def _get_async_lock(k: str) -> asyncio.Lock:
lock = _async_locks.get(k)
if lock:
return lock
lock = asyncio.Lock()
_async_locks[k] = lock
return lock
def decorator(func: Callable[P, R]) -> Callable[P, R]:
is_coro = inspect.iscoroutinefunction(func)
@functools.wraps(func)
def sync_wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
# unless 为 True => 直接调用原函数
if cfg.unless and cfg.unless(*args, **kwargs):
return func(*args, **kwargs)
k = cfg.key(func, cfg.namespace, args, kwargs) # base key
full_key = f"{cfg.namespace}:{k}"
# 1st 尝试读取
cached = cache_backend.get(full_key)
if cached is not None:
return cached
# 防击穿：按 key 加锁
lk = _get_thread_lock(full_key)
with lk:
# 进入临界区后再次尝试读取
cached2 = cache_backend.get(full_key)
if cached2 is not None:
return cached2
# 计算与写入
result = func(*args, **kwargs)
cache_backend.set(full_key, result, cfg.ttl)
return result
@functools.wraps(func)
async def async_wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
if cfg.unless and cfg.unless(*args, **kwargs):
return await func(*args, **kwargs) # type: ignore[misc]
k = cfg.key(func, cfg.namespace, args, kwargs)
full_key = f"{cfg.namespace}:{k}"
cached = await cache_backend.aget(full_key)
if cached is not None:
return cached
lk = _get_async_lock(full_key)
async with lk:
cached2 = await cache_backend.aget(full_key)
if cached2 is not None:
return cached2
result = await func(*args, **kwargs) # type: ignore[misc]
await cache_backend.aset(full_key, result, cfg.ttl)
return result
# 选择同步/异步 wrapper
wrapper: Union[Callable[P, R], Callable[P, R]] = async_wrapper if is_coro else sync_wrapper
# 失效 API 挂载（与函数形态一致）
if is_coro:
async def invalidate(*iargs: P.args, **ikwargs: P.kwargs) -> None:
k = cfg.key(func, cfg.namespace, iargs, ikwargs)
full_key = f"{cfg.namespace}:{k}"
await cache_backend.adelete(full_key)
async def clear_namespace() -> None:
await cache_backend.aclear_namespace(cfg.namespace)
setattr(wrapper, "invalidate", invalidate)
setattr(wrapper, "clear_namespace", clear_namespace)
else:
def invalidate(*iargs: P.args, **ikwargs: P.kwargs) -> None:
k = cfg.key(func, cfg.namespace, iargs, ikwargs)
full_key = f"{cfg.namespace}:{k}"
cache_backend.delete(full_key)
def clear_namespace() -> None:
cache_backend.clear_namespace(cfg.namespace)
setattr(wrapper, "invalidate", invalidate)
setattr(wrapper, "clear_namespace", clear_namespace)
# 便于调试/扩展：暴露 backend 与配置
setattr(wrapper, "cache_backend", cache_backend)
setattr(wrapper, "cache_config", cfg)
return wrapper # type: ignore[return-value]
return decorator
使用示例
示例 1：数据分析计算（同步），内存 LRU 缓存
# 假设为昂贵的聚合计算
@cacheable(
backend="memory",
ttl=120, # 2 分钟
maxsize=1000,
namespace="analytics",
)
def heavy_agg(series: list[int], topk: int = 10) -> dict:
time.sleep(0.5) # 模拟昂贵计算
sorted_vals = sorted(series, reverse=True)[:topk]
return {"topk": sorted_vals, "sum": sum(series)}
# 调用
data = [1,2,3,4,5,6,7,8,9,10]
print(heavy_agg(data, topk=3)) # 首次 ~0.5s
print(heavy_agg(data, topk=3)) # 命中缓存，基本 0ms
# 按参数失效（仅移除对应参数组合的缓存）
heavy_agg.invalidate(data, topk=3)
# 按命名空间清空
heavy_agg.clear_namespace()
示例 2：三方接口聚合（异步），Redis 缓存
import asyncio
from typing import Any
# 使用 redis-py 4.x 的 asyncio 客户端
from redis.asyncio import Redis
redis_client = Redis(host="127.0.0.1", port=6379, db=0)
# unless：当 request_no_cache=True 时跳过缓存
def unless_no_cache(*args, **kwargs) -> bool:
return kwargs.get("request_no_cache", False)
# 自定义键：以 user_id 与 path 构建
def api_key(func, namespace, args, kwargs) -> str:
user_id = kwargs.get("user_id", "anon")
path = kwargs.get("path", "/")
# 与默认行为一致，末尾仍加入参数摘要，避免同 user_id/path 不同 query 的冲突
return f"{func.__module__}.{func.__qualname__}:{user_id}:{path}:{hashlib.blake2b(pickle.dumps(kwargs, protocol=pickle.HIGHEST_PROTOCOL), digest_size=8).hexdigest()}"
@cacheable(
backend="redis",
ttl=30,
key=api_key,
unless=unless_no_cache,
namespace="api-aggregate",
backend_options={"client": redis_client},
)
async def aggregate_third_party(*, user_id: str, path: str, query: dict[str, Any]) -> dict[str, Any]:
# 模拟聚合多个第三方 API
await asyncio.sleep(0.2)
return {"user": user_id, "path": path, "result": {"q": query}, "ts": time.time()}
async def main():
print(await aggregate_third_party(user_id="42", path="/profile", query={"lang": "zh"})) # 首次 ~0.2s
print(await aggregate_third_party(user_id="42", path="/profile", query={"lang": "zh"})) # 命中 Redis，基本 0ms
# 跳过缓存强制刷新
print(await aggregate_third_party(user_id="42", path="/profile", query={"lang": "zh"}, request_no_cache=True))
# 失效与清理
await aggregate_third_party.invalidate(user_id="42", path="/profile", query={"lang": "zh"})
await aggregate_third_party.clear_namespace()
asyncio.run(main())
示例 3：为 Redis 指定 JSON 序列化（便于可视化）
import json
class JsonSerializer:
def dumps(self, obj: Any) -> bytes:
return json.dumps(obj, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
def loads(self, data: bytes) -> Any:
return json.loads(data.decode("utf-8"))
json_serializer = JsonSerializer()
@cacheable(
backend="redis",
ttl=60,
namespace="json-demo",
serializer=json_serializer,
backend_options={"client": redis_client}, # 同上示例
)
def expensive_config() -> dict:
time.sleep(0.3)
return {"feature": True, "version": 3, "meta": {"owner": "platform"}}
print(expensive_config()) # 首次
print(expensive_config()) # 命中
技术说明
-
键生成（key）
- 默认 key 使用函数全名（module + qualname）与参数摘要，摘要由 pickle 序列化失败后回退到 repr 的二次策略产生，再经 blake2b(digest_size=16) 得到短哈希，兼顾稳定性与冲突率。
- 支持自定义 key(func, namespace, args, kwargs) -> str；装饰器会自动加上命名空间前缀 namespace:。
-
TTL 与过期
- MemoryCache 在读时检测过期并懒删除；写入时存储过期时间戳。
- RedisCache 通过 set(..., ex=ttl) 维护 TTL。ttl<=0 视为不过期（Memory 存 None，Redis 不设置 ex）。
-
失效策略
- wrapper.invalidate(*args, **kwargs)：按原始参数组合失效（与 key 逻辑一致）。
- wrapper.clear_namespace()：清理当前命名空间下的全部 key。
- 以上方法在异步函数上为异步方法（需 await）。
-
并发与防击穿
- 内部使用每个 key 一个细粒度锁：同步用 threading.Lock，异步用 asyncio.Lock。
- Double-check：进入锁前后各读一次缓存，避免重复计算。
-
序列化（serializer）
- MemoryCache 直接缓存 Python 对象（无需序列化）。
- RedisCache 使用 serializer 进行 bytes 化；默认 PickleSerializer。可自定义 JSON 等实现，需具备 dumps/loads。
-
后端兼容
- RedisCache 同时兼容 redis.Redis（同步）与 redis.asyncio.Redis（异步）。
- 若异步函数搭配同步 redis 客户端，内部通过 asyncio.to_thread 转入线程池，避免阻塞事件循环。
-
元信息保护
- 使用 functools.wraps 保留原函数的名称、注释与类型注解。
注意事项
- key 冲突与可序列化
- 默认策略已尽量稳定，但若参数包含不可 pickle 的对象将回退到 repr，可能导致不同对象 repr 相同的极端冲突。对关键业务建议提供显式 key 函数。
- Memory LRU 的 maxsize
- 仅对内存后端有效；过大可能增加内存占用。系统进程内缓存，不跨进程共享。
- Redis 客户端
- 建议异步函数使用 redis.asyncio.Redis；同步函数使用 redis.Redis。若混用，同样可工作但会有线程切换开销。
- unless 条件
- unless(*args, **kwargs) 返回 True 时，本次调用完全绕过缓存（不读不写）。常用于强制刷新、灰度或带 no-cache 标志的请求。
- TTL 配置
- ttl<=0 表示长期缓存，需配合 invalidate/clear_namespace 管理生命周期。
- 序列化安全
- Pickle 在不受信任环境下反序列化存在风险；本实现仅在可信环境中使用，并不对外部输入进行反序列化。对跨边界场景建议使用 JSON 或其他安全序列化。
- 大规模清理
- Redis 的 clear_namespace 基于 SCAN/SCAN_ITER，已尽量避免阻塞，但在极大键空间仍可能造成一定负载；建议按业务分层 namespace。
- 异常传播
- 被装饰函数抛出的异常将原样传播，不会缓存失败结果。
- 不修改函数签名
- 装饰器不修改调用签名；失效等能力以方法属性形式暴露（invalidate/clear_namespace）。
如需扩展支持：
- 互斥锁外溢（分布式锁）防止多进程下击穿，可在 Redis 上新增 SETNX 锁与过期机制
- 写后异步回填、穿透保护（布隆过滤器）等，可在当前结构上叠加实现。
