本文目录导读：
的Python方案：
使用Redis批量更新
import redis
import json
class RedisCacheUpdater:
def __init__(self, host='localhost', port=6379, db=0):
self.client = redis.Redis(host=host, port=port, db=db, decode_responses=True)
def batch_update_by_keys(self, updates_dict):
"""
批量更新指定key的值
:param updates_dict: {key: new_value} 格式的字典
"""
pipeline = self.client.pipeline()
for key, value in updates_dict.items():
pipeline.set(key, json.dumps(value) if isinstance(value, (dict, list)) else value)
pipeline.execute()
print(f"成功更新 {len(updates_dict)} 个缓存项")
def batch_update_by_pattern(self, pattern, transform_func):
"""
根据模式匹配批量更新
:param pattern: key匹配模式，如 "user:*"
:param transform_func: 转换函数，接收旧值返回新值
"""
cursor = 0
updated = 0
while True:
cursor, keys = self.client.scan(cursor, match=pattern, count=100)
pipeline = self.client.pipeline()
for key in keys:
old_value = self.client.get(key)
if old_value:
try:
old_value = json.loads(old_value)
except:
pass
new_value = transform_func(key, old_value)
pipeline.set(key, json.dumps(new_value) if isinstance(new_value, (dict, list)) else new_value)
updated += 1
if pipeline:
pipeline.execute()
if cursor == 0:
break
print(f"根据模式 '{pattern}' 更新了 {updated} 个缓存项")
# 使用示例
updater = RedisCacheUpdater()
# 示例1：直接批量更新指定keys
updates = {
"user:1": {"name": "张三", "status": "active"},
"user:2": {"name": "李四", "status": "active"},
"config:app": {"version": "2.0", "maintenance": False}
}
updater.batch_update_by_keys(updates)
# 示例2：根据模式批量更新
def update_user_status(key, old_value):
if isinstance(old_value, dict):
old_value["status"] = "active"
old_value["updated_at"] = "2024-01-01"
return old_value
updater.batch_update_by_pattern("user:*", update_user_status)
使用文件缓存批量更新
import os
import json
from pathlib import Path
import shutil
class FileCacheUpdater:
def __init__(self, cache_dir):
self.cache_dir = Path(cache_dir)
self.cache_dir.mkdir(parents=True, exist_ok=True)
def batch_update_from_dict(self, updates_dict):
"""
从字典批量更新缓存文件
:param updates_dict: {cache_name: content} 格式
"""
for cache_name, content in updates_dict.items():
cache_file = self.cache_dir / f"{cache_name}.json"
with open(cache_file, 'w', encoding='utf-8') as f:
json.dump(content, f, ensure_ascii=False, indent=2)
print(f"成功更新 {len(updates_dict)} 个缓存文件")
def batch_update_by_modification(self, source_dir, pattern="*.json"):
"""
从源目录批量更新缓存
:param source_dir: 源文件目录
:param pattern: 文件匹配模式
"""
source_path = Path(source_dir)
for file_path in source_path.glob(pattern):
# 读取新的缓存内容
with open(file_path, 'r', encoding='utf-8') as f:
content = json.load(f)
# 写入缓存目录
dest_file = self.cache_dir / file_path.name
with open(dest_file, 'w', encoding='utf-8') as f:
json.dump(content, f, ensure_ascii=False, indent=2)
print(f"从 {source_dir} 更新了缓存文件")
def batch_delete(self, patterns_or_keys):
"""
批量删除缓存
:param patterns_or_keys: 文件名列表或模式
"""
for item in patterns_or_keys:
for file_path in self.cache_dir.glob(item):
file_path.unlink()
print(f"删除缓存: {file_path.name}")
# 使用示例
updater = FileCacheUpdater("./cache")
# 示例1：直接更新多个缓存
updates = {
"user_profile": {"name": "张三", "age": 30},
"app_config": {"version": "2.0", "features": ["a", "b"]},
"temp_data": [1, 2, 3, 4]
}
updater.batch_update_from_dict(updates)
# 示例2：从源目录批量更新
updater.batch_update_by_modification("./new_cache_data")
# 示例3：批量删除过期缓存
updater.batch_delete(["temp_*", "session_*"])
使用内存字典缓存批量更新
import time
import threading
from datetime import datetime
class InMemoryCacheUpdater:
def __init__(self):
self._cache = {}
self._expiry = {}
self._lock = threading.Lock()
def batch_update(self, items, ttl=None):
"""
批量更新缓存
:param items: {key: value} 格式
:param ttl: 过期时间（秒）
"""
with self._lock:
expire_time = time.time() + ttl if ttl else None
for key, value in items.items():
self._cache[key] = value
if expire_time:
self._expiry[key] = expire_time
print(f"批量更新 {len(items)} 个缓存项")
def batch_update_with_condition(self, condition_func, transform_func):
"""
根据条件批量更新
:param condition_func: 条件函数，接收(key, value)返回布尔值
:param transform_func: 转换函数，接收(key, value)返回新value
"""
with self._lock:
updated = 0
keys_to_update = []
for key, value in self._cache.items():
if condition_func(key, value):
keys_to_update.append(key)
for key in keys_to_update:
old_value = self._cache[key]
new_value = transform_func(key, old_value)
self._cache[key] = new_value
updated += 1
print(f"根据条件更新了 {updated} 个缓存项")
def batch_delete_expired(self):
"""批量删除过期缓存"""
with self._lock:
now = time.time()
expired_keys = [
key for key, expire_time in self._expiry.items()
if expire_time <= now
]
for key in expired_keys:
del self._cache[key]
del self._expiry[key]
if expired_keys:
print(f"清理了 {len(expired_keys)} 个过期缓存项")
return expired_keys
def get_all(self):
"""获取所有缓存"""
with self._lock:
return dict(self._cache)
# 使用示例
cache = InMemoryCacheUpdater()
# 示例1：批量更新
cache.batch_update({
"user:1": {"name": "张三", "role": "admin"},
"user:2": {"name": "李四", "role": "user"},
"config": {"theme": "dark", "language": "zh"}
}, ttl=3600)
# 示例2：根据条件批量更新
def is_admin(key, value):
return isinstance(value, dict) and value.get("role") == "admin"
def update_admin_role(key, value):
value["permissions"] = ["read", "write", "delete"]
return value
cache.batch_update_with_condition(is_admin, update_admin_role)
混合缓存系统批量更新
class HybridCacheUpdater:
"""支持多层级缓存批量更新"""
def __init__(self):
self.cache_layers = []
def add_cache_layer(self, cache_obj, name):
"""添加缓存层"""
self.cache_layers.append({"obj": cache_obj, "name": name})
def batch_update_all_layers(self, updates_dict):
"""
在所有缓存层批量更新
"""
for layer in self.cache_layers:
print(f"更新缓存层: {layer['name']}")
layer['obj'].batch_update(updates_dict)
def batch_update_selected_layers(self, updates_dict, layer_names):
"""
在指定缓存层批量更新
"""
for layer in self.cache_layers:
if layer['name'] in layer_names:
print(f"更新缓存层: {layer['name']}")
layer['obj'].batch_update(updates_dict)
# 使用示例
# 创建各种缓存实例
redis_cache = RedisCacheUpdater()
file_cache = FileCacheUpdater("./cache")
memory_cache = InMemoryCacheUpdater()
# 组合成混合缓存
hybrid = HybridCacheUpdater()
hybrid.add_cache_layer(redis_cache, "redis")
hybrid.add_cache_layer(file_cache, "file")
hybrid.add_cache_layer(memory_cache, "memory")
# 批量更新所有缓存层
updates = {
"common:config": {"version": "3.0", "updated": True}
}
hybrid.batch_update_all_layers(updates)
配置文件批量更新
import configparser
import yaml
class ConfigCacheUpdater:
"""批量更新配置文件缓存"""
@staticmethod
def update_ini_config(config_file, updates):
"""
更新INI配置文件
:param config_file: 配置文件路径
:param updates: {section: {option: value}} 格式
"""
config = configparser.ConfigParser()
config.read(config_file, encoding='utf-8')
for section, options in updates.items():
if not config.has_section(section):
config.add_section(section)
for option, value in options.items():
config.set(section, option, str(value))
with open(config_file, 'w', encoding='utf-8') as f:
config.write(f)
print(f"INI配置文件 {config_file} 已更新 {len(updates)} 个节")
@staticmethod
def update_yaml_config(config_file, updates):
"""
更新YAML配置文件
:param updates: 嵌套字典格式的更新内容
"""
def deep_update(original, updates):
for key, value in updates.items():
if isinstance(value, dict) and key in original:
original[key] = deep_update(original.get(key, {}), value)
else:
original[key] = value
return original
with open(config_file, 'r', encoding='utf-8') as f:
config = yaml.safe_load(f)
config = deep_update(config, updates)
with open(config_file, 'w', encoding='utf-8') as f:
yaml.dump(config, f, default_flow_style=False, allow_unicode=True)
print(f"YAML配置文件 {config_file} 已更新")
# 使用示例
updater = ConfigCacheUpdater()
# 更新INI配置
updater.update_ini_config("config.ini", {
"database": {"host": "localhost", "port": "3306"},
"cache": {"ttl": "3600", "size": "100MB"}
})
# 更新YAML配置
updater.update_yaml_config("config.yaml", {
"server": {
"host": "0.0.0.0",
"port": 8080
},
"database": {
"pool_size": 10
}
})
这些方案覆盖了常见的缓存更新场景，可以根据实际需求选择合适的方案,需要我详细解释某个方案或提供更多定制化的批量更新逻辑吗？
