From ea6e654be5d4222da63436300589a023f7f9d2be Mon Sep 17 00:00:00 2001 From: fit2cloud-chenyw Date: Fri, 18 Jul 2025 09:16:00 +0800 Subject: [PATCH] perf: Fastapi cache lock --- backend/common/core/sqlbot_cache.py | 90 +++++++++++++---------------- 1 file changed, 39 insertions(+), 51 deletions(-) diff --git a/backend/common/core/sqlbot_cache.py b/backend/common/core/sqlbot_cache.py index 82c8812a..0d32917b 100644 --- a/backend/common/core/sqlbot_cache.py +++ b/backend/common/core/sqlbot_cache.py @@ -1,12 +1,27 @@ from fastapi_cache import FastAPICache from fastapi_cache.decorator import cache as original_cache from functools import partial, wraps -from typing import Optional, Set, Any, Dict, Tuple +from typing import Optional, Set, Any, Dict, Tuple, Callable from inspect import Parameter, signature import logging +from contextlib import asynccontextmanager +import asyncio logger = logging.getLogger(__name__) +# 锁管理 +_cache_locks = {} + +@asynccontextmanager +async def _get_cache_lock(key: str): + lock = _cache_locks.setdefault(key, asyncio.Lock()) + async with lock: + try: + yield + finally: + if key in _cache_locks and not lock.locked(): + del _cache_locks[key] + def should_skip_param(param: Parameter) -> bool: """判断参数是否应该被忽略(依赖注入参数)""" return ( @@ -26,9 +41,7 @@ def custom_key_builder( cacheName: Optional[str] = None, keyExpression: Optional[str] = None, ) -> str: - """ - 完全兼容FastAPICache的键生成器 - """ + """完全兼容FastAPICache的键生成器""" if cacheName: base_key = f"{namespace}:{cacheName}:" @@ -38,7 +51,6 @@ def custom_key_builder( bound_args = sig.bind_partial(*args, **kwargs) bound_args.apply_defaults() - if keyExpression.startswith("args["): import re match = re.match(r"args\[(\d+)\]", keyExpression) @@ -47,7 +59,6 @@ def custom_key_builder( value = bound_args.args[index] base_key += f"{value}:" else: - parts = keyExpression.split('.') value = bound_args.arguments[parts[0]] for part in parts[1:]: @@ -58,37 +69,27 @@ def custom_key_builder( logger.warning(f"Failed to evaluate keyExpression '{keyExpression}': {str(e)}") return base_key - # 获取函数签名 - sig = signature(func) - # 自动识别要跳过的参数 + sig = signature(func) auto_skip_args = { name for name, param in sig.parameters.items() if should_skip_param(param) } - - # 合并用户指定的额外跳过参数 skip_args = auto_skip_args.union(additional_skip_args or set()) - - # 过滤kwargs filtered_kwargs = { k: v for k, v in kwargs.items() if k not in skip_args } - # 过滤args - 将位置参数映射到它们的参数名 bound_args = sig.bind_partial(*args, **kwargs) bound_args.apply_defaults() filtered_args = [] for i, (name, value) in enumerate(bound_args.arguments.items()): - # 只处理位置参数 (在args中的参数) if i < len(args) and name not in skip_args: filtered_args.append(value) filtered_args = tuple(filtered_args) - # 获取默认键生成器 default_key_builder = FastAPICache.get_key_builder() - # 调用默认键生成器（严格按照其要求的参数格式） return default_key_builder( func=func, namespace=namespace, @@ -105,9 +106,7 @@ def cache( cacheName: Optional[str] = None, keyExpression: Optional[str] = None, ): - """ - 完全兼容的缓存装饰器 - """ + """完全兼容的缓存装饰器""" def decorator(func): if key_builder is None: used_key_builder = partial( @@ -121,24 +120,23 @@ def cache( @wraps(func) async def wrapper(*args, **kwargs): - # 准备键生成器参数 - key_builder_args = { - "func": func, - "namespace": namespace, - "args": args, - "kwargs": kwargs - } + cache_key = used_key_builder( + func=func, + namespace=namespace or "", + args=args, + kwargs=kwargs + ) - # 生成缓存键 - cache_key = used_key_builder(**key_builder_args) - logger.debug(f"Generated cache key: {cache_key}") - - # 使用原始缓存装饰器 - return await original_cache( - expire=expire, - namespace=namespace, - key_builder=lambda *_, **__: cache_key # 直接使用预生成的key - )(func)(*args, **kwargs) + async with _get_cache_lock(cache_key): + backend = FastAPICache.get_backend() + cached_value = await backend.get(cache_key) + if cached_value is not None: + return cached_value + + result = await func(*args, **kwargs) + await backend.set(cache_key, result, expire) + return result + return wrapper return decorator @@ -147,17 +145,10 @@ def clear_cache( cacheName: Optional[str] = None, keyExpression: Optional[str] = None, ): - """ - 清除缓存的装饰器，参数与 @cache 保持一致 - 使用方式： - @clear_cache(namespace="user", cacheName="info", keyExpression="user_id") - async def update_user(user_id: int): - ... - """ + """清除缓存的装饰器""" def decorator(func): @wraps(func) async def wrapper(*args, **kwargs): - # 1. 生成缓存键（复用 custom_key_builder 逻辑） cache_key = custom_key_builder( func=func, namespace=namespace or "", @@ -167,13 +158,10 @@ def clear_cache( keyExpression=keyExpression, ) - logger.debug(f"Clearing cache for key: {cache_key}") - # 2. 清除缓存 - if await FastAPICache.get_backend().get(cache_key): + async with _get_cache_lock(cache_key): await FastAPICache.clear(key=cache_key) - - # 3. 执行原函数 - return await func(*args, **kwargs) + result = await func(*args, **kwargs) + return result return wrapper return decorator \ No newline at end of file