djangoadmin-wiki/DjangoAdmin_FastAPI_NaiveVue/zh/volume3/cache.md

缓存架构设计

本章详细描述 的缓存架构设计，包括两级缓存策略（进程内缓存 + Redis 缓存）、缓存应用场景、失效策略和代码实现。

两级缓存架构

:::tip 缓存分层设计 采用两级缓存架构：第一级为进程内字典缓存（dict_util），第二级为 Redis 分布式缓存。两级缓存配合使用，在保证数据一致性的同时最大化查询性能。 :::

┌─────────────────────────────────────────────────────────────────┐
│ 请求处理流程 │
│ │
│ ┌──────────┐ 命中 ┌──────────┐ 命中 ┌───────────┐ │
│ │ 业务代码 │ ─────────→ │ L1 缓存 │ ────────→ │ 返回结果 │ │
│ │ │ │ (进程内) │ │ │ │
│ └──────────┘ └──────────┘ └───────────┘ │
│ │ │ 未命中 │
│ │ ▼ │
│ │ ┌──────────┐ 命中 ┌───────────┐ │
│ │ │ L2 缓存 │ ────────→ │ 写回 L1 │ │
│ │ │ (Redis) │ │ 返回结果 │ │
│ │ └──────────┘ └───────────┘ │
│ │ │ 未命中 │
│ │ ▼ │
│ │ ┌──────────┐ ┌───────────┐ │
│ │ │ 数据库 │ ────────→ │ 写回 L1 │ │
│ │ │ (MySQL) │ │ 写回 L2 │ │
│ │ └──────────┘ └───────────┘ │
│ │ │
│ ┌──────────┐ │
│ │ 缓存失效 │ → 清除 L1 + 删除 L2 │
│ └──────────┘ │
└─────────────────────────────────────────────────────────────────┘

缓存层级对比

维度	L1 进程内缓存	L2 Redis 缓存
存储位置	Python 进程内存	Redis 服务器
访问速度	纳秒级（字典查找）	毫秒级（网络 I/O）
数据共享	仅当前进程	所有进程/实例共享
持久化	进程重启丢失	持久化可配置
容量限制	受进程内存限制	受 Redis 内存限制
适用场景	高频读取、低变更数据	跨进程共享、分布式场景

进程内字典缓存

设计决策

dict_util 采用纯进程内 TTL 缓存（无 Redis L2），原因：

1. 字典数据几乎不变，读写均落在同步 service 路径，无法使用请求级异步 Redis
2. 字典/字典项增删改时由 service 主动失效，TTL 仅作兜底
3. 多 worker 部署下各进程缓存最多在 TTL 内短暂不一致，属可接受范围

实现原理

# src/utils/dict_util.py
import logging
import time

from modules.dictionary.dict.models import Dict
from modules.dictionary.dict_item.models import DictItem
from modules.dictionary.dict.repository import dict_repo
from modules.dictionary.dict_item.repository import dictitem_repo

logger = logging.getLogger(__name__)

# 字典缓存 TTL（秒）：兜底过期时间，实际变更靠主动失效即时生效
_CACHE_TTL = 60
# 进程内字典缓存：{code: (过期时间戳, 字典项列表)}
_dict_cache: dict = {}


def get_dict_items(code):
    """按字典编码读取字典项列表 [{label, value, sort}, ...]，字典不存在返回空列表"""
    now = time.time()
    hit = _dict_cache.get(code)
    if hit and hit[0] > now:
        return hit[1]
    items = _load_dict_items(code)
    _dict_cache[code] = (now + _CACHE_TTL, items)
    return items


def get_dict_map(code):
    """按字典编码读取 {value: label} 映射（value 为字典项存储值，字符串类型）"""
    return {item['value']: item['label'] for item in get_dict_items(code)}


def get_dict_label(code, value, default=None):
    """取单个字典项名称：value 自动转字符串匹配存储值，查不到返回 default"""
    return get_dict_map(code).get(str(value), default)


def invalidate_dict(code):
    """清除指定字典编码的缓存（None/未缓存时安全无操作）"""
    if code:
        _dict_cache.pop(code, None)


def invalidate_dicts_by_ids(dict_ids):
    """按字典ID批量清除缓存：字典项变更后据 dict_id 反查 code 再失效"""
    if not dict_ids:
        return
    try:
        rows = dict_repo.filter(Dict.id.in_(dict_ids), Dict.is_delete == 0).all()
        for row in rows:
            invalidate_dict(row.code)
    except Exception as e:
        logger.warning("字典缓存失效失败（将按 TTL 自动过期）: %s", e)


