本文目录导读：
我来为您介绍几种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"}
使用建议
- 小规模应用：使用字典 + 时间戳方式
- 大规模应用：考虑 Redis 等外部缓存系统
- 文件缓存：定期扫描文件系统
- Web应用：使用 Flask-Caching、Django缓存框架
- 关键业务：确保清理逻辑不影响正常服务
选择合适的方案取决于您的具体需求和缓存的数据量级。
