Python脚本如何自动清理过期缓存数据

wen python案例 2026-06-27 246

本文目录导读：

基于时间的缓存清理
使用LRU缓存（最近最少使用）
文件系统缓存清理
使用缓存库（带自动清理功能）
完整的定时清理方案
使用建议

我来为您介绍几种Python自动清理过期缓存数据的方法：

基于时间的缓存清理

使用字典存储带时间戳的缓存

import time
import threading
class TimeBasedCache:
    def __init__(self, default_ttl=300):  # 默认5分钟
        self.cache = {}
        self.default_ttl = default_ttl
        self._start_cleanup_thread()
    def set(self, key, value, ttl=None):
        ttl = ttl or self.default_ttl
        self.cache[key] = {
            'value': value,
            'expire_time': time.time() + ttl
        }
    def get(self, key):
        if key not in self.cache:
            return None
        item = self.cache[key]
        if time.time() > item['expire_time']:
            del self.cache[key]
            return None
        return item['value']
    def cleanup(self):
        """清理过期缓存"""
        now = time.time()
        expired_keys = [
            key for key, item in self.cache.items()
            if now > item['expire_time']
        ]
        for key in expired_keys:
            del self.cache[key]
    def _start_cleanup_thread(self):
        """启动后台清理线程"""
        def cleanup_loop():
            while True:
                self.cleanup()
                time.sleep(60)  # 每分钟检查一次
        thread = threading.Thread(target=cleanup_loop, daemon=True)
        thread.start()

使用LRU缓存（最近最少使用）

from collections import OrderedDict
class LRUCache:
    def __init__(self, capacity=100, ttl=300):
        self.cache = OrderedDict()
        self.capacity = capacity
        self.ttl = ttl
    def get(self, key):
        if key not in self.cache:
            return None
        value, timestamp = self.cache[key]
        if time.time() - timestamp > self.ttl:
            del self.cache[key]
            return None
        # 移动到最近使用
        self.cache.move_to_end(key)
        return value
    def put(self, key, value):
        if key in self.cache:
            del self.cache[key]
        elif len(self.cache) >= self.capacity:
            # 删除最久未使用的
            self.cache.popitem(last=False)
        self.cache[key] = (value, time.time())

文件系统缓存清理

import os
import shutil
import glob
from datetime import datetime, timedelta
class FileCacheCleaner:
    def __init__(self, cache_dir, max_age_days=7, max_size_mb=500):
        self.cache_dir = cache_dir
        self.max_age = timedelta(days=max_age_days)
        self.max_size = max_size_mb * 1024 * 1024
        os.makedirs(cache_dir, exist_ok=True)
    def clean_by_age(self):
        """按文件年龄清理"""
        now = datetime.now()
        for root, dirs, files in os.walk(self.cache_dir):
            for file in files:
                file_path = os.path.join(root, file)
                mtime = datetime.fromtimestamp(os.path.getmtime(file_path))
                if now - mtime > self.max_age:
                    os.remove(file_path)
                    print(f"已删除过期文件: {file_path}")
    def clean_by_size(self):
        """按总大小清理"""
        total_size = 0
        file_list = []
        for root, dirs, files in os.walk(self.cache_dir):
            for file in files:
                file_path = os.path.join(root, file)
                size = os.path.getsize(file_path)
                mtime = os.path.getmtime(file_path)
                file_list.append((mtime, file_path, size))
                total_size += size
        # 按修改时间排序，删除最旧的文件
        if total_size > self.max_size:
            file_list.sort()
            for mtime, file_path, size in file_list:
                if total_size <= self.max_size:
                    break
                os.remove(file_path)
                total_size -= size
                print(f"已删除文件以释放空间: {file_path}")
    def clean_pattern(self, pattern):
        """按模式清理"""
        for file_path in glob.glob(os.path.join(self.cache_dir, pattern)):
            os.remove(file_path)
            print(f"已删除匹配文件: {file_path}")

使用缓存库（带自动清理功能）

使用 cachetools

from cachetools import TTLCache
import time
# 创建自动过期的缓存
cache = TTLCache(maxsize=100, ttl=300)  # 最多100个，存活5分钟
# 使用缓存
cache['user_1'] = {'name': 'Alice', 'age': 30}
print(cache.get('user_1'))  # 5分钟内有效
# 自定义清理（可配合定时任务）
cache.expire()  # 手动触发过期清理

使用 diskcache

import diskcache as dc
# 基于磁盘的缓存
cache = dc.Cache('/tmp/my_cache')
# 设置自动过期时间
cache.set('key', 'value', expire=300)  # 5分钟后过期
# 自动清理（后台线程自动执行）
cache.expire()  # 运行清理
cache.cull()   # 清理过期项和超出容量的项

完整的定时清理方案

import schedule
import time
from functools import wraps
class AutoCleanCache:
    def __init__(self, cleanup_interval_minutes=30):
        self.cache = {}
        self.cleanup_interval = cleanup_interval_minutes
        self.start_scheduler()
    def cached(self, ttl=300):
        """装饰器：自动缓存并清理"""
        def decorator(func):
            @wraps(func)
            def wrapper(*args, **kwargs):
                key = f"{func.__name__}:{args}:{kwargs}"
                now = time.time()
                if key in self.cache:
                    value, expire_time = self.cache[key]
                    if now < expire_time:
                        return value
                result = func(*args, **kwargs)
                self.cache[key] = (result, now + ttl)
                return result
            return wrapper
        return decorator
    def cleanup_expired(self):
        """清理过期缓存"""
        now = time.time()
        expired = [k for k, (v, exp) in self.cache.items() if now > exp]
        for k in expired:
            del self.cache[k]
        print(f"已清理 {len(expired)} 条过期缓存")
    def start_scheduler(self):
        """启动定时清理任务"""
        schedule.every(self.cleanup_interval).minutes.do(self.cleanup_expired)
        def run_scheduler():
            while True:
                schedule.run_pending()
                time.sleep(1)
        import threading
        thread = threading.Thread(target=run_scheduler, daemon=True)
        thread.start()
# 使用示例
cache_manager = AutoCleanCache(cleanup_interval_minutes=15)
@cache_manager.cached(ttl=120)  # 缓存2分钟
def get_expensive_data(id):
    # 模拟耗时操作
    time.sleep(2)
    return {"id": id, "data": "expensive data"}