本文目录导读：
我来介绍几种Python迁移热点缓存到高速存储的方法：
基于访问频率的缓存迁移
import os
import shutil
import time
import json
from collections import defaultdict
from pathlib import Path
class HotCacheMigrator:
def __init__(self, slow_path, fast_path, threshold=10):
self.slow_path = Path(slow_path)
self.fast_path = Path(fast_path)
self.threshold = threshold # 访问次数阈值
self.access_count = defaultdict(int)
self.cache_file = self.fast_path / ".cache_meta.json"
# 确保目录存在
self.slow_path.mkdir(parents=True, exist_ok=True)
self.fast_path.mkdir(parents=True, exist_ok=True)
def record_access(self, filename):
"""记录文件访问"""
self.access_count[filename] += 1
def check_and_migrate(self):
"""检查并迁移热点文件"""
migrated = []
for filename, count in self.access_count.items():
if count >= self.threshold:
source = self.slow_path / filename
target = self.fast_path / filename
if source.exists() and not target.exists():
try:
# 复制到高速存储
shutil.copy2(source, target)
migrated.append(filename)
print(f"迁移缓存: {filename} -> 高速存储")
except Exception as e:
print(f"迁移失败 {filename}: {e}")
# 保存元数据
self._save_metadata()
return migrated
def get_cache(self, filename):
"""获取缓存文件（优先从高速存储读取）"""
fast_file = self.fast_path / filename
if fast_file.exists():
self.record_access(filename)
return fast_file
slow_file = self.slow_path / filename
if slow_file.exists():
self.record_access(filename)
# 如果缓存满，考虑清理
self._check_cache_size()
return slow_file
return None
def _check_cache_size(self, max_size_mb=100):
"""检查高速缓存大小，必要时淘汰最不活跃文件"""
fast_size = sum(f.stat().st_size for f in self.fast_path.iterdir() if f.is_file())
max_bytes = max_size_mb * 1024 * 1024
if fast_size > max_bytes:
self._evict_cold_cache()
def _evict_cold_cache(self, keep_ratio=0.8):
"""淘汰冷缓存，保留高频访问文件"""
files_info = []
for f in self.fast_path.iterdir():
if f.is_file() and f.name != ".cache_meta.json":
files_info.append((f, self.access_count.get(f.name, 0)))
# 按访问次数排序
files_info.sort(key=lambda x: x[1])
# 淘汰低频文件
evict_count = int(len(files_info) * (1 - keep_ratio))
for f, _ in files_info[:evict_count]:
f.unlink()
print(f"淘汰冷缓存: {f.name}")
def _save_metadata(self):
"""保存缓存元数据"""
meta = {
'access_count': dict(self.access_count),
'timestamp': time.time()
}
with open(self.cache_file, 'w') as f:
json.dump(meta, f)
# 使用示例
migrator = HotCacheMigrator(
slow_path="/mnt/slow_storage/cache",
fast_path="/mnt/ramdisk/cache", # 如内存盘或SSD
threshold=5 # 访问5次后迁移
)
# 模拟缓存访问
for i in range(20):
migrator.get_cache(f"data_{i % 5}.bin")
# 执行迁移
migrator.check_and_migrate()
基于LRU的智能缓存管理
from collections import OrderedDict
import hashlib
import threading
import time
class LRUCacheManager:
"""基于LRU算法的缓存管理器"""
def __init__(self, fast_capacity_gb=10):
self.fast_capacity = fast_capacity_gb * 1024 * 1024 * 1024
self.cache = OrderedDict()
self.current_size = 0
self.lock = threading.Lock()
def access(self, key, data=None):
"""访问缓存项"""
with self.lock:
if key in self.cache:
# 移动到末尾（最近使用）
self.cache.move_to_end(key)
return self.cache[key]
return None
def add_to_fast(self, key, data, size):
"""添加到高速缓存"""
with self.lock:
# 检查是否需要淘汰
while self.current_size + size > self.fast_capacity:
self._evict_one()
# 添加到缓存
self.cache[key] = {
'data': data,
'size': size,
'timestamp': time.time()
}
self.current_size += size
def _evict_one(self):
"""淘汰一个缓存项"""
if self.cache:
key, value = self.cache.popitem(last=False)
self.current_size -= value['size']
# 可以在这里将数据移回慢速存储
self._move_to_slow(key, value)
def _move_to_slow(self, key, value):
"""将数据移回慢速存储"""
# 实现慢速存储回写逻辑
pass
# 实际应用示例
class HotDataMigrator:
def __init__(self, slow_storage, fast_storage):
self.slow = slow_storage
self.fast = fast_storage
self.access_stats = {}
def get_data(self, key):
"""获取数据（自动处理迁移）"""
# 先查找高速存储
data = self.fast.get(key)
if data:
return data
# 从慢速存储读取
data = self.slow.get(key)
if data:
# 记录热点
self._record_hot(key)
# 如果足够热，迁移到高速
if self._is_hot(key):
self.fast.put(key, data)
return data
def _record_hot(self, key):
"""记录热点数据"""
now = time.time()
if key not in self.access_stats:
self.access_stats[key] = []
# 记录最近100次访问
self.access_stats[key].append(now)
if len(self.access_stats[key]) > 100:
self.access_stats[key] = self.access_stats[key][-100:]
def _is_hot(self, key, window_seconds=60, threshold=10):
"""判断是否为热点数据"""
if key not in self.access_stats:
return False
now = time.time()
recent = [t for t in self.access_stats[key]
if now - t < window_seconds]
return len(recent) >= threshold
异步批量迁移
import asyncio
import aiofiles
from concurrent.futures import ThreadPoolExecutor
class AsyncCacheMigrator:
"""异步缓存迁移器"""
def __init__(self, slow_dir, fast_dir, max_concurrent=5):
self.slow_dir = slow_dir
self.fast_dir = fast_dir
self.semaphore = asyncio.Semaphore(max_concurrent)
async def migrate_file(self, filename):
"""异步迁移单个文件"""
async with self.semaphore:
source = os.path.join(self.slow_dir, filename)
target = os.path.join(self.fast_dir, filename)
try:
async with aiofiles.open(source, 'rb') as src:
data = await src.read()
async with aiofiles.open(target, 'wb') as dst:
await dst.write(data)
return True
except Exception as e:
print(f"迁移失败 {filename}: {e}")
return False
async def batch_migrate(self, hot_files):
"""批量迁移热点文件"""
tasks = [self.migrate_file(f) for f in hot_files]
results = await asyncio.gather(*tasks)
return sum(results), len(results)
async def monitor_and_migrate(self, interval=60):
"""持续监控并迁移"""
while True:
hot_files = await self.analyze_hot_files()
if hot_files:
success, total = await self.batch_migrate(hot_files)
print(f"迁移完成: {success}/{total}")
await asyncio.sleep(interval)
async def analyze_hot_files(self):
"""分析热点文件（需要实现具体分析逻辑）"""
# 这里需要根据实际需求实现热点分析
pass
# 使用示例
async def main():
migrator = AsyncCacheMigrator(
slow_dir="/slow/cache",
fast_dir="/fast/cache",
max_concurrent=3
)
# 批量迁移
hot_files = ["data1.bin", "data2.bin", "data3.bin"]
success, total = await migrator.batch_migrate(hot_files)
print(f"成功迁移 {success}/{total} 个文件")
# 运行
asyncio.run(main())
配置文件化方案
# config.py
CACHE_CONFIG = {
'slow_path': '/mnt/hdd/cache',
'fast_path': '/mnt/ssd/cache',
'hot_threshold': 10, # 访问次数阈值
'check_interval': 300, # 检查间隔(秒)
'max_fast_cache': 50 * 1024**3, # 50GB
'eviction_policy': 'lru', # lru/lfu/fifo
'migration_method': 'async', # sync/async
'log_level': 'INFO'
}
# 使用配置初始化
migrator = HotCacheMigrator(
slow_path=CACHE_CONFIG['slow_path'],
fast_path=CACHE_CONFIG['fast_path'],
threshold=CACHE_CONFIG['hot_threshold']
)
最佳实践建议
- 数据统计：基于实际访问模式调整迁移策略
- 预留空间：高速存储预留20%空间用于突发
- 监控告警：添加缓存命中率和迁移效率监控
- 故障恢复：实现迁移失败时的回滚机制
选择哪种方法取决于你的具体场景：
- 简单场景：使用第一种基于访问次数的方案
- 高性能场景：使用异步批量迁移
- 资源受限场景：使用LRU算法
