本文目录导读：
我来为您介绍几种Python批量迁移缓存数据的方案。
通用缓存迁移脚本
import json
import pickle
import os
from typing import Any, Dict, List
import shutil
from datetime import datetime
import logging
# 配置日志
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
class CacheMigrator:
"""缓存数据迁移器"""
def __init__(self, source_path: str, target_path: str, format_type: str = 'json'):
"""
初始化迁移器
:param source_path: 源缓存路径
:param target_path: 目标缓存路径
:param format_type: 数据格式 (json/pickle)
"""
self.source_path = source_path
self.target_path = target_path
self.format_type = format_type
self.stats = {'success': 0, 'failed': 0, 'skipped': 0}
def migrate_single_file(self, filename: str) -> bool:
"""迁移单个缓存文件"""
try:
source_file = os.path.join(self.source_path, filename)
target_file = os.path.join(self.target_path, filename)
# 检查文件是否存在
if not os.path.exists(source_file):
self.stats['skipped'] += 1
logger.warning(f"源文件不存在: {filename}")
return False
# 检查目标文件是否已存在
if os.path.exists(target_file):
# 可以选择覆盖或跳过
if not self.overwrite_existing:
self.stats['skipped'] += 1
logger.info(f"目标文件已存在，跳过: {filename}")
return False
# 读取源文件
data = self._read_cache(source_file)
# 写入目标文件
self._write_cache(target_file, data)
self.stats['success'] += 1
logger.info(f"成功迁移: {filename}")
return True
except Exception as e:
self.stats['failed'] += 1
logger.error(f"迁移失败 {filename}: {str(e)}")
return False
def _read_cache(self, filepath: str) -> Any:
"""读取缓存文件"""
with open(filepath, 'rb') as f:
if self.format_type == 'json':
return json.load(f)
elif self.format_type == 'pickle':
return pickle.load(f)
else:
raise ValueError(f"不支持的数据格式: {self.format_type}")
def _write_cache(self, filepath: str, data: Any) -> None:
"""写入缓存文件"""
os.makedirs(os.path.dirname(filepath), exist_ok=True)
with open(filepath, 'wb' if self.format_type == 'pickle' else 'w') as f:
if self.format_type == 'json':
json.dump(data, f, ensure_ascii=False, indent=2)
elif self.format_type == 'pickle':
pickle.dump(data, f)
def batch_migrate(self, file_pattern: str = '*', overwrite: bool = False) -> Dict:
"""
批量迁移缓存文件
:param file_pattern: 文件匹配模式
:param overwrite: 是否覆盖已存在的文件
"""
self.overwrite_existing = overwrite
import glob
source_files = glob.glob(os.path.join(self.source_path, file_pattern))
logger.info(f"找到 {len(source_files)} 个缓存文件待迁移")
for file_path in source_files:
filename = os.path.basename(file_path)
self.migrate_single_file(filename)
logger.info(f"迁移完成 - 成功: {self.stats['success']}, "
f"失败: {self.stats['failed']}, 跳过: {self.stats['skipped']}")
return self.stats
# 使用示例
def migrate_cache_example():
# 基本迁移
migrator = CacheMigrator(
source_path='/path/to/source/cache',
target_path='/path/to/target/cache',
format_type='json'
)
# 迁移所有文件
stats = migrator.batch_migrate(overwrite=False)
print(f"迁移统计: {stats}")
# 只迁移特定模式的文件
stats = migrator.batch_migrate(file_pattern='*.cache', overwrite=True)
Redis缓存迁移脚本
import redis
import json
from typing import Iterator, Dict, Any
class RedisCacheMigrator:
"""Redis缓存迁移器"""
def __init__(self, source_config: Dict, target_config: Dict,
batch_size: int = 1000):
"""
初始化Redis迁移器
:param source_config: 源Redis配置
:param target_config: 目标Redis配置
:param batch_size: 批量处理大小
"""
self.source_redis = redis.StrictRedis(**source_config)
self.target_redis = redis.StrictRedis(**target_config)
self.batch_size = batch_size
self.stats = {'migrated': 0, 'failed': 0, 'total': 0}
def migrate_keys_by_pattern(self, pattern: str = '*') -> Dict:
"""
根据模式迁移键
:param pattern: 键匹配模式
"""
cursor = 0
total_keys = 0
while True:
cursor, keys = self.source_redis.scan(
cursor=cursor,
match=pattern,
count=self.batch_size
)
if keys:
self._migrate_keys(keys)
total_keys += len(keys)
if cursor == 0:
break
return self.stats
def _migrate_keys(self, keys: list) -> None:
"""批量迁移键值对"""
pipeline = self.target_redis.pipeline()
for key in keys:
try:
# 获取键的类型
key_type = self.source_redis.type(key).decode()
# 获取TTL
ttl = self.source_redis.ttl(key)
# 根据类型迁移
if key_type == 'string':
value = self.source_redis.get(key)
pipeline.set(key, value)
elif key_type == 'hash':
value = self.source_redis.hgetall(key)
pipeline.hset(key, mapping=value)
elif key_type == 'list':
value = self.source_redis.lrange(key, 0, -1)
pipeline.rpush(key, *value)
elif key_type == 'set':
value = self.source_redis.smembers(key)
pipeline.sadd(key, *value)
elif key_type == 'zset':
value = self.source_redis.zrange(key, 0, -1, withscores=True)
pipeline.zadd(key, dict(value))
# 设置TTL
if ttl > 0:
pipeline.expire(key, ttl)
self.stats['migrated'] += 1
except Exception as e:
self.stats['failed'] += 1
print(f"迁移键 {key} 失败: {e}")
# 执行批量操作
pipeline.execute()
def verify_migration(self, pattern: str = '*', sample_size: int = 100) -> bool:
"""
验证迁移结果
"""
source_keys = list(self.source_redis.scan_iter(match=pattern))
target_keys = list(self.target_redis.scan_iter(match=pattern))
print(f"源Redis键数: {len(source_keys)}")
print(f"目标Redis键数: {len(target_keys)}")
# 采样验证
import random
sample_keys = random.sample(source_keys, min(sample_size, len(source_keys)))
for key in sample_keys:
source_val = self.source_redis.get(key)
target_val = self.target_redis.get(key)
if source_val != target_val:
print(f"键 {key} 值不匹配")
return False
return True
# 使用示例
def redis_migration_example():
# 配置源和目标Redis
source_config = {
'host': 'source-redis-host',
'port': 6379,
'db': 0,
'password': 'source-password'
}
target_config = {
'host': 'target-redis-host',
'port': 6379,
'db': 0,
'password': 'target-password'
}
# 创建迁移器
migrator = RedisCacheMigrator(source_config, target_config, batch_size=500)
# 执行迁移
stats = migrator.migrate_keys_by_pattern('user:*')
print(f"迁移完成: {stats}")
# 验证迁移
if migrator.verify_migration('user:*', sample_size=50):
print("迁移验证通过")
else:
print("迁移验证失败")
Memcached缓存迁移脚本
from pymemcache.client.base import Client
from pymemcache import serde
import pickle
import sys
class MemcachedMigrator:
"""Memcached缓存迁移器"""
def __init__(self, source_server: tuple, target_server: tuple):
"""
初始化Memcached迁移器
:param source_server: (host, port)
:param target_server: (host, port)
"""
self.source = Client(source_server, serde=serde.pickle_serde)
self.target = Client(target_server, serde=serde.pickle_serde)
self.stats = {'migrated': 0, 'failed': 0}
def migrate_key(self, key: str) -> bool:
"""迁移单个键"""
try:
value = self.source.get(key)
if value is not None:
# 获取过期时间
expires = self.source.get_expiration(key)
# 设置到目标
self.target.set(key, value, expire=expires)
self.stats['migrated'] += 1
return True
except Exception as e:
self.stats['failed'] += 1
print(f"迁移键 {key} 失败: {e}")
return False
def migrate_all_keys(self, keys: list) -> Dict:
"""迁移所有指定的键"""
for key in keys:
self.migrate_key(key)
return self.stats
# 使用示例
def memcached_migration_example():
migrator = MemcachedMigrator(
source_server=('localhost', 11211),
target_server=('localhost', 11212)
)
# 假设我们有一个要迁移的键列表
keys_to_migrate = ['user:1', 'user:2', 'session:abc']
stats = migrator.migrate_all_keys(keys_to_migrate)
print(f"Memcached迁移完成: {stats}")
文件系统缓存迁移（带验证）
import hashlib
from pathlib import Path
class FileCacheMigrator:
"""文件系统缓存迁移器（带完整性验证）"""
def __init__(self, source_dir: str, target_dir: str):
self.source_dir = Path(source_dir)
self.target_dir = Path(target_dir)
self.stats = {'copied': 0, 'verified': 0, 'failed': 0}
def copy_with_verification(self, relative_path: str) -> bool:
"""复制文件并验证完整性"""
source_file = self.source_dir / relative_path
target_file = self.target_dir / relative_path
if not source_file.exists():
self.stats['failed'] += 1
return False
try:
# 创建目标目录
target_file.parent.mkdir(parents=True, exist_ok=True)
# 计算源文件哈希
source_hash = self._calculate_hash(source_file)
# 复制文件
import shutil
shutil.copy2(source_file, target_file)
# 验证目标文件哈希
target_hash = self._calculate_hash(target_file)
if source_hash == target_hash:
self.stats['verified'] += 1
return True
else:
target_file.unlink() # 删除损坏的文件
self.stats['failed'] += 1
return False
except Exception as e:
self.stats['failed'] += 1
print(f"复制失败 {relative_path}: {e}")
return False
def _calculate_hash(self, filepath: Path) -> str:
"""计算文件哈希"""
sha256_hash = hashlib.sha256()
with open(filepath, 'rb') as f:
for byte_block in iter(lambda: f.read(4096), b""):
sha256_hash.update(byte_block)
return sha256_hash.hexdigest()
def batch_migrate(self, pattern: str = '**/*') -> Dict:
"""批量迁移文件"""
files = list(self.source_dir.glob(pattern))
for filepath in files:
if filepath.is_file():
relative_path = filepath.relative_to(self.source_dir)
if self.copy_with_verification(str(relative_path)):
self.stats['copied'] += 1
return self.stats
# 使用示例
def file_cache_migration_example():
migrator = FileCacheMigrator(
source_dir='/var/cache/app_v1',
target_dir='/var/cache/app_v2'
)
stats = migrator.batch_migrate('*.cache')
print(f"文件缓存迁移完成: {stats}")
增量迁移与进度显示
import time
from tqdm import tqdm
class IncrementalCacheMigrator:
"""增量缓存迁移器（带进度显示）"""
def __init__(self, migrator, checkpoint_file: str = 'migration_checkpoint.json'):
"""
初始化增量迁移器
:param migrator: 基础迁移器实例
:param checkpoint_file: 检查点文件路径
"""
self.migrator = migrator
self.checkpoint_file = checkpoint_file
self.checkpoint = self._load_checkpoint()
def _load_checkpoint(self) -> set:
"""加载检查点"""
if os.path.exists(self.checkpoint_file):
with open(self.checkpoint_file, 'r') as f:
return set(json.load(f))
return set()
def _save_checkpoint(self, migrated_keys: set):
"""保存检查点"""
self.checkpoint.update(migrated_keys)
with open(self.checkpoint_file, 'w') as f:
json.dump(list(self.checkpoint), f)
def incremental_migrate(self, keys: list, show_progress: bool = True) -> Dict:
"""增量迁移（只迁移未迁移过的键）"""
keys_to_migrate = [k for k in keys if k not in self.checkpoint]
if show_progress:
pbar = tqdm(total=len(keys_to_migrate), desc="迁移进度")
for key in keys_to_migrate:
success = self.migrator.migrate_key(key)
if success:
self.checkpoint.add(key)
if show_progress:
pbar.update(1)
pbar.set_postfix({
'成功': len(self.checkpoint),
'总': len(keys)
})
if show_progress:
pbar.close()
# 保存检查点
self._save_checkpoint(self.checkpoint)
return {
'total': len(keys),
'migrated': len(self.checkpoint),
'remaining': len(keys) - len(self.checkpoint)
}
# 使用示例
def incremental_migration_example():
# 假设我们有一个简单的迁移器
class SimpleMigrator:
def migrate_key(self, key):
# 模拟迁移操作
time.sleep(0.1)
return True
basic_migrator = SimpleMigrator()
incremental_migrator = IncrementalCacheMigrator(basic_migrator)
# 模拟要迁移的键
keys = [f"key_{i}" for i in range(100)]
# 执行增量迁移
result = incremental_migrator.incremental_migrate(keys)
print(f"增量迁移结果: {result}")
# 第二次运行只会迁移剩余的键
result2 = incremental_migrator.incremental_migrate(keys)
print(f"第二次迁移结果: {result2}")
使用建议
- 备份数据：迁移前务必备份源数据
- 分批处理：大数据量时分批迁移，避免内存溢出
- 验证完整性：迁移后验证数据完整性
- 错误处理：实现断点续传机制
- 性能优化：使用批量操作和并行处理
安装依赖
pip install redis pymemcache tqdm
选择合适的迁移方案取决于您的具体缓存系统和需求。
