本文目录导读：
我来为您提供一个归档冷门缓存到低速存储的Python脚本方案：
核心脚本实现
#!/usr/bin/env python3
"""
冷门缓存归档工具 - 将使用频率低的缓存文件迁移到低速存储
"""
import os
import shutil
import time
import json
import logging
from pathlib import Path
from datetime import datetime, timedelta
from typing import List, Dict, Optional
import hashlib
# 配置日志
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('cache_archiver.log'),
logging.StreamHandler()
]
)
class CacheArchiver:
def __init__(self, config: Dict):
"""
初始化归档器
Args:
config: 配置字典，包含以下键：
- fast_cache_path: 高速缓存路径
- slow_cache_path: 低速存储路径
- metadata_file: 元数据文件路径
- cold_threshold_days: 冷门判定天数
- max_cache_size_gb: 高速缓存最大容量(GB)
- archive_batch_size: 每次归档的文件数量
"""
self.fast_path = Path(config['fast_cache_path'])
self.slow_path = Path(config['slow_cache_path'])
self.metadata_file = Path(config.get('metadata_file', 'cache_metadata.json'))
self.cold_threshold = timedelta(days=config.get('cold_threshold_days', 30))
self.max_cache_size = config.get('max_cache_size_gb', 100) * 1024 * 1024 * 1024
self.batch_size = config.get('archive_batch_size', 100)
# 确保目录存在
self.fast_path.mkdir(parents=True, exist_ok=True)
self.slow_path.mkdir(parents=True, exist_ok=True)
# 加载元数据
self.metadata = self._load_metadata()
def _load_metadata(self) -> Dict:
"""加载缓存文件元数据"""
if self.metadata_file.exists():
try:
with open(self.metadata_file, 'r') as f:
return json.load(f)
except Exception as e:
logging.error(f"加载元数据失败: {e}")
return {}
return {}
def _save_metadata(self):
"""保存缓存文件元数据"""
try:
with open(self.metadata_file, 'w') as f:
json.dump(self.metadata, f, indent=2)
except Exception as e:
logging.error(f"保存元数据失败: {e}")
def _get_file_access_time(self, file_path: Path) -> Optional[datetime]:
"""获取文件最后访问时间"""
try:
stat = file_path.stat()
return datetime.fromtimestamp(stat.st_atime)
except Exception as e:
logging.error(f"获取文件访问时间失败 {file_path}: {e}")
return None
def _calculate_file_hash(self, file_path: Path) -> str:
"""计算文件哈希值用于唯一标识"""
try:
hasher = hashlib.md5()
with open(file_path, 'rb') as f:
# 只读取文件头部和尾部来平衡性能和准确性
hasher.update(f.read(8192))
f.seek(-8192, 2)
hasher.update(f.read())
return hasher.hexdigest()
except Exception as e:
logging.error(f"计算文件哈希失败 {file_path}: {e}")
return str(file_path.name)
def scan_cache_files(self) -> List[Dict]:
"""扫描高速缓存中的文件"""
files_info = []
for file_path in self.fast_path.rglob('*'):
if file_path.is_file() and not file_path.name.startswith('.'):
file_info = {
'path': str(file_path),
'name': file_path.name,
'size': file_path.stat().st_size,
'last_access': self._get_file_access_time(file_path),
'hash': self._calculate_file_hash(file_path)
}
files_info.append(file_info)
return files_info
def identify_cold_files(self, files_info: List[Dict]) -> List[Dict]:
"""
识别冷门缓存文件
规则：
1. 超过冷门阈值的文件
2. 在元数据中标记为低访问频率的文件
"""
cold_files = []
now = datetime.now()
for file_info in files_info:
# 检查最后访问时间
if file_info['last_access']:
age = now - file_info['last_access']
# 更新元数据中的访问信息
file_hash = file_info['hash']
if file_hash not in self.metadata:
self.metadata[file_hash] = {
'access_count': 0,
'first_seen': now.isoformat(),
'last_access': file_info['last_access'].isoformat()
}
# 判断是否为冷门文件
if age > self.cold_threshold:
cold_files.append(file_info)
return cold_files
def calculate_cache_size(self) -> int:
"""计算当前缓存总大小"""
total_size = 0
for file_path in self.fast_path.rglob('*'):
if file_path.is_file():
total_size += file_path.stat().st_size
return total_size
def archive_files(self, files_to_archive: List[Dict]):
"""
归档文件到低速存储
归档策略：
1. 保持目录结构
2. 创建符号链接或快捷方式
3. 更新元数据
"""
archived_count = 0
for file_info in files_to_archive:
try:
src_path = Path(file_info['path'])
# 计算相对路径以保持结构
rel_path = src_path.relative_to(self.fast_path)
dst_path = self.slow_path / rel_path
# 确保目标目录存在
dst_path.parent.mkdir(parents=True, exist_ok=True)
# 移动文件到低速存储
shutil.move(str(src_path), str(dst_path))
# 在原始位置创建符号链接（可选）
try:
os.symlink(dst_path, src_path)
except:
# 如果符号链接失败，创建标记文件
marker_path = src_path.with_suffix('.archived')
marker_path.write_text(str(dst_path))
# 更新元数据
file_hash = file_info['hash']
if file_hash in self.metadata:
self.metadata[file_hash]['archived'] = True
self.metadata[file_hash]['archived_path'] = str(dst_path)
self.metadata[file_hash]['archived_time'] = datetime.now().isoformat()
archived_count += 1
logging.info(f"归档文件: {file_info['name']} ({archived_count}/{len(files_to_archive)})")
except Exception as e:
logging.error(f"归档失败 {file_info['name']}: {e}")
# 保存更新后的元数据
self._save_metadata()
return archived_count
def run_archival_process(self):
"""运行完整的归档流程"""
logging.info("开始冷门缓存归档流程")
# 1. 扫描缓存文件
logging.info("扫描高速缓存文件...")
files_info = self.scan_cache_files()
logging.info(f"发现 {len(files_info)} 个缓存文件")
# 2. 检查缓存大小
current_size = self.calculate_cache_size()
logging.info(f"当前缓存大小: {current_size / (1024**3):.2f} GB")
# 3. 识别冷门文件
logging.info("识别冷门缓存文件...")
cold_files = self.identify_cold_files(files_info)
logging.info(f"发现 {len(cold_files)} 个冷门文件")
# 4. 判断是否需要归档
if cold_files and current_size > self.max_cache_size * 0.8: # 达到80%容量触发
# 根据最后访问时间排序
cold_files.sort(key=lambda x: x['last_access'] if x['last_access'] else datetime.min)
# 分批归档
for i in range(0, len(cold_files), self.batch_size):
batch = cold_files[i:i + self.batch_size]
archived = self.archive_files(batch)
logging.info(f"批次归档完成: {archived} 个文件")
# 重新检查缓存大小
current_size = self.calculate_cache_size()
if current_size <= self.max_cache_size * 0.6: # 降到60%以下停止
break
else:
logging.info("缓存空间充足，无需归档")
class SmartCacheManager:
"""智能缓存管理器 - 自动管理缓存策略"""
def __init__(self):
self.archiver = None
def configure_smart_policy(self):
"""配置智能缓存策略"""
# 根据文件类型设置不同的冷门阈值
self.policies = {
'.tmp': {'threshold_days': 1, 'priority': 'low'},
'.log': {'threshold_days': 7, 'priority': 'low'},
'.cache': {'threshold_days': 14, 'priority': 'medium'},
'.thumb': {'threshold_days': 30, 'priority': 'high'},
'*': {'threshold_days': 30, 'priority': 'normal'}
}
def adaptive_threshold(self, file_info: Dict) -> int:
"""自适应调整冷门阈值"""
ext = Path(file_info['name']).suffix
policy = self.policies.get(ext, self.policies['*'])
return policy['threshold_days']
def create_default_config() -> Dict:
"""创建默认配置"""
return {
'fast_cache_path': '/path/to/fast/cache',
'slow_cache_path': '/path/to/slow/storage',
'metadata_file': '/path/to/metadata/cache_metadata.json',
'cold_threshold_days': 30,
'max_cache_size_gb': 100,
'archive_batch_size': 100
}
if __name__ == "__main__":
# 使用示例
config = create_default_config()
# 根据实际情况修改路径
config['fast_cache_path'] = './fast_cache'
config['slow_cache_path'] = './slow_cache'
archiver = CacheArchiver(config)
archiver.run_archival_process()
配置文件示例
# config.yaml
cache_archiver:
fast_cache_path: "/data/fast_cache"
slow_cache_path: "/data/slow_cache"
metadata_file: "/data/metadata/cache_metadata.json"
# 冷门判定参数
cold_threshold_days: 30
hot_threshold_hours: 24 # 热数据保留时间
# 缓存大小限制
max_cache_size_gb: 100
min_free_space_gb: 20 # 保留的最小空闲空间
# 归档策略
archive_batch_size: 100
preserve_symlinks: true # 是否保留符号链接
# 文件过滤
exclude_patterns:
- "*.tmp"
- "*.lock"
# 日志配置
log_level: "INFO"
log_file: "/var/log/cache_archiver.log"
使用说明
基本用法
# 命令行使用 python cache_archiver.py # 定时任务配置 (crontab) 0 2 * * * /usr/bin/python3 /path/to/cache_archiver.py
高级功能
-
智能冷门识别
- 基于文件类型设置不同的阈值
- 分析访问模式（频率、时间分布）
-
渐进式归档
- 按文件大小、访问时间优先级排序
- 控制归档速度和网络/IO负载
-
回滚机制
def restore_archived_file(self, file_hash: str): """从低速存储恢复文件到高速缓存""" if file_hash in self.metadata and self.metadata[file_hash].get('archived'): src_path = Path(self.metadata[file_hash]['archived_path']) dst_path = Path(self.metadata[file_hash]['original_path']) shutil.copy2(src_path, dst_path) # 移除符号链接或标记
最佳实践
-
监控与告警
- 监控缓存命中率
- 设置缓存大小告警阈值
-
性能优化
- 使用多线程处理大文件
- 批量IO操作减少磁盘开销
-
数据安全
- 归档前验证文件完整性
- 保留归档日志用于审计
这个脚本可以根据您的具体需求进行调整，比如添加特定的归档策略、调整冷门判定算法等。