def _load_dict_items(code):
    """从数据库读取字典项（未删除、按 sort+id 升序），字典不存在返回空列表"""
    dict_obj = dict_repo.get_one(code=code)
    if not dict_obj:
        return []
    rows = dictitem_repo.filter_by(dict_id=dict_obj.id).all()
    rows.sort(key=lambda item: (item.sort, item.id))
    return [{'label': item.name, 'value': item.value, 'sort': item.sort} for item in rows]

公开接口

函数	签名	说明
get_dict_items	(code) → [{label, value, sort}]	字典项列表（供前端下拉）
get_dict_map	(code) → {value: label}	值→显示名映射（供后端翻译）
get_dict_label	(code, value, default) → str	单值翻译
invalidate_dict	(code) → None	按字典编码清除缓存
invalidate_dicts_by_ids	(dict_ids) → None	按字典 ID 批量清除缓存

使用场景

场景	调用方式	说明
枚举显示名	get_dict_map('status_type')	序列化时将值转为显示名
表单下拉	get_dict_items('gender')	前端获取字典项列表
单值翻译	get_dict_label('status', 1)	将 1 转为 "正常"
业务校验	value in get_dict_map('xxx')	校验值是否在字典范围内

缓存失效

字典数据变更时（新增/编辑/删除字典项），通过 invalidate_dicts_by_ids 按字典 ID 批量清除：

# src/modules/dictionary/dict_item/service.py
class DictItemService(BaseService[DictItem]):
    repo = dictitem_repo
    model = DictItem

    def add(self, request, data):
        """新增字典项，新增后失效所属字典的缓存"""
        result = super().add(request, data)
        if data.dictId:
            invalidate_dicts_by_ids({data.dictId})
        return result

    def update(self, request, data):
        """更新字典项，更新后失效新旧所属字典的缓存"""
        old = self.repo.get_by_id(data.id) if data.id else None
        result = super().update(request, data)
        dict_ids = {data.dictId} if data.dictId else set()
        if old and old.dict_id:
            dict_ids.add(old.dict_id)
        invalidate_dicts_by_ids(dict_ids)
        return result

    def delete(self, ids):
        """删除字典项，删除后失效所属字典的缓存"""
        from utils.request import parse_id_list
        id_list = parse_id_list(str(ids))
        dict_ids = set()
        if id_list:
            for row in dictitem_repo.filter(DictItem.id.in_(id_list), DictItem.is_delete == 0).all():
                if row.dict_id:
                    dict_ids.add(row.dict_id)
        result = super().delete(ids)
        invalidate_dicts_by_ids(dict_ids)
        return result

L2 缓存：Redis 缓存

Redis 连接管理

# src/core/redis_client.py
from contextvars import ContextVar
from redis.asyncio import Redis

_redis_ctx: ContextVar[Redis] = ContextVar('_redis_ctx')


def get_redis() -> Redis:
    """获取当前请求的 Redis 客户端"""
    return _redis_ctx.get()

• 每个请求通过中间件注入独立的 Redis 客户端到 contextvars
• 连接池在应用启动时创建，所有请求共享
• 异步客户端（redis.asyncio）与 FastAPI 异步模型无缝集成

缓存应用场景

应用场景	数据结构	Key 格式	TTL	说明
JWT 黑名单	String	jwt:blacklist:{token_hash}	Token 剩余有效期	登出时将 Token 加入黑名单
登录失败锁	String	login:lock:{username}	600 秒	连续 5 次失败后锁定
登录失败计数	String (Lua)	login:fail:{username}	300 秒	原子递增 + 首次设置 TTL
滑动窗口限流	String (Lua)	rl:{ip}:{window_index}	2 倍窗口时长	滑动窗口计数器（前缀可配置）
权限列表缓存	String (JSON)	perm:list:{user_id}:{version}	7200 秒	用户权限字符串列表（版本号策略）
菜单缓存	String (JSON)	perm:menu:{user_id}:{version}	7200 秒	用户菜单树（版本号策略）
用户有效状态	String	user:active:{user_id}	300 秒	login_required 每请求校验
验证码	String	captcha:{uuid}	300 秒	图形验证码文本

:::tip 字典缓存不在 Redis 中 数据字典（dict_util）采用纯进程内 TTL 缓存，不经过 Redis。原因是字典读取落在同步 service 路径，无法使用请求级异步 Redis 客户端。详见上方「进程内字典缓存」章节。 :::

Lua 脚本

原子递增 + 过期

