本文目录导读：
我来提供几种批量刷新过期缓存数据的Python脚本方案：
Redis缓存批量刷新
使用SCAN命令清理过期缓存
import redis
import time
from typing import List, Optional
class RedisCacheRefresh:
def __init__(self, host='localhost', port=6379, db=0, password=None):
self.client = redis.Redis(
host=host,
port=port,
db=db,
password=password,
decode_responses=True
)
def flush_expired_keys(self, pattern: str = '*', batch_size: int = 100) -> int:
"""
批量刷新过期缓存
:param pattern: 匹配的键模式
:param batch_size: 每批处理的键数量
:return: 删除的键数量
"""
cursor = 0
deleted_count = 0
while True:
# 使用SCAN命令分批获取键
cursor, keys = self.client.scan(
cursor=cursor,
match=pattern,
count=batch_size
)
if keys:
# 检查每个键是否过期
for key in keys:
ttl = self.client.ttl(key)
if ttl <= 0: # 已过期或不存在
self.client.delete(key)
deleted_count += 1
print(f"删除过期缓存: {key}")
# 游标为0表示遍历完成
if cursor == 0:
break
return deleted_count
def refresh_with_ttl_check(self, pattern: str = '*', min_ttl: int = 3600) -> List[str]:
"""
刷新TTL小于指定值的缓存
:param pattern: 匹配模式
:param min_ttl: 最小TTL阈值（秒）
:return: 刷新的键列表
"""
refreshed_keys = []
cursor = 0
while True:
cursor, keys = self.client.scan(cursor=cursor, match=pattern, count=100)
for key in keys:
ttl = self.client.ttl(key)
if 0 < ttl < min_ttl: # 存在但TTL小于阈值
# 重新设置缓存（这里需要根据实际业务逻辑实现）
self._refresh_cache(key)
refreshed_keys.append(key)
print(f"刷新缓存: {key}, 原TTL: {ttl}s")
if cursor == 0:
break
return refreshed_keys
def _refresh_cache(self, key: str):
"""根据键重新生成缓存数据（需要根据业务逻辑实现）"""
# 示例：重新设置TTL为1小时
# 实际应用中应该重新计算数据并设置
self.client.expire(key, 3600)
# 使用示例
cache_refresher = RedisCacheRefresh(host='localhost', port=6379)
deleted = cache_refresher.flush_expired_keys(pattern='user:*')
print(f"共删除 {deleted} 个过期缓存")
本地文件缓存批量刷新
import os
import json
import time
from pathlib import Path
from typing import Dict, Any
class FileCacheRefresh:
def __init__(self, cache_dir: str = './cache'):
self.cache_dir = Path(cache_dir)
self.cache_dir.mkdir(exist_ok=True)
def clean_expired_files(self, max_age_hours: int = 24) -> int:
"""
清理过期缓存文件
:param max_age_hours: 文件最大存活时间（小时）
:return: 删除的文件数量
"""
current_time = time.time()
deleted_count = 0
for file_path in self.cache_dir.iterdir():
if file_path.is_file():
# 检查文件最后修改时间
file_age = current_time - file_path.stat().st_mtime
max_age_seconds = max_age_hours * 3600
if file_age > max_age_seconds:
file_path.unlink()
deleted_count += 1
print(f"删除过期缓存文件: {file_path}")
return deleted_count
def refresh_cache_files(self, pattern: str = '*.json',
refresh_func=None) -> int:
"""
批量刷新缓存文件
:param pattern: 文件匹配模式
:param refresh_func: 刷新函数，接收文件路径参数
:return: 刷新的文件数量
"""
refreshed_count = 0
for file_path in self.cache_dir.glob(pattern):
try:
# 读取当前缓存数据
with open(file_path, 'r', encoding='utf-8') as f:
data = json.load(f)
# 检查是否过期
if self._is_expired(data):
# 执行刷新策略
if refresh_func:
new_data = refresh_func(file_path)
else:
new_data = self._default_refresh(file_path)
# 保存更新后的数据
with open(file_path, 'w', encoding='utf-8') as f:
json.dump(new_data, f, ensure_ascii=False, indent=2)
refreshed_count += 1
print(f"刷新缓存文件: {file_path}")
except Exception as e:
print(f"刷新文件 {file_path} 失败: {e}")
return refreshed_count
def _is_expired(self, data: Dict[str, Any]) -> bool:
"""检查缓存数据是否过期"""
# 假设数据中包含timestamp字段
if 'timestamp' in data:
current_time = time.time()
return current_time - data['timestamp'] > 3600 # 1小时过期
return True
def _default_refresh(self, file_path: Path) -> Dict:
"""默认刷新策略"""
data = {
'timestamp': time.time(),
'data': f'refreshed at {time.strftime("%Y-%m-%d %H:%M:%S")}'
}
return data
# 使用示例
file_cache = FileCacheRefresh(cache_dir='./data_cache')
deleted_files = file_cache.clean_expired_files(max_age_hours=48)
print(f"清理了 {deleted_files} 个过期缓存文件")
数据库查询缓存批量刷新
import sqlite3
import time
from datetime import datetime, timedelta
from typing import List, Tuple, Optional
class DatabaseCacheRefresh:
def __init__(self, db_path: str = 'cache.db'):
self.conn = sqlite3.connect(db_path)
self.cursor = self.conn.cursor()
self._init_cache_table()
def _init_cache_table(self):
"""初始化缓存表"""
self.cursor.execute('''
CREATE TABLE IF NOT EXISTS cache_data (
key TEXT PRIMARY KEY,
value BLOB,
created_at TIMESTAMP,
expires_at TIMESTAMP,
last_accessed TIMESTAMP
)
''')
self.conn.commit()
def clean_expired_cache(self) -> int:
"""
清理所有过期缓存
:return: 清理的缓存数量
"""
current_time = datetime.now()
self.cursor.execute(
'DELETE FROM cache_data WHERE expires_at < ?',
(current_time,)
)
deleted_count = self.cursor.rowcount
self.conn.commit()
return deleted_count
def refresh_stale_cache(self, stale_hours: int = 24) -> List[str]:
"""
刷新陈旧的缓存（长时间未访问）
:param stale_hours: 缓存被认为陈旧的小时数
:return: 刷新的缓存键列表
"""
stale_time = datetime.now() - timedelta(hours=stale_hours)
# 查找陈旧缓存
self.cursor.execute('''
SELECT key, value FROM cache_data
WHERE last_accessed < ? OR last_accessed IS NULL
''', (stale_time,))
stale_records = self.cursor.fetchall()
refreshed_keys = []
for key, old_value in stale_records:
# 执行刷新逻辑
new_value = self._refresh_cache_value(key, old_value)
# 更新缓存
current_time = datetime.now()
new_expiry = current_time + timedelta(hours=48) # 新的过期时间
self.cursor.execute('''
UPDATE cache_data
SET value = ?, created_at = ?, expires_at = ?, last_accessed = ?
WHERE key = ?
''', (new_value, current_time, new_expiry, current_time, key))
refreshed_keys.append(key)
print(f"刷新缓存: {key}")
self.conn.commit()
return refreshed_keys
def _refresh_cache_value(self, key: str, old_value: bytes) -> bytes:
"""
根据业务逻辑刷新缓存值
这里需要根据实际应用场景实现
"""
# 示例：模拟新的缓存数据
import json
try:
old_data = json.loads(old_value.decode('utf-8'))
# 更新数据逻辑
old_data['refreshed_at'] = datetime.now().isoformat()
return json.dumps(old_data).encode('utf-8')
except:
return old_value
def batch_refresh_by_pattern(self, pattern: str = '%') -> int:
"""
按模式批量刷新缓存
:param pattern: SQL LIKE模式
:return: 刷新的缓存数量
"""
self.cursor.execute('''
SELECT key FROM cache_data WHERE key LIKE ?
''', (pattern,))
keys = [row[0] for row in self.cursor.fetchall()]
refreshed_count = 0
for key in keys:
# 执行全量刷新
current_time = datetime.now()
new_value = f'refreshed_{key}_{current_time.timestamp()}'.encode()
new_expiry = current_time + timedelta(hours=48)
self.cursor.execute('''
UPDATE cache_data
SET value = ?, created_at = ?, expires_at = ?, last_accessed = ?
WHERE key = ?
''', (new_value, current_time, new_expiry, current_time, key))
refreshed_count += 1
self.conn.commit()
return refreshed_count
def close(self):
"""关闭数据库连接"""
self.conn.close()
# 使用示例
db_cache = DatabaseCacheRefresh('my_cache.db')
deleted = db_cache.clean_expired_cache()
print(f"清理了 {deleted} 个过期缓存")
refreshed = db_cache.refresh_stale_cache(stale_hours=12)
print(f"刷新了 {len(refreshed)} 个陈旧缓存")
db_cache.close()
通用的缓存刷新管理器
import asyncio
import logging
from typing import Callable, Dict, Any
from datetime import datetime
class CacheRefreshManager:
"""通用缓存刷新管理器"""
def __init__(self):
self.refresh_strategies: Dict[str, Callable] = {}
self.logger = logging.getLogger(__name__)
def register_strategy(self, name: str, strategy: Callable):
"""注册刷新策略"""
self.refresh_strategies[name] = strategy
async def refresh_cache(self, strategy_name: str, **kwargs) -> bool:
"""执行单个缓存刷新"""
if strategy_name not in self.refresh_strategies:
raise ValueError(f"未知策略: {strategy_name}")
try:
start_time = datetime.now()
result = await self.refresh_strategies[strategy_name](**kwargs)
elapsed = (datetime.now() - start_time).total_seconds()
self.logger.info(f"策略 {strategy_name} 刷新完成，耗时: {elapsed:.2f}s")
return result
except Exception as e:
self.logger.error(f"策略 {strategy_name} 刷新失败: {e}")
return False
async def refresh_all(self, **kwargs) -> Dict[str, bool]:
"""刷新所有注册的缓存策略"""
results = {}
tasks = []
for strategy_name in self.refresh_strategies:
tasks.append(
asyncio.create_task(
self.refresh_cache(strategy_name, **kwargs)
)
)
completed = await asyncio.gather(*tasks, return_exceptions=True)
for i, strategy_name in enumerate(self.refresh_strategies.keys()):
results[strategy_name] = not isinstance(completed[i], Exception)
return results
def run_batch_refresh(self, strategies: list = None, **kwargs):
"""运行批量的缓存刷新任务"""
if strategies:
selected_strategies = {
name: self.refresh_strategies[name]
for name in strategies
if name in self.refresh_strategies
}
else:
selected_strategies = self.refresh_strategies
results = {}
for name, strategy in selected_strategies.items():
try:
result = strategy(**kwargs)
results[name] = result
print(f"策略 {name} 刷新成功: {result}")
except Exception as e:
results[name] = None
print(f"策略 {name} 刷新失败: {e}")
return results
# 自定义刷新策略示例
def refresh_user_cache(**kwargs):
"""刷新用户缓存策略"""
# 实现具体的用户缓存刷新逻辑
print("正在刷新用户缓存...")
return True
def refresh_product_cache(**kwargs):
"""刷新商品缓存策略"""
# 实现具体的商品缓存刷新逻辑
print("正在刷新商品缓存...")
return True
# 使用示例
manager = CacheRefreshManager()
manager.register_strategy('user_cache', refresh_user_cache)
manager.register_strategy('product_cache', refresh_product_cache)
# 批量刷新所有缓存
results = manager.run_batch_refresh()
print("批量刷新结果:", results)
# 选择特定缓存刷新
results = manager.run_batch_refresh(strategies=['user_cache'])
使用建议
- 选择合适的刷新策略：根据缓存类型（Redis、文件、数据库）选择对应方案
- 设置适当的执行频率：可以使用定时任务（cron、schedule）定期执行
- 添加错误处理和日志记录：确保刷新过程可追踪和调试
- 考虑并发控制：使用锁机制避免重复刷新
- 监控刷新性能：记录刷新时间和效率，优化策略
选择适合你应用场景的缓存刷新方案,并根据实际需求进行调整。
版权声明：NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
