"""多级缓存管理器 实现高性能的多级缓存系统： - L1缓存：内存缓存，1000项，60秒TTL，用于热点数据 - L2缓存：扩展缓存，10000项，300秒TTL，用于温数据 - LRU淘汰策略：自动淘汰最少使用的数据 - 智能预热：启动时预加载高频数据 - 统计信息：命中率、淘汰率等监控数据 支持多种缓存后端： - memory: 内存多级缓存（默认） - redis: Redis 分布式缓存 """ import asyncio import builtins import time from collections import OrderedDict from collections.abc import Callable from dataclasses import dataclass from typing import Any, Generic, TypeVar from src.common.database.optimization.cache_backend import CacheBackend from src.common.logger import get_logger from src.common.memory_utils import estimate_cache_item_size logger = get_logger("cache_manager") T = TypeVar("T") @dataclass class CacheEntry(Generic[T]): """缓存条目 Attributes: value: 缓存的值 created_at: 创建时间戳 last_accessed: 最后访问时间戳 access_count: 访问次数 size: 数据大小（字节） """ value: T created_at: float last_accessed: float access_count: int = 0 size: int = 0 @dataclass class CacheStats: """缓存统计信息 Attributes: hits: 命中次数 misses: 未命中次数 evictions: 淘汰次数 total_size: 总大小（字节） item_count: 条目数量 """ hits: int = 0 misses: int = 0 evictions: int = 0 total_size: int = 0 item_count: int = 0 @property def hit_rate(self) -> float: """命中率""" total = self.hits + self.misses return self.hits / total if total > 0 else 0.0 @property def eviction_rate(self) -> float: """淘汰率""" return self.evictions / self.item_count if self.item_count > 0 else 0.0 class LRUCache(Generic[T]): """LRU缓存实现 使用OrderedDict实现O(1)的get/set操作 """ def __init__( self, max_size: int, ttl: float, name: str = "cache", ): """初始化LRU缓存 Args: max_size: 最大缓存条目数 ttl: 过期时间（秒） name: 缓存名称，用于日志 """ self.max_size = max_size self.ttl = ttl self.name = name self._cache: OrderedDict[str, CacheEntry[T]] = OrderedDict() self._lock = asyncio.Lock() self._stats = CacheStats() async def get(self, key: str) -> T | None: """获取缓存值 Args: key: 缓存键 Returns: 缓存值，如果不存在或已过期返回None """ async with self._lock: entry = self._cache.get(key) if entry is None: self._stats.misses += 1 return None # 检查是否过期 now = time.time() if now - entry.created_at > self.ttl: # 过期，删除条目 del self._cache[key] self._stats.misses += 1 self._stats.evictions += 1 self._stats.item_count -= 1 self._stats.total_size -= entry.size return None # 命中，更新访问信息 entry.last_accessed = now entry.access_count += 1 self._stats.hits += 1 # 移到末尾（最近使用） self._cache.move_to_end(key) return entry.value async def set( self, key: str, value: T, size: int | None = None, ttl: float | None = None, ) -> None: """设置缓存值 Args: key: 缓存键 value: 缓存值 size: 数据大小（字节），如果为None则尝试估算 ttl: 自定义过期时间（秒），如果为None则使用默认TTL """ async with self._lock: now = time.time() # 如果键已存在，更新值 if key in self._cache: old_entry = self._cache[key] self._stats.total_size -= old_entry.size # 估算大小 if size is None: size = self._estimate_size(value) # 创建新条目（如果指定了ttl，则修改created_at来实现自定义TTL） # 通过调整created_at，使得: now - created_at + custom_ttl = self.ttl # 即: created_at = now - (self.ttl - custom_ttl) if ttl is not None and ttl != self.ttl: # 调整创建时间以实现自定义TTL adjusted_created_at = now - (self.ttl - ttl) logger.debug( f"[{self.name}] 使用自定义TTL {ttl}s (默认{self.ttl}s) for key: {key}" ) else: adjusted_created_at = now entry = CacheEntry( value=value, created_at=adjusted_created_at, last_accessed=now, access_count=0, size=size, ) # 如果缓存已满，淘汰最久未使用的条目 while len(self._cache) >= self.max_size: oldest_key, oldest_entry = self._cache.popitem(last=False) self._stats.evictions += 1 self._stats.item_count -= 1 self._stats.total_size -= oldest_entry.size logger.debug( f"[{self.name}] 淘汰缓存条目: {oldest_key} " f"(访问{oldest_entry.access_count}次)" ) # 添加新条目 self._cache[key] = entry self._stats.item_count += 1 self._stats.total_size += size async def delete(self, key: str) -> bool: """删除缓存条目 Args: key: 缓存键 Returns: 是否成功删除 """ async with self._lock: entry = self._cache.pop(key, None) if entry: self._stats.item_count -= 1 self._stats.total_size -= entry.size return True return False async def clear(self) -> None: """清空缓存""" async with self._lock: self._cache.clear() self._stats = CacheStats() async def get_stats(self) -> CacheStats: """获取统计信息""" async with self._lock: return CacheStats( hits=self._stats.hits, misses=self._stats.misses, evictions=self._stats.evictions, total_size=self._stats.total_size, item_count=self._stats.item_count, ) def _estimate_size(self, value: Any) -> int: """估算数据大小（字节）- 使用准确的估算方法 使用深度递归估算，比 sys.getsizeof() 更准确 """ try: return estimate_cache_item_size(value) except (TypeError, AttributeError): # 无法获取大小，返回默认值 return 1024 class MultiLevelCache(CacheBackend): """多级缓存管理器 实现两级缓存架构： - L1: 高速缓存，小容量，短TTL - L2: 扩展缓存，大容量，长TTL 查询时先查L1，未命中再查L2，未命中再从数据源加载 实现 CacheBackend 接口，可与 Redis 缓存互换使用 """ def __init__( self, l1_max_size: int = 1000, l1_ttl: float = 60, l2_max_size: int = 10000, l2_ttl: float = 300, max_memory_mb: int = 100, max_item_size_mb: int = 1, ): """初始化多级缓存 Args: l1_max_size: L1缓存最大条目数 l1_ttl: L1缓存TTL（秒） l2_max_size: L2缓存最大条目数 l2_ttl: L2缓存TTL（秒） max_memory_mb: 最大内存占用（MB） max_item_size_mb: 单个缓存条目最大大小（MB） """ self.l1_cache: LRUCache[Any] = LRUCache(l1_max_size, l1_ttl, "L1") self.l2_cache: LRUCache[Any] = LRUCache(l2_max_size, l2_ttl, "L2") self.max_memory_bytes = max_memory_mb * 1024 * 1024 self.max_item_size_bytes = max_item_size_mb * 1024 * 1024 self._cleanup_task: asyncio.Task | None = None self._is_closing = False # 🔧 添加关闭标志 logger.info( f"多级缓存初始化: L1({l1_max_size}项/{l1_ttl}s) " f"L2({l2_max_size}项/{l2_ttl}s) 内存上限({max_memory_mb}MB) " f"单项上限({max_item_size_mb}MB)" ) async def get( self, key: str, loader: Callable[[], Any] | None = None, ) -> Any | None: """从缓存获取数据 查询顺序：L1 -> L2 -> loader Args: key: 缓存键 loader: 数据加载函数，当缓存未命中时调用 Returns: 缓存值或加载的值，如果都不存在返回None """ # 1. 尝试从L1获取 value = await self.l1_cache.get(key) if value is not None: return value # 2. 尝试从L2获取 value = await self.l2_cache.get(key) if value is not None: # 提升到L1 await self.l1_cache.set(key, value) return value # 3. 使用loader加载 if loader is not None: logger.debug(f"缓存未命中，从数据源加载: {key}") value = await loader() if asyncio.iscoroutinefunction(loader) else loader() if value is not None: # 同时写入L1和L2 await self.set(key, value) return value return None async def set( self, key: str, value: Any, ttl: float | None = None, size: int | None = None, ) -> None: """设置缓存值 同时写入L1和L2 Args: key: 缓存键 value: 缓存值 ttl: 自定义过期时间（秒），如果为None则使用默认TTL size: 数据大小（字节） """ # 估算数据大小（如果未提供） if size is None: size = estimate_cache_item_size(value) # 检查单个条目大小是否超过限制 if size > self.max_item_size_bytes: logger.warning( f"缓存条目过大，跳过缓存: key={key}, " f"size={size / (1024 * 1024):.2f}MB, " f"limit={self.max_item_size_bytes / (1024 * 1024):.2f}MB" ) return # 根据TTL决定写入哪个缓存层 if ttl is not None: # 有自定义TTL，根据TTL大小决定写入层级 if ttl <= self.l1_cache.ttl: # 短TTL，只写入L1 await self.l1_cache.set(key, value, size, ttl) elif ttl <= self.l2_cache.ttl: # 中等TTL，写入L1和L2 await self.l1_cache.set(key, value, size, ttl) await self.l2_cache.set(key, value, size, ttl) else: # 长TTL，只写入L2 await self.l2_cache.set(key, value, size, ttl) else: # 没有自定义TTL，使用默认行为（同时写入L1和L2） await self.l1_cache.set(key, value, size) await self.l2_cache.set(key, value, size) async def delete(self, key: str) -> bool: """删除缓存条目 同时从L1和L2删除 Args: key: 缓存键 Returns: 是否有条目被删除 """ l1_deleted = await self.l1_cache.delete(key) l2_deleted = await self.l2_cache.delete(key) return l1_deleted or l2_deleted async def exists(self, key: str) -> bool: """检查键是否存在于缓存中 Args: key: 缓存键 Returns: 键是否存在 """ # 检查 L1 if await self.l1_cache.get(key) is not None: return True # 检查 L2 if await self.l2_cache.get(key) is not None: return True return False async def close(self) -> None: """关闭缓存（停止清理任务并清空）""" await self.stop_cleanup_task() await self.clear() logger.info("多级缓存已关闭") @property def backend_type(self) -> str: """返回缓存后端类型标识""" return "memory" @property def is_distributed(self) -> bool: """内存缓存不是分布式的""" return False async def clear(self) -> None: """清空所有缓存""" await self.l1_cache.clear() await self.l2_cache.clear() logger.info("所有缓存已清空") async def get_stats(self) -> dict[str, Any]: """获取所有缓存层的统计信息（修复版：避免锁嵌套，使用超时）""" # 🔧 修复：并行获取统计信息，避免锁嵌套 l1_stats_task = asyncio.create_task(self._get_cache_stats_safe(self.l1_cache, "L1")) l2_stats_task = asyncio.create_task(self._get_cache_stats_safe(self.l2_cache, "L2")) # 使用超时避免死锁 results = await asyncio.gather( asyncio.wait_for(l1_stats_task, timeout=1.0), asyncio.wait_for(l2_stats_task, timeout=1.0), return_exceptions=True ) l1_stats = results[0] l2_stats = results[1] # 处理异常情况 if isinstance(l1_stats, BaseException): logger.error(f"L1统计获取失败: {l1_stats}") l1_stats = CacheStats() if isinstance(l2_stats, BaseException): logger.error(f"L2统计获取失败: {l2_stats}") l2_stats = CacheStats() assert isinstance(l1_stats, CacheStats) assert isinstance(l2_stats, CacheStats) # 🔧 修复：并行获取键集合，避免锁嵌套 l1_keys_task = asyncio.create_task(self._get_cache_keys_safe(self.l1_cache)) l2_keys_task = asyncio.create_task(self._get_cache_keys_safe(self.l2_cache)) results = await asyncio.gather( asyncio.wait_for(l1_keys_task, timeout=1.0), asyncio.wait_for(l2_keys_task, timeout=1.0), return_exceptions=True ) l1_keys = results[0] l2_keys = results[1] # 处理异常情况 if isinstance(l1_keys, BaseException): logger.warning(f"L1键获取失败: {l1_keys}") l1_keys = set() if isinstance(l2_keys, BaseException): logger.warning(f"L2键获取失败: {l2_keys}") l2_keys = set() assert isinstance(l1_keys, set) assert isinstance(l2_keys, set) # 计算共享键和独占键 shared_keys = l1_keys & l2_keys l1_only_keys = l1_keys - l2_keys # noqa: F841 l2_only_keys = l2_keys - l1_keys # noqa: F841 # 🔧 修复：并行计算内存使用，避免锁嵌套 l1_size_task = asyncio.create_task(self._calculate_memory_usage_safe(self.l1_cache, l1_keys)) l2_size_task = asyncio.create_task(self._calculate_memory_usage_safe(self.l2_cache, l2_keys)) results = await asyncio.gather( asyncio.wait_for(l1_size_task, timeout=1.0), asyncio.wait_for(l2_size_task, timeout=1.0), return_exceptions=True ) l1_size = results[0] l2_size = results[1] # 处理异常情况 if isinstance(l1_size, BaseException): logger.warning(f"L1内存计算失败: {l1_size}") l1_size = l1_stats.total_size if isinstance(l2_size, BaseException): logger.warning(f"L2内存计算失败: {l2_size}") l2_size = l2_stats.total_size assert isinstance(l1_size, int) assert isinstance(l2_size, int) # 计算实际总内存（避免重复计数） actual_total_size = l1_size + l2_size - min(l1_stats.total_size, l2_stats.total_size) return { "l1": l1_stats, "l2": l2_stats, "total_memory_mb": actual_total_size / (1024 * 1024), "l1_only_mb": l1_size / (1024 * 1024), "l2_only_mb": l2_size / (1024 * 1024), "shared_mb": min(l1_stats.total_size, l2_stats.total_size) / (1024 * 1024), "shared_keys_count": len(shared_keys), "dedup_savings_mb": (l1_stats.total_size + l2_stats.total_size - actual_total_size) / (1024 * 1024), "max_memory_mb": self.max_memory_bytes / (1024 * 1024), "memory_usage_percent": (actual_total_size / self.max_memory_bytes * 100) if self.max_memory_bytes > 0 else 0, } async def _get_cache_stats_safe(self, cache, cache_name: str) -> CacheStats: """安全获取缓存统计信息（带超时）""" try: return await asyncio.wait_for(cache.get_stats(), timeout=0.5) except asyncio.TimeoutError: logger.warning(f"{cache_name}统计获取超时") return CacheStats() except Exception as e: logger.error(f"{cache_name}统计获取异常: {e}") return CacheStats() async def _get_cache_keys_safe(self, cache) -> builtins.set[str]: """安全获取缓存键集合（带超时）""" try: # 快速获取键集合，使用超时避免死锁 return await asyncio.wait_for( self._extract_keys_with_lock(cache), timeout=0.5 ) except asyncio.TimeoutError: logger.warning(f"缓存键获取超时: {cache.name}") return set() except Exception as e: logger.error(f"缓存键获取异常: {e}") return set() async def _extract_keys_with_lock(self, cache) -> builtins.set[str]: """在锁保护下提取键集合""" async with cache._lock: return set(cache._cache.keys()) async def _calculate_memory_usage_safe(self, cache, keys: builtins.set[str]) -> int: """安全计算内存使用（带超时）""" if not keys: return 0 try: return await asyncio.wait_for( self._calc_memory_with_lock(cache, keys), timeout=0.5 ) except asyncio.TimeoutError: logger.warning(f"内存计算超时: {cache.name}") return 0 except Exception as e: logger.error(f"内存计算异常: {e}") return 0 async def _calc_memory_with_lock(self, cache, keys: builtins.set[str]) -> int: """在锁保护下计算内存使用""" total_size = 0 async with cache._lock: for key in keys: entry = cache._cache.get(key) if entry: total_size += entry.size return total_size async def check_memory_limit(self) -> None: """检查并强制清理超出内存限制的缓存（修复版：避免嵌套锁）""" try: # 🔧 修复：使用超时获取统计，避免死锁 stats = await asyncio.wait_for(self.get_stats(), timeout=2.0) total_size = stats["total_memory_mb"] * (1024 * 1024) # 转换回字节 if total_size > self.max_memory_bytes: memory_mb = total_size / (1024 * 1024) max_mb = self.max_memory_bytes / (1024 * 1024) logger.warning( f"缓存内存超限: {memory_mb:.2f}MB / {max_mb:.2f}MB " f"({stats['memory_usage_percent']:.1f}%)，开始分阶段清理" ) # 🔧 修复：分阶段清理，每阶段都有超时保护 cleanup_success = False # 阶段1: 清理过期条目 try: await asyncio.wait_for(self._clean_expired_entries(), timeout=3.0) # 重新检查内存使用 stats_after_clean = await asyncio.wait_for(self.get_stats(), timeout=1.0) total_after_clean = stats_after_clean["total_memory_mb"] * (1024 * 1024) if total_after_clean <= self.max_memory_bytes: logger.info("清理过期条目后内存使用正常") cleanup_success = True except asyncio.TimeoutError: logger.warning("清理过期条目超时，跳到强制清理") # 阶段2: 如果过期清理不够，清理L2缓存 if not cleanup_success: try: logger.info("开始清理L2缓存") await asyncio.wait_for(self.l2_cache.clear(), timeout=2.0) logger.info("L2缓存清理完成") # 检查L1缓存是否还需要清理 stats_after_l2 = await asyncio.wait_for(self.get_stats(), timeout=1.0) total_after_l2 = stats_after_l2["total_memory_mb"] * (1024 * 1024) if total_after_l2 > self.max_memory_bytes: logger.warning("清理L2后仍超限，继续清理L1缓存") await asyncio.wait_for(self.l1_cache.clear(), timeout=2.0) logger.info("L1缓存清理完成") except asyncio.TimeoutError: logger.error("强制清理超时，内存可能仍有问题") except Exception as e: logger.error(f"强制清理失败: {e}") logger.info("缓存内存限制检查完成") except asyncio.TimeoutError: logger.warning("内存限制检查超时，跳过本次检查") except Exception as e: logger.error(f"内存限制检查失败: {e}") async def start_cleanup_task(self, interval: float = 60) -> None: """启动定期清理任务 Args: interval: 清理间隔（秒） """ if self._cleanup_task is not None: logger.warning("清理任务已在运行") return async def cleanup_loop(): while not self._is_closing: try: await asyncio.sleep(interval) if self._is_closing: break stats = await self.get_stats() l1_stats = stats["l1"] l2_stats = stats["l2"] logger.info( f"缓存统计 - L1: {l1_stats.item_count}项, " f"命中率{l1_stats.hit_rate:.2%} | " f"L2: {l2_stats.item_count}项, " f"命中率{l2_stats.hit_rate:.2%} | " f"内存: {stats['total_memory_mb']:.2f}MB/{stats['max_memory_mb']:.2f}MB " f"({stats['memory_usage_percent']:.1f}%) | " f"共享: {stats['shared_keys_count']}键/{stats['shared_mb']:.2f}MB " f"(去重节省{stats['dedup_savings_mb']:.2f}MB)" ) # 🔧 清理过期条目 await self._clean_expired_entries() # 检查内存限制 await self.check_memory_limit() except asyncio.CancelledError: break except Exception as e: logger.error(f"清理任务异常: {e}") self._cleanup_task = asyncio.create_task(cleanup_loop()) logger.info(f"缓存清理任务已启动，间隔{interval}秒") async def stop_cleanup_task(self) -> None: """停止清理任务""" self._is_closing = True if self._cleanup_task is not None: self._cleanup_task.cancel() try: await self._cleanup_task except asyncio.CancelledError: pass self._cleanup_task = None logger.info("缓存清理任务已停止") async def _clean_expired_entries(self) -> None: """清理过期的缓存条目（修复版：并行清理，避免锁嵌套）""" try: current_time = time.time() # 🔧 修复：并行清理 L1 和 L2，使用超时避免死锁 async def clean_l1_expired(): """清理L1过期条目""" try: # 使用超时避免长时间持锁 await asyncio.wait_for( self._clean_cache_layer_expired(self.l1_cache, current_time, "L1"), timeout=2.0 ) except asyncio.TimeoutError: logger.warning("L1缓存清理超时，跳过本次清理") except Exception as e: logger.error(f"L1缓存清理异常: {e}") async def clean_l2_expired(): """清理L2过期条目""" try: # 使用超时避免长时间持锁 await asyncio.wait_for( self._clean_cache_layer_expired(self.l2_cache, current_time, "L2"), timeout=2.0 ) except asyncio.TimeoutError: logger.warning("L2缓存清理超时，跳过本次清理") except Exception as e: logger.error(f"L2缓存清理异常: {e}") # 🔧 关键修复：并行执行清理，避免串行等待 l1_task = asyncio.create_task(clean_l1_expired()) l2_task = asyncio.create_task(clean_l2_expired()) # 等待两个清理任务完成（使用return_exceptions避免一个失败影响另一个） results = await asyncio.gather(l1_task, l2_task, return_exceptions=True) # 检查清理结果 for i, result in enumerate(results): if isinstance(result, Exception): logger.error(f"缓存清理任务 {'L1' if i == 0 else 'L2'} 失败: {result}") else: logger.debug(f"缓存清理任务 {'L1' if i == 0 else 'L2'} 完成") except Exception as e: logger.error(f"清理过期条目失败: {e}") async def _clean_cache_layer_expired(self, cache_layer, current_time: float, layer_name: str) -> int: """清理单个缓存层的过期条目（避免锁嵌套）""" expired_keys = [] cleaned_count = 0 try: # 快速扫描过期键（短暂持锁） async with cache_layer._lock: expired_keys = [ key for key, entry in cache_layer._cache.items() if current_time - entry.created_at > cache_layer.ttl ] # 分批删除过期键，避免长时间持锁 batch_size = 50 # 每批处理50个键 for i in range(0, len(expired_keys), batch_size): batch = expired_keys[i:i + batch_size] async with cache_layer._lock: for key in batch: entry = cache_layer._cache.pop(key, None) if entry: cache_layer._stats.evictions += 1 cache_layer._stats.item_count -= 1 cache_layer._stats.total_size -= entry.size cleaned_count += 1 # 在批次之间短暂让出控制权，避免长时间阻塞 if i + batch_size < len(expired_keys): await asyncio.sleep(0.001) # 1ms if cleaned_count > 0: logger.debug(f"{layer_name}缓存清理完成: {cleaned_count} 个过期条目") except Exception as e: logger.error(f"{layer_name}缓存层清理失败: {e}") raise return cleaned_count # 全局缓存实例（支持多种后端类型） _global_cache: CacheBackend | None = None _cache_lock = asyncio.Lock() _cache_backend_type: str = "memory" # 记录当前使用的后端类型 async def get_cache() -> CacheBackend: """获取全局缓存实例（单例） 根据配置自动选择缓存后端： - cache_backend = "memory": 使用内存多级缓存（默认） - cache_backend = "redis": 使用 Redis 分布式缓存 如果配置中禁用了缓存，返回一个最小化的缓存实例 """ global _global_cache, _cache_backend_type if _global_cache is None: async with _cache_lock: if _global_cache is None: # 尝试从配置读取参数 try: from src.config.config import global_config assert global_config is not None db_config = global_config.database # 检查是否启用缓存 if not db_config.enable_database_cache: logger.info("数据库缓存已禁用，使用最小化内存缓存实例") _global_cache = MultiLevelCache( l1_max_size=1, l1_ttl=1, l2_max_size=1, l2_ttl=1, max_memory_mb=1, ) _cache_backend_type = "memory" return _global_cache # 根据配置选择缓存后端 backend = db_config.cache_backend.lower() _cache_backend_type = backend if backend == "redis": # 使用 Redis 缓存 _global_cache = await _create_redis_cache(db_config) else: # 默认使用内存缓存 _global_cache = await _create_memory_cache(db_config) except Exception as e: # 配置未加载，使用默认内存缓存 logger.warning(f"无法从配置加载缓存参数，使用默认内存缓存: {e}") _global_cache = MultiLevelCache() _cache_backend_type = "memory" await _global_cache.start_cleanup_task(interval=60) return _global_cache async def _create_memory_cache(db_config: Any) -> MultiLevelCache: """创建内存多级缓存""" l1_max_size = db_config.cache_l1_max_size l1_ttl = db_config.cache_l1_ttl l2_max_size = db_config.cache_l2_max_size l2_ttl = db_config.cache_l2_ttl max_memory_mb = db_config.cache_max_memory_mb max_item_size_mb = db_config.cache_max_item_size_mb cleanup_interval = db_config.cache_cleanup_interval logger.info( f"创建内存缓存: L1({l1_max_size}/{l1_ttl}s), " f"L2({l2_max_size}/{l2_ttl}s), 内存限制({max_memory_mb}MB)" ) cache = MultiLevelCache( l1_max_size=l1_max_size, l1_ttl=l1_ttl, l2_max_size=l2_max_size, l2_ttl=l2_ttl, max_memory_mb=max_memory_mb, max_item_size_mb=max_item_size_mb, ) await cache.start_cleanup_task(interval=cleanup_interval) return cache async def _create_redis_cache(db_config: Any) -> CacheBackend: """创建 Redis 缓存 Raises: RuntimeError: Redis 连接失败时抛出异常 """ from src.common.database.optimization.redis_cache import RedisCache logger.info( f"创建 Redis 缓存: {db_config.redis_host}:{db_config.redis_port}/{db_config.redis_db}, " f"前缀={db_config.redis_key_prefix}, TTL={db_config.redis_default_ttl}s" ) cache = RedisCache( host=db_config.redis_host, port=db_config.redis_port, password=db_config.redis_password or None, db=db_config.redis_db, key_prefix=db_config.redis_key_prefix, default_ttl=db_config.redis_default_ttl, pool_size=db_config.redis_connection_pool_size, socket_timeout=db_config.redis_socket_timeout, ssl=db_config.redis_ssl, ) # 测试连接 if await cache.health_check(): logger.info("Redis 缓存连接成功") return cache else: await cache.close() raise RuntimeError( f"Redis 连接测试失败: {db_config.redis_host}:{db_config.redis_port}，" "请检查 Redis 服务是否运行，或将 cache_backend 改为 'memory'" ) def get_cache_backend_type() -> str: """获取当前使用的缓存后端类型 Returns: "memory" 或 "redis" """ return _cache_backend_type async def close_cache() -> None: """关闭全局缓存""" global _global_cache, _cache_backend_type if _global_cache is not None: await _global_cache.close() logger.info(f"全局缓存已关闭 (后端: {_cache_backend_type})") _global_cache = None _cache_backend_type = "memory"