Python案例怎么实现缓存装饰器？

wen python案例 2026-06-10 200

本文目录导读：

Python案例怎么实现缓存装饰器？

基础缓存装饰器（字典缓存）
带过期时间的缓存装饰器
LRU（最近最少使用）缓存装饰器
支持自定义键函数的缓存装饰器
使用Python内置的LRU缓存
异步函数缓存装饰器
使用建议

下面我通过几个案例,由浅入深地介绍如何在 Python 中实现缓存装饰器。

基础缓存装饰器（字典缓存）

import functools
def cache_decorator(func):
    """最简单的缓存装饰器"""
    cache = {}  # 缓存存储
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        # 创建缓存的键
        key = str(args) + str(kwargs)
        if key not in cache:
            cache[key] = func(*args, **kwargs)
            print(f"缓存未命中，计算结果: {cache[key]}")
        else:
            print(f"缓存命中，返回结果: {cache[key]}")
        return cache[key]
    return wrapper
@cache_decorator
def fibonacci(n):
    """计算斐波那契数列"""
    if n <= 1:
        return n
    return fibonacci(n-1) + fibonacci(n-2)
# 测试
print(fibonacci(5))  # 首次计算
print(fibonacci(5))  # 缓存命中
print(fibonacci(6))  # 部分缓存命中

带过期时间的缓存装饰器

import time
import functools
def timed_cache(expire_time=60):
    """带过期时间的缓存装饰器"""
    def decorator(func):
        cache = {}  # {key: (value, timestamp)}
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            key = str(args) + str(kwargs)
            current_time = time.time()
            # 检查缓存是否有效
            if key in cache:
                value, timestamp = cache[key]
                if current_time - timestamp < expire_time:
                    print(f"缓存命中，剩余时间: {expire_time - (current_time - timestamp):.2f}秒")
                    return value
                else:
                    print("缓存已过期")
            # 重新计算并缓存
            value = func(*args, **kwargs)
            cache[key] = (value, current_time)
            print(f"新建缓存，过期时间: {expire_time}秒")
            return value
        return wrapper
    return decorator
@timed_cache(expire_time=5)  # 5秒过期
def get_time():
    return time.time()
# 测试
print(f"第一次调用: {get_time()}")
time.sleep(2)
print(f"2秒后调用: {get_time()}")
time.sleep(4)
print(f"6秒后调用: {get_time()}")  # 缓存过期

LRU（最近最少使用）缓存装饰器

import functools
from collections import OrderedDict
def lru_cache(maxsize=128):
    """LRU缓存装饰器"""
    def decorator(func):
        cache = OrderedDict()
        hits = 0
        misses = 0
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            nonlocal hits, misses
            key = str(args) + str(kwargs)
            if key in cache:
                hits += 1
                # 将访问的项移到末尾（最近使用）
                cache.move_to_end(key)
                print(f"缓存命中! 命中率: {hits/(hits+misses)*100:.2f}%")
                return cache[key]
            misses += 1
            value = func(*args, **kwargs)
            # 如果缓存满了，删除最久未使用的项
            if len(cache) >= maxsize:
                cache.popitem(last=False)  # 删除最早插入的项
                print("缓存已满，移除最久未使用项")
            cache[key] = value
            print(f"缓存未命中，缓存大小: {len(cache)}/{maxsize}")
            return value
        # 添加统计方法
        def cache_info():
            return f"命中: {hits}, 未命中: {misses}, 命中率: {hits/(hits+misses)*100:.2f}%"
        wrapper.cache_info = cache_info
        return wrapper
    return decorator
@lru_cache(maxsize=3)
def get_data(id):
    """模拟获取数据"""
    return f"数据{id}"
# 测试
print(get_data(1))  # 未命中
print(get_data(2))  # 未命中
print(get_data(3))  # 未命中
print(get_data(1))  # 命中
print(get_data(4))  # 未命中，移除最久未使用的
print(f"缓存统计: {get_data.cache_info()}")

支持自定义键函数的缓存装饰器

import functools
import hashlib
import json
def custom_cache(key_func=None):
    """支持自定义键函数的缓存装饰器"""
    def decorator(func):
        cache = {}
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            # 使用自定义键函数或默认键
            if key_func:
                cache_key = key_func(*args, **kwargs)
            else:
                # 默认使用参数的JSON序列化作为键
                cache_key = hashlib.md5(
                    json.dumps((args, kwargs), sort_keys=True).encode()
                ).hexdigest()
            if cache_key in cache:
                print(f"缓存命中: {cache_key}")
                return cache[cache_key]
            result = func(*args, **kwargs)
            cache[cache_key] = result
            print(f"创建缓存: {cache_key}")
            return result
        return wrapper
    return decorator
# 自定义键函数：只根据用户ID缓存
def user_cache_key(user_id, **kwargs):
    return f"user_{user_id}"
@custom_cache(key_func=user_cache_key)
def get_user_info(user_id, refresh=False):
    """获取用户信息"""
    return {"id": user_id, "name": f"用户{user_id}", "time": time.time()}
# 测试
print(get_user_info(1))              # 创建缓存
print(get_user_info(1, refresh=True))  # 缓存命中（忽略refresh参数）
print(get_user_info(2))              # 创建新缓存

使用Python内置的LRU缓存

from functools import lru_cache
import time
@lru_cache(maxsize=10)
def expensive_function(n):
    """耗时计算函数"""
    time.sleep(2)  # 模拟耗时操作
    return n * n
# 测试
print("开始计算...")
print(expensive_function(5))  # 首次计算，等待2秒
print(expensive_function(5))  # 立即返回，缓存命中
print(expensive_function(10)) # 新计算，等待2秒
print(expensive_function(5))  # 立即返回
# 查看缓存统计
print(f"缓存信息: {expensive_function.cache_info()}")

异步函数缓存装饰器

import asyncio
import functools
def async_cache(expire_time=60):
    """异步函数缓存装饰器"""
    def decorator(func):
        cache = {}
        @functools.wraps(func)
        async def wrapper(*args, **kwargs):
            key = str(args) + str(kwargs)
            if key in cache:
                value, timestamp = cache[key]
                if time.time() - timestamp < expire_time:
                    print("缓存命中")
                    return value
            value = await func(*args, **kwargs)
            cache[key] = (value, time.time())
            print("创建缓存")
            return value
        return wrapper
    return decorator
@async_cache(expire_time=10)
async def fetch_data(url):
    """模拟异步获取数据"""
    await asyncio.sleep(2)  # 模拟网络请求
    return f"数据来自 {url}"
# 测试
async def main():
    print(await fetch_data("http://example.com"))
    print(await fetch_data("http://example.com"))  # 缓存命中
    print(await fetch_data("http://other.com"))
asyncio.run(main())