缓存架构设计

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

两级缓存架构

缓存分层设计

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

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

缓存层级对比

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

L1 缓存：进程内字典缓存

实现原理

dict_util 模块实现了带 TTL 的进程内字典缓存。数据首次查询时从 Redis 加载到进程内存，后续查询直接从内存读取，TTL 过期后自动重新加载。

python

# src/common/utils/dict_util.py
import time
import json
from typing import Dict

# 进程内缓存：{cache_key: (data, expire_timestamp)}
_cache: Dict[str, tuple] = {}

CACHE_TTL = 300  # 默认 5 分钟


def get_dict_map(dict_code: str) -> Dict[str, str]:
    """获取数据字典的 {value: label} 映射（进程内缓存 + Redis 二级缓存）"""
    cache_key = f"dict:{dict_code}"
    now = time.time()

    # L1: 进程内缓存命中
    if cache_key in _cache:
        data, expire_at = _cache[cache_key]
        if now < expire_at:
            return data
        del _cache[cache_key]

    # L2: Redis 缓存
    from src.core.redis_client import get_redis
    redis = get_redis()
    cached = redis.get(cache_key)
    if cached:
        data = json.loads(cached)
        _cache[cache_key] = (data, now + CACHE_TTL)
        return data

    # 数据库查询（L3 回源）
    from src.modules.dictionary.dict_item.repository import DictItemRepository
    items = DictItemRepository.get_all(dict_code=dict_code)
    data = {str(item.value): item.label for item in items}

    # 写回两级缓存
    _cache[cache_key] = (data, now + CACHE_TTL)
    redis.set(cache_key, json.dumps(data, ensure_ascii=False), ex=CACHE_TTL)
    return data


def invalidate_dict_cache(dict_code: str):
    """失效指定字典的缓存（L1 + L2）"""
    cache_key = f"dict:{dict_code}"

    # 清除 L1（进程内缓存）
    _cache.pop(cache_key, None)

    # 清除 L2（Redis 缓存）
    from src.core.redis_client import get_redis
    redis = get_redis()
    redis.delete(cache_key)

使用场景

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

缓存失效

字典数据变更时（新增/编辑/删除字典项），主动清除两级缓存：

python

# src/modules/dictionary/dict_item/service.py
class DictItemService(BaseService):
    def add(self, data) -> R:
        result = super().add(data)
        if result.code == 0:
            invalidate_dict_cache(data['dict_code'])  # 新增成功后失效缓存
        return result

    def update(self, data) -> R:
        result = super().update(data)
        if result.code == 0:
            invalidate_dict_cache(data['dict_code'])  # 更新成功后失效缓存
        return result

    @classmethod
    def before_delete(cls, ids):
        """删除前钩子：失效缓存"""
        for obj_id in ids:
            item = cls.repository.get_by_id(obj_id)
            if item:
                invalidate_dict_cache(item.dict_code)
        return result

L2 缓存：Redis 缓存

Redis 连接管理

python

# src/core/redis_client.py
import redis

# Redis 连接池
_pool = None

def get_redis():
    """获取 Redis 客户端"""
    global _pool
    if _pool is None:
        _pool = redis.ConnectionPool(
            host=REDIS_HOST,
            port=REDIS_PORT,
            password=REDIS_PASSWORD,
            db=REDIS_DB,
            decode_responses=True
        )
    return redis.Redis(connection_pool=_pool)

连接池在应用启动时创建，所有请求共享
同步客户端与 Flask 同步模型无缝集成

缓存应用场景

应用场景	数据结构	Key 格式	TTL	说明
JWT 黑名单	String	`jwt:blacklist:{token_hash}`	Token 剩余有效期	登出时将 Token 加入黑名单
登录失败锁	String	`login:lock:{username}`	600 秒	连续 5 次失败后锁定
登录失败计数	String (Lua)	`login:fail:{username}`	300 秒	原子递增 + 首次设置 TTL
数据字典缓存	String (JSON)	`dict:{dict_code}`	300 秒	字典项 {value: label} 映射
权限列表缓存	String (JSON)	`perm:user:{user_id}`	变更时失效	用户权限字符串列表
在线用户	String	`online:user:{user_id}`	Token 有效期	在线用户 Token 存储
验证码	String	`captcha:{uuid}`	120 秒	图形验证码文本

Lua 脚本

原子递增 + 过期

python

# 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
"""

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

为什么需要 Lua 脚本？

incr 和 expire 分开执行时，两步之间若进程崩溃，key 会以无 TTL 的状态永久残留。Lua 脚本保证原子性，消除此风险。

缓存策略对比

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

缓存失效策略

主动失效

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

python

# 字典项变更 → 清除字典缓存
invalidate_dict_cache(dict_code)

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

# 用户角色变更 → 清除该用户的权限缓存
invalidate_perm_cache(user_id)

TTL 自动过期

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

数据字典缓存：TTL 300 秒（5 分钟）
登录失败计数：TTL 300 秒（5 分钟）
验证码：TTL 120 秒（2 分钟）

缓存穿透防护

缓存穿透

当查询一个不存在的数据时，缓存和数据库都不会命中，导致每次请求都穿透到数据库。对于字典查询，如果 dict_code 不存在，应缓存空结果（空字典），避免反复查询数据库。

python

def get_dict_map(dict_code: str) -> Dict[str, str]:
    # ... 缓存查询逻辑 ...
    items = DictItemRepository.get_all(dict_code=dict_code)
    data = {str(item.value): item.label for item in items}
    # 即使 data 为空字典，也写入缓存，防止穿透
    _cache[cache_key] = (data, now + CACHE_TTL)
    return data

缓存监控

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

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

bash

# 查看 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 操作的原子性，防止进程崩溃导致的缓存异常。

🔐 认证与安全

📊 数据与配置

📝 日志与监控

🛡️ 防护与治理

🔄 事务管理

📁 文件处理

🔧 核心工具

🧩 辅助工具

缓存架构设计

两级缓存架构

缓存层级对比

L1 缓存：进程内字典缓存

实现原理

使用场景

缓存失效

L2 缓存：Redis 缓存

Redis 连接管理

缓存应用场景

Lua 脚本

原子递增 + 过期

缓存策略对比

缓存失效策略

主动失效

TTL 自动过期

缓存穿透防护

缓存监控

总结

Become a sponsor

企业级中台解决方案

企业级中台解决方案

缓存架构设计 ​

两级缓存架构 ​

缓存层级对比 ​

L1 缓存：进程内字典缓存 ​

实现原理 ​

使用场景 ​

缓存失效 ​

L2 缓存：Redis 缓存 ​

Redis 连接管理 ​

缓存应用场景 ​

Lua 脚本 ​

原子递增 + 过期 ​

缓存策略对比 ​

缓存失效策略 ​

主动失效 ​

TTL 自动过期 ​

缓存穿透防护 ​

缓存监控 ​

总结 ​

缓存架构设计

两级缓存架构

缓存层级对比

L1 缓存：进程内字典缓存

实现原理

使用场景

缓存失效

L2 缓存：Redis 缓存

Redis 连接管理

缓存应用场景

Lua 脚本

原子递增 + 过期

缓存策略对比

缓存失效策略

主动失效

TTL 自动过期

缓存穿透防护

缓存监控

总结