【Python3】装饰器 自动更新缓存

自动更新缓存的需求场景

在某些应用中，我们可能需要定期从外部数据源（如 Redis 或者远程接口）拉取数据，并将其缓存在内存中。当有其他代码需要访问这些数据时，可以立刻从内存获取最新数据，而无需每次都进行耗时的外部操作。

关键思路

1. 初次加载：程序启动时立即执行目标函数，从数据源获取数据并将结果缓存。
2. 定时更新：借助定时任务调度器（如 APScheduler），在指定的时间间隔（如30秒）自动再次执行目标函数，刷新缓存中的数据。
3. 快速访问：对外暴露的函数调用时直接返回缓存中的数据，不会再次执行耗时的外部操作，从而实现快速访问。

使用装饰器实现

我们使用一个自定义的装饰器 @auto_update(update_time=30) 来封装这一逻辑：

• 装饰器初始化：
  当程序加载被装饰的函数时，装饰器会先执行一次目标函数，将返回值存入缓存。

• 定时任务调度：
  使用 APScheduler 的 BackgroundScheduler 来定期调用该函数更新缓存数据。APScheduler 可以独立运行后台线程，不会阻塞主程序的其他逻辑。

• 缓存访问：
  被装饰的函数在对外调用时，不再直接执行原始函数，而是直接返回缓存中的数据。这样，在任意时刻调用该函数，都可以瞬间获取最新数据。

代码

import time
from functools import wraps
from apscheduler.schedulers.background import BackgroundScheduler# 创建并启动全局调度器
scheduler = BackgroundScheduler()
scheduler.start()def auto_update(update_time=30):def decorator(func):cache = {"value": None, "initialized": False}def update_cache():new_value = func()cache["value"] = new_valuecache["initialized"] = Trueprint(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] Cache updated.")# 启动时先加载一次update_cache()# 每隔 update_time 秒自动刷新数据scheduler.add_job(update_cache, 'interval', seconds=update_time)@wraps(func)def wrapper(*args, **kwargs):return cache["value"]return wrapperreturn decorator@auto_update(update_time=30)
def load_data_from_redis():# 模拟从Redis获取数据的函数（实际中可替换为耗时的外部操作）return f"data_from_redis_{int(time.time())}"# 当你调用 load_data_from_redis() 时，能立即获得最新的缓存数据
print("Initial data:", load_data_from_redis())
time.sleep(35)
print("Data after 35s:", load_data_from_redis())

总结

通过上述装饰器和 APScheduler 的组合，你就能轻松实现：

• 程序启动即从外部数据源加载数据到缓存；
• 周期性、自动地刷新缓存中的数据；
• 在任意时间调用时都能快速获取最新数据，而无需阻塞或消耗额外时间。

---