Python脚本如何迁移热点缓存至高速存储

wen python案例 2026-06-26 245

本文目录导读：

基于访问频率的缓存迁移
基于LRU的智能缓存管理
异步批量迁移
配置文件化方案
最佳实践建议

我来介绍几种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']
)