# 权限缓存 :::tip 说明 项目通过 Redis 实现用户权限和菜单的缓存,采用用户级缓存 + 全局版本号双层策略。源码位于 `src/core/permission.py`,供 `src/modules/system/menu/service.py` 中的权限查询函数调用。 ::: ## 缓存策略 采用「用户级缓存 + 全局版本号」双层策略: ``` 1. 每个用户的缓存 Key 带版本号后缀 2. 版本号存全局 Key(perm:version) 3. 菜单变更时递增全局版本号,所有用户缓存自然失效 4. 用户角色变更时,直接删除该用户缓存 5. 角色菜单变更时,查询该角色下所有用户 ID,批量删除缓存 ``` ## 缓存 Key 结构 ```python _VERSION_KEY = 'perm:version' # 全局版本号 _PERM_PREFIX = 'perm:list:' # 用户权限列表 _MENU_PREFIX = 'perm:menu:' # 用户菜单树 _USER_ACTIVE_PREFIX = 'user:active:' # 用户有效状态 ``` Key 格式示例: ``` perm:version → 5 perm:list:1001:5 → ["sys:user:page", "sys:user:add", ...] perm:menu:1001:5 → [{id: 1, name: "系统管理", children: [...]}] user:active:1001 → "1" ``` ## 权限缓存读写 ### get_cached_permissions() ```python def get_cached_permissions(user_id: int): """获取缓存的用户权限列表,未命中返回 None""" redis = get_redis() ver = _get_version() key = f'{_PERM_PREFIX}{user_id}:{ver}' data = redis.get(key) if data is not None: return json.loads(data) return None ``` ### set_cached_permissions() ```python def set_cached_permissions(user_id: int, perm_list: list): """缓存用户权限列表(空列表不缓存,避免缓存空壳后无法感知后续数据变更)""" if not perm_list: return redis = get_redis() ver = _get_version() key = f'{_PERM_PREFIX}{user_id}:{ver}' redis.set(key, json.dumps(perm_list), ex=_CACHE_TTL) ``` ## 缓存失效 ### 菜单变更 ```python def invalidate_menu_change(): """菜单增删改时调用,递增全局版本号使所有用户缓存失效""" _incr_version() ``` 递增版本号后,旧版本的缓存 Key 不再被访问,自然过期(TTL 2小时)。 ### 用户角色变更 ```python def invalidate_user_perms(user_id: int): """用户角色变更时调用,删除该用户的权限和菜单缓存""" redis = get_redis() ver = _get_version() keys = [ f'{_PERM_PREFIX}{user_id}:{ver}', f'{_MENU_PREFIX}{user_id}:{ver}', ] redis.delete(*keys) ``` ### 角色菜单变更 ```python def invalidate_role_perms(role_id: int): """角色菜单变更时调用,清理该角色下所有用户的权限和菜单缓存""" # 查询该角色下所有用户 ID user_ids = _query_user_ids(role_id) # 批量删除缓存 redis = get_redis() ver = _get_version() keys = [] for uid in user_ids: keys.append(f'{_PERM_PREFIX}{uid}:{ver}') keys.append(f'{_MENU_PREFIX}{uid}:{ver}') redis.delete(*keys) ``` ## 用户有效状态缓存 用于 `login_required` 每请求校验,避免每次请求都查库: ```python _USER_ACTIVE_TTL = 300 # 5分钟 def get_cached_user_active(user_id: int): """读取用户有效状态缓存;未命中返回 None(调用方需回源 DB 校验)""" def set_cached_user_active(user_id: int, active: bool): """写入用户有效状态缓存""" def invalidate_user_active(user_id: int): """用户状态/删除变更时调用,删除有效状态缓存""" ``` ## 异常降级 所有缓存操作都包裹 try/except,Redis 不可用时降级直查 DB: ```python except Exception as e: logger.warning(f"读取权限缓存异常: {e}") return None # 调用方回源 DB ``` 删除失败时递增全局版本号孤立旧缓存(fail-closed): ```python except Exception as e: logger.error(f"清理用户权限缓存异常,降级递增全局版本号: {e}") try: _incr_version() except Exception as e2: logger.error(f"降级递增版本号失败: {e2}") ``` :::tip 温馨提示 权限缓存的 TTL 为 2小时,用户有效状态缓存 TTL 为 5分钟。正常情况下缓存由业务操作即时失效,TTL 仅作为兜底防止 Redis 故障时缓存永久残留。 ::: ## 总结 权限缓存采用用户级缓存 + 全局版本号双层策略,菜单变更通过递增版本号使所有缓存自然失效,无需遍历删除。所有缓存操作都有异常降级机制,Redis 不可用时自动回源数据库查询。