# src/core/redis_client.py
_INCR_EXPIRE_LUA = """
local count = redis.call('incr', KEYS[1])
if count == 1 then
    redis.call('expire', KEYS[1], ARGV[1])
end
return count
"""

async def incr_with_expire(redis, key: str, seconds: int) -> int:
    """原子执行 incr，并在 key 首次创建时一并设置过期时间"""
    return await redis.eval(_INCR_EXPIRE_LUA, 1, key, seconds)

:::tip 为什么需要 Lua 脚本？ incr 和 expire 分开执行时，两步之间若进程崩溃，key 会以无 TTL 的状态永久残留，导致基于 Redis 的限流/登录锁定永久封禁。Lua 脚本保证原子性，消除此风险。 :::

滑动窗口限流

_SLIDING_WINDOW_LUA = """
local cur = redis.call('incr', KEYS[1])
if cur == 1 then
 redis.call('expire', KEYS[1], ARGV[1])
end
local prev = tonumber(redis.call('get', KEYS[2]) or '0')
local elapsed = tonumber(ARGV[3]) - tonumber(ARGV[2]) * tonumber(ARGV[1])
local f = elapsed / tonumber(ARGV[1])
if f < 0 then f = 0 elseif f > 1 then f = 1 end
local effective = prev * (1 - f) + cur
return effective
"""

滑动窗口限流通过当前窗口与上一窗口的计数按时间比例加权，消除固定窗口边界处的 2x 突发问题。

缓存策略对比

策略	说明	适用场景	应用
Cache-Aside	先查缓存，未命中查 DB，写回缓存	读多写少	数据字典、权限列表
Write-Through	写操作同时更新缓存	数据一致性要求高	字典项变更时主动失效
Write-Behind	写操作只更新缓存，异步写 DB	写入性能要求高	未使用
TTL 过期	缓存自动过期	允许短暂不一致	登录锁、限流计数
主动失效	数据变更时主动清除缓存	数据一致性要求高	字典缓存、权限缓存

缓存失效策略

主动失效

数据变更时主动清除相关缓存，保证数据一致性：

# 字典项变更 → 按字典 ID 批量清除字典缓存
invalidate_dicts_by_ids(dict_ids)

# 菜单变更 → 递增全局版本号，所有用户权限缓存自然失效
await invalidate_menu_change()

# 用户角色变更 → 删除该用户的权限和菜单缓存
await invalidate_user_perms(user_id)

# 角色菜单变更 → 清除该角色下所有用户的权限缓存
await invalidate_role_perms(role_id)

TTL 自动过期

对于允许短暂不一致的数据，使用 TTL 自动过期：

数据字典缓存：TTL 60 秒（1 分钟，兜底过期，实际变更靠主动失效即时生效）
登录失败计数：TTL 300 秒（5 分钟）
滑动窗口限流：TTL 2 倍窗口时长
验证码：TTL 300 秒（5 分钟）

缓存穿透防护

:::warning 缓存穿透 当查询一个不存在的字典编码时，缓存和数据库都不会命中，导致每次请求都穿透到数据库。dict_util 通过缓存空列表来防护：即使 _load_dict_items 返回空列表，也写入缓存，避免反复查询数据库。 :::

def get_dict_items(code):
    now = time.time()
    hit = _dict_cache.get(code)
    if hit and hit[0] > now:
        return hit[1]
    items = _load_dict_items(code)  # 字典不存在时返回 []
    _dict_cache[code] = (now + _CACHE_TTL, items)  # 空列表也缓存，防止穿透
    return items

缓存监控

Redis 缓存的运行状态可通过以下方式监控：

1. Redis INFO：连接数、内存使用、命中率
2. 应用日志：缓存命中/未命中日志（DEBUG 级别）
3. Redis TTL：检查 key 的剩余有效期

# 查看 Redis 缓存统计
redis-cli INFO stats | grep -E "keyspace_hits|keyspace_misses"

# 查看指定 key 的 TTL
redis-cli TTL "dict:status_type"

# 查看所有字典缓存 key
redis-cli KEYS "dict:*"

总结

采用进程内缓存（L1）+ Redis 缓存（L2）的两级缓存架构。L1 缓存提供纳秒级访问速度，适合高频读取的字典数据；L2 缓存提供跨进程数据共享，适合分布式场景。缓存失效采用主动失效 + TTL 过期的混合策略，数据变更时主动清除缓存保证一致性，允许短暂不一致的场景使用 TTL 自动过期。Lua 脚本保证 Redis 操作的原子性，防止进程崩溃导致的缓存异常。