perf: 添加多层缓存优化减少数据库查询

- 新增 ProviderCacheService 缓存 Provider 和 ProviderAPIKey 数据 - SystemConfigService 添加进程内缓存（TTL 60秒） - API Key last_used_at 更新添加节流策略（60秒间隔） - HTTP 连接池配置改为可配置，支持根据 Worker 数量自动计算 - 前端优先级管理改用 health_score 显示健康度
2026-01-12 04:28:28 +08:00 · 2026-01-08 02:34:59 +08:00
parent d9e6346911
commit d378630b38
9 changed files with 374 additions and 31 deletions
--- a/src/services/auth/service.py
+++ b/src/services/auth/service.py
@@ -8,7 +8,9 @@ import hashlib
 import secrets
 import time
 import uuid
+from collections import OrderedDict
 from datetime import datetime, timedelta, timezone
+from threading import Lock
 from typing import TYPE_CHECKING, Any, Dict, Optional

 import jwt
@@ -30,6 +32,44 @@ from src.services.cache.user_cache import UserCacheService
 from src.services.user.apikey import ApiKeyService


+# API Key last_used_at 更新节流配置
+# 同一个 API Key 在此时间间隔内只会更新一次 last_used_at
+_LAST_USED_UPDATE_INTERVAL = 60  # 秒
+_LAST_USED_CACHE_MAX_SIZE = 10000  # LRU 缓存最大条目数
+
+# 进程内缓存：记录每个 API Key 最后一次更新 last_used_at 的时间
+# 使用 OrderedDict 实现 LRU，避免内存无限增长
+_api_key_last_update_times: OrderedDict[str, float] = OrderedDict()
+_last_update_lock = Lock()
+
+
+def _should_update_last_used(api_key_id: str) -> bool:
+    """判断是否应该更新 API Key 的 last_used_at
+
+    使用节流策略，同一个 Key 在指定间隔内只更新一次。
+    线程安全，使用 LRU 策略限制缓存大小。
+
+    Returns:
+        True 表示应该更新，False 表示跳过
+    """
+    now = time.time()
+
+    with _last_update_lock:
+        last_update = _api_key_last_update_times.get(api_key_id, 0)
+
+        if now - last_update >= _LAST_USED_UPDATE_INTERVAL:
+            _api_key_last_update_times[api_key_id] = now
+            # LRU: 移到末尾（最近使用）
+            _api_key_last_update_times.move_to_end(api_key_id)
+
+            # 超过最大容量时，移除最旧的条目
+            while len(_api_key_last_update_times) > _LAST_USED_CACHE_MAX_SIZE:
+                _api_key_last_update_times.popitem(last=False)
+
+            return True
+        return False
+
+
 # JWT配置从config读取
 if not config.jwt_secret_key:
    # 如果没有配置，生成一个随机密钥并警告
@@ -367,9 +407,10 @@ class AuthService:
            logger.warning(f"API认证失败 - 用户已禁用: {user.email}")
            return None

-        # 更新最后使用时间
-        key_record.last_used_at = datetime.now(timezone.utc)
-        db.commit()  # 立即提交事务,释放数据库锁,避免阻塞后续请求
+        # 更新最后使用时间（使用节流策略，减少数据库写入）
+        if _should_update_last_used(key_record.id):
+            key_record.last_used_at = datetime.now(timezone.utc)
+            db.commit()  # 立即提交事务,释放数据库锁,避免阻塞后续请求

        api_key_fp = hashlib.sha256(api_key.encode()).hexdigest()[:12]
        logger.debug("API认证成功: 用户 {} (api_key_fp={})", user.email, api_key_fp)
