Python脚本如何定时刷新热点缓存内容

wen python案例 2026-06-26 30

本文目录导读：

使用 schedule 库（简单常用）
使用 threading.Timer（轻量级）
使用 APScheduler（功能强大的调度器）
使用 asyncio（异步方式）
企业级实现（带错误处理和监控）
选择建议
常见缓存刷新策略

我来介绍几种在Python中定时刷新热点缓存内容的方法：

使用 `schedule` 库（简单常用）

import schedule
import time
import redis
import json
# 连接Redis
r = redis.Redis(host='localhost', port=6379, db=0)
def refresh_hot_cache():
    """刷新热点缓存"""
    print(f"开始刷新热点缓存... [{time.strftime('%Y-%m-%d %H:%M:%S')}]")
    # 示例：刷新热点数据
    hot_data = get_hot_data_from_database()
    # 更新Redis缓存
    r.set('hot_cache', json.dumps(hot_data))
    r.expire('hot_cache', 3600)  # 设置过期时间1小时
    print("热点缓存刷新完成")
# 每隔5分钟执行一次
schedule.every(5).minutes.do(refresh_hot_cache)
# 或者每天固定时间执行
# schedule.every().day.at("06:00").do(refresh_hot_cache)
# 或者每小时执行
# schedule.every().hour.do(refresh_hot_cache)
while True:
    schedule.run_pending()
    time.sleep(1)

使用 `threading.Timer`（轻量级）

import threading
import time
from datetime import datetime
class HotCacheRefresher:
    def __init__(self, interval_seconds=300):
        self.interval = interval_seconds
        self.timer = None
    def refresh(self):
        """执行刷新任务"""
        print(f"[{datetime.now()}] 刷新热点缓存...")
        # 模拟刷新数据
        # your_refresh_logic()
        # 重新设置定时器
        self.start()
    def start(self):
        """启动定时器"""
        self.timer = threading.Timer(self.interval, self.refresh)
        self.timer.daemon = True
        self.timer.start()
    def stop(self):
        """停止定时器"""
        if self.timer:
            self.timer.cancel()
# 使用示例
refresher = HotCacheRefresher(interval_seconds=300)
refresher.start()

使用 `APScheduler`（功能强大的调度器）

# 安装
pip install apscheduler

from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.triggers.interval import IntervalTrigger
from apscheduler.triggers.cron import CronTrigger
import time
import logging
logging.basicConfig(level=logging.INFO)
def refresh_hot_cache():
    """刷新热点缓存"""
    # 获取热点数据
    hot_data = fetch_hot_data()
    # 更新缓存
    update_cache('hot_key', hot_data)
    print(f"缓存刷新完成: {time.ctime()}")
def fetch_hot_data():
    """模拟获取热点数据"""
    return {
        'items': ['data1', 'data2', 'data3'],
        'timestamp': time.time()
    }
def update_cache(key, data):
    """模拟更新缓存"""
    import json
    # 这里替换为实际的缓存更新逻辑
    print(f"更新缓存 {key}: {json.dumps(data)[:50]}...")
# 创建调度器
scheduler = BackgroundScheduler()
# 每隔5分钟执行一次
scheduler.add_job(
    refresh_hot_cache,
    IntervalTrigger(seconds=300),
    id='hot_cache_job'
)
# 或者使用Cron表达式（每天凌晨2点和下午2点执行）
# scheduler.add_job(
#     refresh_hot_cache,
#     CronTrigger(hour='2,14', minute=0),
#     id='hot_cache_job'
# )
# 或者每隔一段时间（例如30分钟）
# scheduler.add_job(
#     refresh_hot_cache,
#     IntervalTrigger(minutes=30),
#     id='hot_cache_job'
# )
# 启动调度器
scheduler.start()
try:
    # 让程序保持运行
    while True:
        time.sleep(1)
except KeyboardInterrupt:
    scheduler.shutdown()

使用 `asyncio`（异步方式）

