xiaozhi-esp32-server/main/xiaozhi-server/core/utils/cache/manager.py

"""
全局缓存管理器
"""

import time
import threading
from typing import Any, Optional, Dict
from collections import OrderedDict
from .strategies import CacheStrategy, CacheEntry
from .config import CacheConfig, CacheType


class GlobalCacheManager:
    """全局缓存管理器"""

    def __init__(self):
        self._logger = None
        self._caches: Dict[str, Dict[str, CacheEntry]] = {}
        self._configs: Dict[str, CacheConfig] = {}
        self._locks: Dict[str, threading.RLock] = {}
        self._global_lock = threading.RLock()
        self._last_cleanup = time.time()
        self._stats = {"hits": 0, "misses": 0, "evictions": 0, "cleanups": 0}

    @property
    def logger(self):
        """延迟初始化 logger 以避免循环导入"""
        if self._logger is None:
            from config.logger import setup_logging

            self._logger = setup_logging()
        return self._logger

    def _get_cache_name(self, cache_type: CacheType, namespace: str = "") -> str:
        """生成缓存名称"""
        if namespace:
            return f"{cache_type.value}:{namespace}"
        return cache_type.value

    def _get_or_create_cache(
        self, cache_name: str, config: CacheConfig
    ) -> Dict[str, CacheEntry]:
        """获取或创建缓存空间"""
        with self._global_lock:
            if cache_name not in self._caches:
                self._caches[cache_name] = (
                    OrderedDict()
                    if config.strategy in [CacheStrategy.LRU, CacheStrategy.TTL_LRU]
                    else {}
                )
                self._configs[cache_name] = config
                self._locks[cache_name] = threading.RLock()
            return self._caches[cache_name]

    def set(
        self,
        cache_type: CacheType,
        key: str,
        value: Any,
        ttl: Optional[float] = None,
        namespace: str = "",
    ) -> None:
        """设置缓存值"""
        cache_name = self._get_cache_name(cache_type, namespace)
        config = self._configs.get(cache_name) or CacheConfig.for_type(cache_type)
        cache = self._get_or_create_cache(cache_name, config)

        # 使用配置的TTL或传入的TTL
        effective_ttl = ttl if ttl is not None else config.ttl

        with self._locks[cache_name]:
            # 创建缓存条目
            entry = CacheEntry(value=value, timestamp=time.time(), ttl=effective_ttl)

            # 处理不同策略
            if config.strategy in [CacheStrategy.LRU, CacheStrategy.TTL_LRU]:
                # LRU策略：如果已存在则移动到末尾
                if key in cache:
                    del cache[key]
                cache[key] = entry

                # 检查大小限制
                if config.max_size and len(cache) > config.max_size:
                    # 移除最旧的条目
                    oldest_key = next(iter(cache))
                    del cache[oldest_key]
                    self._stats["evictions"] += 1

            else:
                cache[key] = entry

                # 检查大小限制
                if config.max_size and len(cache) > config.max_size:
                    # 简单策略：随机移除一个条目
                    victim_key = next(iter(cache))
                    del cache[victim_key]
                    self._stats["evictions"] += 1

        # 定期清理过期条目
        self._maybe_cleanup(cache_name)

    def get(
        self, cache_type: CacheType, key: str, namespace: str = ""
    ) -> Optional[Any]:
        """获取缓存值"""
        cache_name = self._get_cache_name(cache_type, namespace)

        if cache_name not in self._caches:
            self._stats["misses"] += 1
            return None

        cache = self._caches[cache_name]
        config = self._configs[cache_name]

        with self._locks[cache_name]:
            if key not in cache:
                self._stats["misses"] += 1
                return None

            entry = cache[key]

            # 检查过期
            if entry.is_expired():
                del cache[key]
                self._stats["misses"] += 1
                return None

            # 更新访问信息
            entry.touch()

            # LRU策略：移动到末尾
            if config.strategy in [CacheStrategy.LRU, CacheStrategy.TTL_LRU]:
                del cache[key]
                cache[key] = entry

            self._stats["hits"] += 1
            return entry.value

    def delete(self, cache_type: CacheType, key: str, namespace: str = "") -> bool:
        """删除缓存条目"""
        cache_name = self._get_cache_name(cache_type, namespace)

        if cache_name not in self._caches:
            return False

        cache = self._caches[cache_name]

        with self._locks[cache_name]:
            if key in cache:
                del cache[key]
                return True
            return False

    def clear(self, cache_type: CacheType, namespace: str = "") -> None:
        """清空指定缓存"""
        cache_name = self._get_cache_name(cache_type, namespace)

        if cache_name not in self._caches:
            return

        with self._locks[cache_name]:
            self._caches[cache_name].clear()

    def invalidate_pattern(
        self, cache_type: CacheType, pattern: str, namespace: str = ""
    ) -> int:
        """按模式失效缓存条目"""
        cache_name = self._get_cache_name(cache_type, namespace)

        if cache_name not in self._caches:
            return 0

        cache = self._caches[cache_name]
        deleted_count = 0

        with self._locks[cache_name]:
            keys_to_delete = [key for key in cache.keys() if pattern in key]
            for key in keys_to_delete:
                del cache[key]
                deleted_count += 1

        return deleted_count

    def _cleanup_expired(self, cache_name: str) -> int:
        """清理过期条目"""
        if cache_name not in self._caches:
            return 0

        cache = self._caches[cache_name]
        deleted_count = 0

        with self._locks[cache_name]:
            expired_keys = [key for key, entry in cache.items() if entry.is_expired()]
            for key in expired_keys:
                del cache[key]
                deleted_count += 1

        return deleted_count

    def _maybe_cleanup(self, cache_name: str):
        """定期清理检查"""
        config = self._configs.get(cache_name)
        if not config:
            return

        now = time.time()
        if now - self._last_cleanup > config.cleanup_interval:
            self._last_cleanup = now
            deleted = self._cleanup_expired(cache_name)
            if deleted > 0:
                self._stats["cleanups"] += 1
                self.logger.debug(f"清理缓存 {cache_name}: 删除 {deleted} 个过期条目")


# 创建全局缓存管理器实例
cache_manager = GlobalCacheManager()