--- a/src/services/cache/provider_cache.py
+++ b/src/services/cache/provider_cache.py
@@ -0,0 +1,171 @@
+"""
+Provider 缓存服务 - 减少 Provider 和 ProviderAPIKey 查询
+
+用于缓存 Provider 的 billing_type 和 ProviderAPIKey 的 rate_multiplier，
+这些数据在 UsageService.record_usage() 中被频繁查询但变化不频繁。
+"""
+
+from typing import Optional, Tuple
+
+from sqlalchemy.orm import Session
+
+from src.config.constants import CacheTTL
+from src.core.cache_service import CacheService
+from src.core.enums import ProviderBillingType
+from src.core.logger import logger
+from src.models.database import Provider, ProviderAPIKey
+
+
+class ProviderCacheService:
+    """Provider 缓存服务
+
+    提供 Provider 和 ProviderAPIKey 的缓存查询功能，减少数据库访问。
+    主要用于 UsageService 中获取费率倍数和计费类型。
+    """
+
+    CACHE_TTL = CacheTTL.PROVIDER  # 5 分钟
+
+    @staticmethod
+    async def get_provider_api_key_rate_multiplier(
+        db: Session, provider_api_key_id: str
+    ) -> Optional[float]:
+        """
+        获取 ProviderAPIKey 的 rate_multiplier（带缓存）
+
+        Args:
+            db: 数据库会话
+            provider_api_key_id: ProviderAPIKey ID
+
+        Returns:
+            rate_multiplier 或 None（如果找不到）
+        """
+        cache_key = f"provider_api_key:rate_multiplier:{provider_api_key_id}"
+
+        # 1. 尝试从缓存获取
+        cached_data = await CacheService.get(cache_key)
+        if cached_data is not None:
+            logger.debug(f"ProviderAPIKey rate_multiplier 缓存命中: {provider_api_key_id[:8]}...")
+            # 缓存的 "NOT_FOUND" 表示数据库中不存在
+            if cached_data == "NOT_FOUND":
+                return None
+            return float(cached_data)
+
+        # 2. 缓存未命中，查询数据库
+        provider_key = (
+            db.query(ProviderAPIKey.rate_multiplier)
+            .filter(ProviderAPIKey.id == provider_api_key_id)
+            .first()
+        )
+
+        # 3. 写入缓存
+        if provider_key:
+            rate_multiplier = provider_key.rate_multiplier or 1.0
+            await CacheService.set(
+                cache_key, rate_multiplier, ttl_seconds=ProviderCacheService.CACHE_TTL
+            )
+            logger.debug(f"ProviderAPIKey rate_multiplier 已缓存: {provider_api_key_id[:8]}...")
+            return rate_multiplier
+        else:
+            # 缓存负结果
+            await CacheService.set(
+                cache_key, "NOT_FOUND", ttl_seconds=ProviderCacheService.CACHE_TTL
+            )
+            return None
+
+    @staticmethod
+    async def get_provider_billing_type(
+        db: Session, provider_id: str
+    ) -> Optional[ProviderBillingType]:
+        """
+        获取 Provider 的 billing_type（带缓存）
+
+        Args:
+            db: 数据库会话
+            provider_id: Provider ID
+
+        Returns:
+            billing_type 或 None（如果找不到）
+        """
+        cache_key = f"provider:billing_type:{provider_id}"
+
+        # 1. 尝试从缓存获取
+        cached_data = await CacheService.get(cache_key)
+        if cached_data is not None:
+            logger.debug(f"Provider billing_type 缓存命中: {provider_id[:8]}...")
+            if cached_data == "NOT_FOUND":
+                return None
+            try:
+                return ProviderBillingType(cached_data)
+            except ValueError:
+                # 缓存值无效，删除并重新查询
+                await CacheService.delete(cache_key)
+
+        # 2. 缓存未命中，查询数据库
+        provider = (
+            db.query(Provider.billing_type).filter(Provider.id == provider_id).first()
+        )
+
+        # 3. 写入缓存
+        if provider:
+            billing_type = provider.billing_type
+            await CacheService.set(
+                cache_key, billing_type.value, ttl_seconds=ProviderCacheService.CACHE_TTL
+            )
+            logger.debug(f"Provider billing_type 已缓存: {provider_id[:8]}...")
+            return billing_type
+        else:
+            # 缓存负结果
+            await CacheService.set(
+                cache_key, "NOT_FOUND", ttl_seconds=ProviderCacheService.CACHE_TTL
+            )
+            return None
+
+    @staticmethod
+    async def get_rate_multiplier_and_free_tier(
+        db: Session,
+        provider_api_key_id: Optional[str],
+        provider_id: Optional[str],
+    ) -> Tuple[float, bool]:
+        """
+        获取费率倍数和是否免费套餐（带缓存）
+
+        这是 UsageService._get_rate_multiplier_and_free_tier 的缓存版本。
+
+        Args:
+            db: 数据库会话
+            provider_api_key_id: ProviderAPIKey ID（可选）
+            provider_id: Provider ID（可选）
+
+        Returns:
+            (rate_multiplier, is_free_tier) 元组
+        """
+        actual_rate_multiplier = 1.0
+        is_free_tier = False
+
+        # 获取费率倍数
+        if provider_api_key_id:
+            rate_multiplier = await ProviderCacheService.get_provider_api_key_rate_multiplier(
+                db, provider_api_key_id
+            )
+            if rate_multiplier is not None:
+                actual_rate_multiplier = rate_multiplier
+
+        # 获取计费类型
+        if provider_id:
+            billing_type = await ProviderCacheService.get_provider_billing_type(db, provider_id)
+            if billing_type == ProviderBillingType.FREE_TIER:
+                is_free_tier = True
+
+        return actual_rate_multiplier, is_free_tier
+
+    @staticmethod
+    async def invalidate_provider_api_key_cache(provider_api_key_id: str) -> None:
+        """清除 ProviderAPIKey 缓存"""
+        await CacheService.delete(f"provider_api_key:rate_multiplier:{provider_api_key_id}")
+        logger.debug(f"ProviderAPIKey 缓存已清除: {provider_api_key_id[:8]}...")
+
+    @staticmethod
+    async def invalidate_provider_cache(provider_id: str) -> None:
+        """清除 Provider 缓存"""
+        await CacheService.delete(f"provider:billing_type:{provider_id}")
+        logger.debug(f"Provider 缓存已清除: {provider_id[:8]}...")
--- a/src/services/system/config.py
+++ b/src/services/system/config.py
@@ -3,8 +3,9 @@
 """

 import json
+import time
 from enum import Enum
-from typing import Any, Dict, List, Optional
+from typing import Any, Dict, List, Optional, Tuple

 from sqlalchemy.orm import Session

@@ -20,6 +21,49 @@ class LogLevel(str, Enum):
    FULL = "full"  # 记录完整请求和响应（包含body，敏感信息会脱敏）


+# 进程内缓存 TTL（秒）- 系统配置变化不频繁，使用较长的 TTL
+_CONFIG_CACHE_TTL = 60  # 1 分钟
+
+# 进程内缓存存储: {key: (value, expire_time)}
+_config_cache: Dict[str, Tuple[Any, float]] = {}
+
+
+def _get_cached_config(key: str) -> Tuple[bool, Any]:
+    """从进程内缓存获取配置值
+
+    Returns:
+        (hit, value): hit=True 表示缓存命中，value 为缓存的值
+    """
+    if key in _config_cache:
+        value, expire_time = _config_cache[key]
+        if time.time() < expire_time:
+            return True, value
+        # 缓存过期，安全删除（避免并发时 KeyError）
+        _config_cache.pop(key, None)
+    return False, None
+
+
+def _set_cached_config(key: str, value: Any) -> None:
+    """设置进程内缓存"""
+    _config_cache[key] = (value, time.time() + _CONFIG_CACHE_TTL)
+
+
+def invalidate_config_cache(key: Optional[str] = None) -> None:
+    """清除配置缓存
+
+    Args:
+        key: 配置键，如果为 None 则清除所有缓存
+    """
+    global _config_cache
+    if key is None:
+        _config_cache = {}
+        logger.debug("已清除所有系统配置缓存")
+    else:
+        # 使用 pop 安全删除，避免并发时 KeyError
+        if _config_cache.pop(key, None) is not None:
+            logger.debug(f"已清除系统配置缓存: {key}")
+
+
 class SystemConfigService:
    """系统配置服务类"""

@@ -127,14 +171,23 @@ class SystemConfigService:

    @classmethod
    def get_config(cls, db: Session, key: str, default: Any = None) -> Optional[Any]:
-        """获取系统配置值"""
+        """获取系统配置值（带进程内缓存）"""
+        # 1. 检查进程内缓存
+        hit, cached_value = _get_cached_config(key)
+        if hit:
+            return cached_value
+
+        # 2. 查询数据库
        config = db.query(SystemConfig).filter(SystemConfig.key == key).first()
        if config:
+            _set_cached_config(key, config.value)
            return config.value

-        # 如果配置不存在，检查默认值
+        # 3. 如果配置不存在，使用默认值
        if key in cls.DEFAULT_CONFIGS:
-            return cls.DEFAULT_CONFIGS[key]["value"]
+            value = cls.DEFAULT_CONFIGS[key]["value"]
+            _set_cached_config(key, value)
+            return value

        return default

@@ -185,6 +238,9 @@ class SystemConfigService:
        db.commit()
        db.refresh(config)

+        # 清除缓存
+        invalidate_config_cache(key)
+
        return config

    @staticmethod
@@ -243,6 +299,8 @@ class SystemConfigService:
        if config:
            db.delete(config)
            db.commit()
+            # 清除缓存
+            invalidate_config_cache(key)
            return True
        return False

--- a/src/services/usage/service.py
+++ b/src/services/usage/service.py
@@ -10,7 +10,6 @@ from typing import Any, Dict, List, Optional, Tuple
 from sqlalchemy import func
 from sqlalchemy.orm import Session

-from src.core.enums import ProviderBillingType
 from src.core.logger import logger
 from src.models.database import ApiKey, Provider, ProviderAPIKey, Usage, User, UserRole
 from src.services.model.cost import ModelCostService
@@ -362,22 +361,12 @@ class UsageService:
        provider_api_key_id: Optional[str],
        provider_id: Optional[str],
    ) -> Tuple[float, bool]:
-        """获取费率倍数和是否免费套餐"""
-        actual_rate_multiplier = 1.0
-        if provider_api_key_id:
-            provider_key = (
-                db.query(ProviderAPIKey).filter(ProviderAPIKey.id == provider_api_key_id).first()
-            )
-            if provider_key and provider_key.rate_multiplier:
-                actual_rate_multiplier = provider_key.rate_multiplier
+        """获取费率倍数和是否免费套餐（使用缓存）"""
+        from src.services.cache.provider_cache import ProviderCacheService

-        is_free_tier = False
-        if provider_id:
-            provider_obj = db.query(Provider).filter(Provider.id == provider_id).first()
-            if provider_obj and provider_obj.billing_type == ProviderBillingType.FREE_TIER:
-                is_free_tier = True
-
-        return actual_rate_multiplier, is_free_tier
+        return await ProviderCacheService.get_rate_multiplier_and_free_tier(
+            db, provider_api_key_id, provider_id
+        )

    @classmethod
    async def _calculate_costs(