"""Redis 读缓存(可选基础设施)。 - REDIS_URL 留空、连接失败或超时:所有操作静默退化为「无缓存」,接口照常直查数据库, 且本进程内禁用重试(避免每个请求都陪跑一次连接超时)。 - 失效策略:TTL 自然过期 + 版本号(INCR)作废。自选股增删等写操作只 INCR 版本 key, 旧缓存 key 里带着旧版本号,无需 SCAN 批量删除。 - 只缓存「读多写少、可容忍短暂陈旧」的聚合数据(股票列表、筛选项等); K线/回测等口径敏感数据不走这里。 - Redis 之前还有一层进程内本地缓存(local_get/local_set,0 RTT):只存已序列化好的 JSON 字符串,命中后接口直接 Response(content=raw) 原样返回,跳过 json.loads + pydantic 校验/序列化(大响应这两步合计可达数百 ms)。本地 TTL 恒 ≤ Redis TTL, 多进程部署时本地条目最多比 Redis 多陈旧 120s;版本号 bump 在同进程立即生效。 """ from __future__ import annotations import hashlib import json import time from typing import Any import redis.asyncio as aioredis from .config import settings _pool: aioredis.ConnectionPool | None = None _disabled = False # 一次失败后本进程禁用(Redis 属加速件,坏了不能拖慢接口) def _client() -> aioredis.Redis | None: global _pool, _disabled if not settings.redis_url or _disabled: return None if _pool is None: _pool = aioredis.ConnectionPool.from_url( settings.redis_url, decode_responses=True, socket_connect_timeout=1.0, socket_timeout=1.0, health_check_interval=60, max_connections=32, ) return aioredis.Redis(connection_pool=_pool) def _bail() -> None: global _disabled _disabled = True def digest(*parts: Any) -> str: """参数指纹(拼接后 md5,仅用于拼缓存 key,非安全用途)""" raw = "\x1f".join(repr(p) for p in parts) return hashlib.md5(raw.encode()).hexdigest() # noqa: S324 # --- 进程内本地缓存层(Redis 之前,0 RTT)----------------------------------- # asyncio 单线程,dict 读写无锁安全;按插入序淘汰近似 LRU(命中不续期,足够用)。 _LOCAL_MAX_BYTES = 64 * 1024 * 1024 # 总字节预算(preview 响应 ~250KB,留足 1y 大窗口) _LOCAL_MAX_ENTRIES = 128 _local_store: dict[str, tuple[float, str]] = {} # key -> (expires_monotonic, raw_json) _local_bytes = 0 def local_get(key: str) -> str | None: """命中返回已序列化的 JSON 字符串(调用方直接 Response 原样返回)。""" global _local_bytes ent = _local_store.get(key) if ent is None: return None expires, raw = ent if expires < time.monotonic(): _local_store.pop(key, None) _local_bytes -= len(raw) + 64 return None return raw def local_set(key: str, raw: str, ttl: int) -> None: """ttl 为秒;调用方应传 min(本地默认, Redis TTL),保证本地不比 Redis 活得久。""" global _local_bytes old = _local_store.pop(key, None) if old is not None: _local_bytes -= len(old[1]) _local_store[key] = (time.monotonic() + max(1, min(ttl, 120)), raw) _local_bytes += len(raw) + 64 # 连同 dict/tuple 开销粗略计入 while _local_store and (len(_local_store) > _LOCAL_MAX_ENTRIES or _local_bytes > _LOCAL_MAX_BYTES): _local_bytes -= len(_local_store.popitem(last=False)[1][1]) + 64 # --- 版本号本地缓存:热请求连 Redis GET ver:xx 都省掉 ------------------------ # 同进程 bump_version 立即刷新本地;其他进程 bump 后本地最多陈旧 _LOCAL_VERSION_TTL。 _LOCAL_VERSION_TTL = 60.0 _local_versions: dict[str, tuple[int, float]] = {} # name -> (version, fetched_monotonic) async def cache_get(key: str) -> Any | None: c = _client() if c is None: return None try: raw = await c.get(key) return json.loads(raw) if raw is not None else None except Exception: # noqa: BLE001 —— 缓存层任何故障都不影响主流程 _bail() return None async def cache_set(key: str, value: Any, ttl: int) -> None: c = _client() if c is None: return try: # default=str:value 里混入 datetime/date 也不炸(炸了会触发 _bail 毒死整个缓存层) await c.set(key, json.dumps(value, ensure_ascii=False, default=str), ex=max(1, ttl)) except Exception: # noqa: BLE001 _bail() async def get_version(name: str) -> int: """读版本号(缺省 0)。版本号参与缓存 key:INCR 后旧 key 全部失效。 先查本地(60s),热请求 0 RTT。""" ent = _local_versions.get(name) now = time.monotonic() if ent is not None and ent[1] > now: return ent[0] c = _client() if c is None: return 0 try: v = await c.get(f"ver:{name}") val = int(v) if v is not None else 0 _local_versions[name] = (val, now + _LOCAL_VERSION_TTL) return val except Exception: # noqa: BLE001 _bail() return 0 async def bump_version(name: str) -> None: c = _client() if c is None: return try: v = await c.incr(f"ver:{name}") # 同进程 bump 立即可见(夜同步跑在本进程时响应缓存零陈旧窗口) _local_versions[name] = (int(v), time.monotonic() + _LOCAL_VERSION_TTL) except Exception: # noqa: BLE001 _bail() async def aclose() -> None: """进程退出时释放连接池(由 main.lifespan 调用)。""" global _pool if _pool is not None: try: await _pool.disconnect() except Exception: # noqa: BLE001 pass _pool = None