Mars

外观

缓存与 Redis ​

概述 ​

后端统一基于 Redis 提供缓存与分布式能力，相关实现集中在 app/framework/common/redis.py。它在项目里承担的角色包括：

  • 会话与令牌缓存：登录态、Access/Refresh Token 的存活标记，支撑强制下线、滑动过期。
  • 热点数据缓存：系统参数、数据字典等几乎只读的数据，启动时预热，避免每次请求都查库。
  • HTTP 响应缓存：通过轻量的 fastapi_cache 装饰器缓存只读接口响应。
  • 分布式锁：定时任务初始化、BPM 流程超时等需要跨实例互斥的场景。
  • 发布/订阅：WebSocket 在集群多实例部署时，借助 Redis 频道完成跨进程广播。
  • 限流：请求限流器（fastapi-limiter 自研实现）的计数器也存储在 Redis 中。

管理端对 Redis 的连接监控、键值查看、在线用户等功能集中在基础设施模块，详见《05-基础设施/监控与连接》。

配置说明 ​

Redis 连接参数由配置中心统一管理（app/config/settings.py）：

配置项默认值说明
REDIS_ENABLETrue是否启用 Redis，关闭后相关功能不可用
REDIS_HOST127.0.0.1主机地址
REDIS_PORT6379端口
REDIS_DB_NAME1逻辑库号（0-15）
REDIS_USER空用户名（可选）
REDIS_PASSWORD空密码（可选）

连接池与超时复用数据库相关配置：POOL_SIZE（连接池大小，默认 10）、POOL_TIMEOUT（连接超时秒数，默认 30）。

Redis 是否在启动时建立连接由 EVENT_LIST 控制：仅当 REDIS_ENABLE 为 True 时，才会把 app.framework.common.redis.redis_connect 加入全局事件清单。

连接与生命周期 ​

redis_connect(app, status) 负责维护 FastAPI 应用级的 Redis 单例：

  • 启动时（status=True）：根据配置拼装 redis:// 连接串，使用 redis.asyncio.Redis.from_url 建立异步连接，开启 decode_responses（返回字符串）、health_check_interval=20（健康检查）与连接池；连接存入 app.state.redis，并通过 ping() 校验连通性。
  • 关闭时（status=False）：调用 app.state.redis.close() 释放连接。

在 app_init.py 的 lifespan 中，连接就绪后会依次完成一系列缓存预热与依赖初始化：系统参数缓存、数据字典缓存、定时任务调度器、HTTP 缓存 fastapi_cache 以及全局请求限流器。关闭阶段会主动清空 fastapi_cache。

键名规范 ​

系统内置键名通过 RedisKey 枚举集中定义，避免散落的字符串导致冲突或难以排查。每个枚举项包含 key（实际键名）与 remark（用途说明）：

  • ACCESS_TOKEN / REFRESH_TOKEN：令牌存活标记
  • USER_SESSION：用户会话信息
  • CAPTCHA_CODES：图片验证码
  • SYSTEM_CONFIG：系统配置
  • SYSTEM_DICT：数据字典
  • APSCHEDULER_LOCK_KEY：定时任务初始化锁
  • AI_MODEL_CONFIG：用户 AI 模型配置

业务代码中应优先复用 RedisKey.xxx.key，再拼接业务维度（如类型）组成完整键，例如字典键 system_dict:{tenant_id}:{dict_type}。

缓存工具类 RedisCURD ​

RedisCURD 是面向业务封装的缓存操作工具类，以 redis.asyncio.Redis 实例构造。推荐通过 RedisCURD(app.state.redis)（或注入的 redis 参数）使用。

常用能力一览：

  • 字符串读写：get(key)、set(key, value, expire=86400)、delete(*keys)、exists(key)、ttl(key)、expire(key, expire)。
  • 批量与模式：mget(keys)、get_keys(pattern)、clear(pattern)（按模式批量删除）。
  • 哈希结构：hash_set(name, key, value)、hash_get(name, keys)。
  • 监控统计：info()、db_size()、commandstats()。
  • 发布订阅：publish(channel, message)。
  • 分布式锁：lock、unlock、unlock_simple、renew_lock。

value 序列化策略：set 会自动处理——字节串直接存储，基础类型转为字符串，其余对象走 json.dumps(ensure_ascii=False)。读取时返回字符串或 None，需要结构化数据时自行 json.loads。

典型用法（参考系统字典缓存）：

python
from app.framework.common.redis import RedisCURD, RedisKey

# 写入：字典数据按 租户:类型 维度缓存，不设过期（随数据变更主动清理）
redis_key = f"{RedisKey.SYSTEM_DICT.key}:{tenant_id}:{dict_type}"
await RedisCURD(redis).set(key=redis_key, value=json.dumps(dict_data, ensure_ascii=False), expire=None)

# 读取
raw = await RedisCURD(redis).get(redis_key)
dict_data = json.loads(raw) if raw else None

# 变更后清理
await RedisCURD(redis).delete(redis_key)

启动时预热（init_cache）在应用启动阶段把全量字典/系统配置从数据库加载并写入 Redis，使高频只读接口直接命中缓存。