import asyncio
import time
from datetime import datetime
class AsyncHotCacheRefresher:
    def __init__(self, interval=300):
        self.interval = interval
        self.running = False
    async def refresh_cache(self):
        """异步刷新缓存"""
        print(f"[{datetime.now()}] 异步刷新缓存开始...")
        # 模拟异步IO操作
        await asyncio.sleep(0.1)  # 实际替换为数据库查询
        # 更新缓存
        await self.update_hot_cache()
        print(f"[{datetime.now()}] 缓存刷新完成")
    async def update_hot_cache(self):
        """模拟异步更新缓存"""
        # 实际实现替换为Redis或内存缓存更新
        await asyncio.sleep(0.05)
        print("缓存更新成功")
    async def start(self):
        """启动异步刷新循环"""
        self.running = True
        while self.running:
            await self.refresh_cache()
            await asyncio.sleep(self.interval)
    def stop(self):
        """停止刷新"""
        self.running = False
# 运行
async def main():
    refresher = AsyncHotCacheRefresher(interval=300)
    # 创建多个刷新任务
    tasks = [
        refresher.start(),
        other_task()
    ]
    await asyncio.gather(*tasks)
# asyncio.run(main())

企业级实现（带错误处理和监控）

import time
import logging
from functools import wraps
from datetime import datetime
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class HotCacheManager:
    def __init__(self, refresh_interval=300):
        self.refresh_interval = refresh_interval
        self.last_refresh_time = None
        self.cache_data = None
    def retry_on_failure(max_retries=3, delay=5):
        """重试装饰器"""
        def decorator(func):
            @wraps(func)
            def wrapper(*args, **kwargs):
                for attempt in range(max_retries):
                    try:
                        return func(*args, **kwargs)
                    except Exception as e:
                        logger.error(f"刷新失败 (尝试 {attempt+1}/{max_retries}): {e}")
                        if attempt < max_retries - 1:
                            time.sleep(delay)
                raise Exception(f"刷新失败，已重试{max_retries}次")
            return wrapper
        return decorator
    @retry_on_failure(max_retries=3)
    def refresh_cache(self):
        """刷新缓存（带重试机制）"""
        logger.info(f"开始刷新热点缓存... [{datetime.now()}]")
        try:
            # 获取热点数据
            hot_data = self.fetch_hot_data()
            # 验证数据有效性
            if not self.validate_data(hot_data):
                raise ValueError("数据验证失败")
            # 更新内存缓存
            self.cache_data = hot_data
            self.last_refresh_time = datetime.now()
            # 更新外部缓存（Redis等）
            self.update_external_cache(hot_data)
            logger.info(f"缓存刷新成功，数据大小: {len(str(hot_data))} bytes")
        except Exception as e:
            logger.error(f"缓存刷新失败: {e}")
            raise
    def fetch_hot_data(self):
        """从数据源获取热点数据"""
        # 实现具体的数据获取逻辑
        return {
            'timestamp': time.time(),
            'data': ['hot_item_1', 'hot_item_2']
        }
    def validate_data(self, data):
        """验证数据有效性"""
        return data is not None and len(data.get('data', [])) > 0
    def update_external_cache(self, data):
        """更新外部缓存"""
        # 实现Redis等外部缓存更新
        pass
    def get_cache(self):
        """获取缓存数据"""
        if self.should_refresh():
            self.refresh_cache()
        return self.cache_data
    def should_refresh(self):
        """判断是否需要刷新"""
        if self.last_refresh_time is None:
            return True
        elapsed = (datetime.now() - self.last_refresh_time).total_seconds()
        return elapsed >= self.refresh_interval
    def run_scheduler(self):
        """运行定时调度器"""
        while True:
            try:
                self.refresh_cache()
                time.sleep(self.refresh_interval)
            except Exception as e:
                logger.error(f"调度器错误: {e}")
                time.sleep(30)  # 出错后等待30秒重试
# 使用示例
if __name__ == "__main__":
    cache_manager = HotCacheManager(refresh_interval=300)
    cache_manager.run_scheduler()

选择建议

方法	适用场景	优点	缺点
`schedule`	简单定时任务	易用，代码简洁	功能较少
`threading.Timer`	轻量级定时	内存占用小	精确度一般
`APScheduler`	企业级应用	功能强大，支持cron	依赖较多
`asyncio`	异步程序	高性能，非阻塞	需要理解异步编程