Files
stock/backend/app/cache.py
cirry 2f9c8bee2b 日K加载提速:adj_factor 覆盖索引 + 冷路径并发 + 热路径进程内缓存直返
- adj_factor 覆盖索引 (ts_code,trade_date) INCLUDE (adj_factor):根治堆碎片化
  (单股 6516 行散 6516 块,满载时位图堆扫 1.5s+),Index Only Scan ~2ms;
  因子查询全部改 2 列投影,lag() 窗口只取变点(6516→32 行)
- preview 冷路径 2 会话 2 波并发,信息卡合并为 LEFT JOIN LATERAL 一条
- 鉴权会话 60s 进程内缓存(登出/全端登出即时失效),全站请求省 ~80ms
- pvj/chipsj/stocksj/facetsj 存 model_dump_json 原串直返(与 response_model
  字节一致),热路径 230-2190ms → 1-2ms;get_version 本地缓存+bump 即时可见
- 连接池 10+20;smoke_test 适配鉴权缓存
2026-09-02 16:52:28 +08:00

160 lines
5.7 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""Redis 读缓存(可选基础设施)。
- REDIS_URL 留空、连接失败或超时:所有操作静默退化为「无缓存」,接口照常直查数据库,
且本进程内禁用重试(避免每个请求都陪跑一次连接超时)。
- 失效策略TTL 自然过期 + 版本号INCR作废。自选股增删等写操作只 INCR 版本 key
旧缓存 key 里带着旧版本号,无需 SCAN 批量删除。
- 只缓存「读多写少、可容忍短暂陈旧」的聚合数据(股票列表、筛选项等);
K线/回测等口径敏感数据不走这里。
- Redis 之前还有一层进程内本地缓存local_get/local_set0 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=strvalue 里混入 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。版本号参与缓存 keyINCR 后旧 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