分布式锁 ​

RedisCURD 提供基于 SET NX 的分布式锁，并配合 Lua 脚本保证「谁加锁谁释放」「续约」的原子性：

  • lock(key, expire, value=None)：以 nx=True 抢占锁，返回 (是否成功, 锁值)；若不传 value 则自动生成 UUID。
  • unlock(key, value)：仅当持有者匹配时才删除，防止误删他人锁。
  • unlock_simple(key)：直接删除（适用于无需校验持有者的简单场景）。
  • renew_lock(key, expire, value)：持有者续期，避免长任务执行期间锁过期。

该能力用于定时任务调度器初始化（防止多实例重复初始化）以及 BPM 流程超时等需要跨实例互斥的场合。

发布/订阅与 WebSocket ​

WebSocket 连接管理器（framework/websocket/manager.py）在单机内按「用户类型 + 用户 ID」维护连接集合，支持单发与群发。当部署多个后端实例时，借助 RedisCURD.publish(channel, message) 将消息广播到 Redis 频道，由各实例的订阅端接收后转投本地连接，从而打通跨实例的消息投递。

HTTP 响应缓存 fastapi_cache ​

fastapi_cache 是一套极简的响应缓存方案，适合缓存「入参相同、结果稳定」的只读列表接口。核心入口在同一文件 app/framework/common/redis.py 中。

初始化在应用启动时完成：

python
from app.framework.common import redis as cache_util
await cache_util.init(redis=app.state.redis)   # prefix 默认 fastapi_cache，expire 默认 3600，可关闭

缓存装饰器 @cache ​

在控制器方法上使用 @cache(expire=300, namespace="xxx") 即可开启缓存：

python
from app.framework.common import redis as cache_util
from app.framework.common.redis import cache

_ROLE_NS = "role"

@cache(expire=300, namespace=_ROLE_NS)
async def get_role_list_controller(...):
    ...
  • expire：缓存过期秒数。
  • namespace：缓存命名空间，用于分组清理。
  • 缓存键由 namespace + 函数路径 + 入参哈希构成，对相同请求自动命中；命中时直接返回缓存的 JSON 响应。
  • 装饰器仅在 init 完成后、_ENABLE 为 True 且 Redis 可用时生效，否则直接透传原函数，对业务无侵入。

主动清理 clear ​

当数据发生写操作（增删改）后，应调用 cache_util.clear(namespace=...) 清理对应命名空间的缓存，保证一致性：

python
result = await RoleService(auth).create(data=data)
await cache_util.clear(namespace=_ROLE_NS)   # 失效角色相关缓存
return SuccessResponse(data=result_dict, msg="创建角色成功")

应用关闭阶段也会统一调用 cache_util.clear() 释放全部缓存键。

典型应用场景 ​

  • 会话与令牌：登录后将 access_token / refresh_token / user_session 写入 Redis，鉴权时校验其存在性实现强制下线；开启滑动过期后按 TTL 自动续期。
  • 系统参数与字典：启动时预热，中间件、限流白名单、IP 黑白名单、操作日志保留天数等高频读取项直接从 Redis 获取，并辅以 60 秒内存缓存进一步降低延迟。
  • IP 归属地：缓存 7 天，减少外部接口调用。
  • 限流：全局请求限流器在 Redis 中维护计数，支持集群维度统一限速。

前端相关 ​

前端为减少对后端的重复请求，对服务端已缓存的「系统参数」「数据字典」等资源通常只拉取一次并在本地（store / 内存）缓存复用；当管理端在「参数配置」「字典管理」中修改数据后，服务端会同步刷新 Redis 缓存，前端通过刷新或重新初始化即可获取最新值。监控页中展示的 Redis 连接信息、内存占用、命令统计等，均来自服务端对 RedisCURD.info() / db_size() / commandstats() 的封装接口（详见 05 基础设施）。

缓存一致性建议 ​

  • 读多写少的数据优先走 init_cache 预热 + 主动清理模式，避免缓存雪崩。
  • 写操作后务必清理对应 namespace 或精确键，防止脏读；清理采用「先更库、再清缓存」顺序。
  • 对设置了过期时间的键（如验证码、令牌），依赖 TTL 自然失效，无需手动清理。
  • 多实例部署时，写操作触发的 clear 只清除 Redis 中的键，配合 HTTP 缓存的命名空间机制即可保证所有实例一致。

常见问题 ​

  • 未开启 Redis：REDIS_ENABLE=False 时启动会抛出「请先开启Redis连接」，且缓存、限流、会话强制下线等功能不可用。请确认配置并启用。
  • 连接超时/认证失败：检查 REDIS_HOST / REDIS_PORT / REDIS_PASSWORD 是否正确，以及网络与防火墙。redis_connect 会对 TimeoutError、AuthenticationError 分别记录日志。
  • 缓存不生效：确认 fastapi_cache.init 是否已在 lifespan 中执行；装饰器对入参敏感，不同查询条件会生成不同键。
  • 数据不一致：确认增删改接口是否调用了 clear(namespace)，以及是否存在未覆盖到的写路径。