网站 微信 app,商标注册官网入口官网,网站建设 dw 时间轴,上海网站建设千元漂亮自动更新缓存的需求场景
在某些应用中#xff0c;我们可能需要定期从外部数据源#xff08;如 Redis 或者远程接口#xff09;拉取数据#xff0c;并将其缓存在内存中。当有其他代码需要访问这些数据时#xff0c;可以立刻从内存获取最新数据#xff0c;而无需每次都进行…自动更新缓存的需求场景
在某些应用中我们可能需要定期从外部数据源如 Redis 或者远程接口拉取数据并将其缓存在内存中。当有其他代码需要访问这些数据时可以立刻从内存获取最新数据而无需每次都进行耗时的外部操作。
关键思路
初次加载程序启动时立即执行目标函数从数据源获取数据并将结果缓存。定时更新借助定时任务调度器如 APScheduler在指定的时间间隔如30秒自动再次执行目标函数刷新缓存中的数据。快速访问对外暴露的函数调用时直接返回缓存中的数据不会再次执行耗时的外部操作从而实现快速访问。
使用装饰器实现
我们使用一个自定义的装饰器 auto_update(update_time30) 来封装这一逻辑 装饰器初始化 当程序加载被装饰的函数时装饰器会先执行一次目标函数将返回值存入缓存。 定时任务调度 使用 APScheduler 的 BackgroundScheduler 来定期调用该函数更新缓存数据。APScheduler 可以独立运行后台线程不会阻塞主程序的其他逻辑。 缓存访问 被装饰的函数在对外调用时不再直接执行原始函数而是直接返回缓存中的数据。这样在任意时刻调用该函数都可以瞬间获取最新数据。
代码
import time
from functools import wraps
from apscheduler.schedulers.background import BackgroundScheduler# 创建并启动全局调度器
scheduler BackgroundScheduler()
scheduler.start()def auto_update(update_time30):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, secondsupdate_time)wraps(func)def wrapper(*args, **kwargs):return cache[value]return wrapperreturn decoratorauto_update(update_time30)
def load_data_from_redis():# 模拟从Redis获取数据的函数实际中可替换为耗时的外部操作return fdata_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 的组合你就能轻松实现
程序启动即从外部数据源加载数据到缓存周期性、自动地刷新缓存中的数据在任意时间调用时都能快速获取最新数据而无需阻塞或消耗额外时间。
