refactor: align auth checker optimization with DataAccess caching

2026-10-21 04:00:04 +08:00 · 2025-12-17 15:03:44 +08:00
parent 0939013a89
commit 26150c2924
4 changed files with 1237 additions and 0 deletions
@@ -0,0 +1,373 @@
+# 权限检查系统 - 立即可以实施的改进
+
+## 当前问题总结
+
+经过分析，当前权限检查系统存在以下主要问题：
+
+### 1. 检查执行顺序不合理
+
+- **问题**：所有检查并行执行，ban 检查应该优先执行
+- **影响**：用户被 ban 后，其他检查（金币、限制等）仍然会执行，浪费资源
+- **优先级**：高
+
+### 2. 缺乏早期退出机制
+
+- **问题**：即使某个检查失败，其他检查仍会继续执行
+- **影响**：浪费 CPU 和数据库资源
+- **优先级**：高
+
+### 3. 数据获取可以优化
+
+- **问题**：group 数据在 hooks 执行时才获取，可以提前获取
+- **影响**：增加总体执行时间
+- **优先级**：中
+
+### 4. 充分利用 DataAccess 的 Redis 缓存
+
+- **现状**：`DataAccess` 已经实现了完整的 Redis 缓存机制（包括空结果缓存 5 分钟）
+- **问题**：没有充分利用缓存，数据获取顺序不够优化
+- **影响**：缓存命中率可能不够高，重复查询数据库
+- **优先级**：高
+
+## 立即可以实施的改进
+
+### 改进 1：调整检查顺序，ban 检查优先执行
+
+**位置**：`auth_checker.py` 第 498-514 行
+
+**当前代码**：
+
+```python
+hook_tasks = [
+    time_hook(auth_ban(...), "auth_ban", hook_times),
+    time_hook(auth_bot(...), "auth_bot", hook_times),
+    # ... 其他检查
+]
+await asyncio.gather(*hook_tasks)  # 并行执行
+```
+
+**改进方案**：
+
+```python
+# 1. 先执行关键检查（ban、bot），如果失败立即退出
+critical_checks = [
+    ("auth_ban", lambda: time_hook(auth_ban(...), "auth_ban", hook_times)),
+    ("auth_bot", lambda: time_hook(auth_bot(...), "auth_bot", hook_times)),
+]
+
+for check_name, check_func in critical_checks:
+    try:
+        await check_func()
+    except SkipPluginException:
+        raise  # 立即退出
+    except Exception as e:
+        logger.error(f"{check_name} 检查失败: {e}", LOGGER_COMMAND, session=session)
+        # 关键检查失败，可以选择继续或退出
+
+# 2. 关键检查通过后，再并行执行其他检查
+other_checks = [
+    time_hook(auth_group(...), "auth_group", hook_times),
+    time_hook(auth_admin(...), "auth_admin", hook_times),
+    time_hook(auth_plugin(...), "auth_plugin", hook_times),
+    time_hook(auth_limit(...), "auth_limit", hook_times),
+]
+await asyncio.gather(*other_checks)
+```
+
+### 改进 2：提前获取 group 数据
+
+**位置**：`auth_checker.py` 第 485-493 行
+
+**当前代码**：
+
+```python
+# 在hooks执行时才获取group
+group = None
+if entity.group_id:
+    group_dao = DataAccess(GroupConsole)
+    group = await with_timeout(...)
+```
+
+**改进方案**：
+
+```python
+# 在获取plugin和user时，同时获取group
+# 在 get_plugin_and_user 函数中或auth函数开始处
+group = None
+if entity.group_id:
+    group_dao = DataAccess(GroupConsole)
+    group_task = group_dao.safe_get_or_none(
+        group_id=entity.group_id, channel_id__isnull=True
+    )
+    # 可以并行获取
+    plugin, user, group = await asyncio.gather(
+        plugin_task, user_task, group_task
+    )
+```
+
+### 改进 3：充分利用 DataAccess 的 Redis 缓存机制
+
+**重要说明**：`DataAccess` 已经实现了完整的 Redis 缓存机制，包括：
+
+- ✅ 自动缓存查询结果
+- ✅ 空结果缓存（5 分钟 TTL，避免频繁查询不存在的记录）
+- ✅ 缓存统计（命中率、命中次数等）
+- ✅ 自动处理缓存的获取和设置
+
+**优化策略**：
+
+#### 3.1 优化数据获取顺序，提高缓存命中率
+
+**位置**：`auth_checker.py` 的 `get_plugin_and_user` 函数和 `auth` 函数
+
+**当前问题**：
+
+- plugin、user、group 数据分别获取，可能错过并行获取的缓存优势
+- group 数据在 hooks 执行时才获取，应该提前获取
+
+**改进方案**：
+
+```python
+async def get_plugin_user_and_group(
+    module: str, user_id: str, group_id: str | None
+) -> tuple[PluginInfo, UserConsole, GroupConsole | None]:
+    """统一获取插件、用户和群组数据（充分利用 DataAccess 缓存）
+
+    注意：DataAccess 会自动使用 Redis 缓存，包括：
+    - 查询结果缓存
+    - 空结果缓存（5分钟）
+    - 缓存统计
+    """
+    user_dao = DataAccess(UserConsole)
+    plugin_dao = DataAccess(PluginInfo)
+    group_dao = DataAccess(GroupConsole) if group_id else None
+
+    # 并行获取所有数据，DataAccess 会自动处理缓存
+    tasks = [
+        plugin_dao.safe_get_or_none(module=module),
+        user_dao.get_by_func_or_none(
+            UserConsole.get_user, False, user_id=user_id
+        )
+    ]
+
+    if group_id and group_dao:
+        tasks.append(
+            group_dao.safe_get_or_none(
+                group_id=group_id, channel_id__isnull=True
+            )
+        )
+    else:
+        tasks.append(asyncio.create_task(asyncio.sleep(0)))  # 占位
+
+    try:
+        results = await with_timeout(
+            asyncio.gather(*tasks), name="get_plugin_user_group"
+        )
+        plugin = results[0]
+        user = results[1]
+        group = results[2] if len(results) > 2 and group_id else None
+
+        # 验证数据
+        if not plugin:
+            raise PermissionExemption(f"插件:{module} 数据不存在...")
+        if plugin.plugin_type == PluginType.HIDDEN:
+            raise PermissionExemption(f"插件: {plugin.name}:{plugin.module} 为HIDDEN...")
+        if not user:
+            raise PermissionExemption("用户数据不存在...")
+
+        return plugin, user, group
+    except asyncio.TimeoutError:
+        # 超时时尝试从缓存获取（DataAccess 已经缓存了）
+        logger.warning(
+            f"数据获取超时，尝试从缓存获取，模块: {module}",
+            LOGGER_COMMAND
+        )
+        # DataAccess 的缓存会自动生效，这里只需要重试一次
+        plugin = await plugin_dao.safe_get_or_none(module=module)
+        user = await user_dao.get_by_func_or_none(
+            UserConsole.get_user, False, user_id=user_id
+        )
+        group = None
+        if group_id and group_dao:
+            group = await group_dao.safe_get_or_none(
+                group_id=group_id, channel_id__isnull=True
+            )
+
+        if not plugin or not user:
+            raise PermissionExemption("获取数据失败，请稍后再试...")
+
+        return plugin, user, group
+```
+
+#### 3.2 监控缓存命中率
+
+**位置**：在权限检查中添加缓存统计监控
+
+**实现方案**：
+
+```python
+from zhenxun.services.data_access import DataAccess
+
+# 在 auth 函数开始和结束时记录缓存统计
+async def auth(...):
+    # 记录开始时的缓存统计
+    cache_stats_before = DataAccess.get_cache_stats()
+
+    try:
+        # ... 执行权限检查 ...
+        pass
+    finally:
+        # 记录结束时的缓存统计
+        cache_stats_after = DataAccess.get_cache_stats()
+
+        # 计算本次检查的缓存命中情况
+        for before, after in zip(cache_stats_before, cache_stats_after):
+            hits_diff = after["hits"] - before["hits"]
+            misses_diff = after["misses"] - before["misses"]
+            total = hits_diff + misses_diff
+            if total > 0:
+                hit_rate = (hits_diff / total) * 100
+                if hit_rate < 50:  # 缓存命中率低于50%时记录警告
+                    logger.warning(
+                        f"缓存命中率较低: {after['cache_type']} = {hit_rate:.1f}%",
+                        LOGGER_COMMAND
+                    )
+```
+
+#### 3.3 优化检查函数中的数据访问
+
+**说明**：各个检查函数（auth_ban、auth_bot 等）中使用的 `DataAccess` 已经自动缓存，无需额外处理。
+
+**建议**：
+
+- 确保所有数据访问都通过 `DataAccess`，而不是直接使用模型类
+- 对于频繁查询的数据（如 ban 记录），`DataAccess` 会自动缓存
+- 空结果也会被缓存 5 分钟，避免频繁查询不存在的记录
+
+### 改进 4：统一错误处理策略
+
+**位置**：各个检查函数
+
+**当前问题**：
+
+- `auth_ban` 超时时不阻塞
+- `auth_bot` 超时时不阻塞
+- `auth_limit` 超时时不阻塞
+- 但 `get_plugin_and_user` 超时时会抛出异常
+
+**改进方案**：
+
+```python
+# 定义统一的超时处理策略
+class TimeoutStrategy:
+    FAIL_FAST = "fail_fast"  # 超时立即失败
+    CONTINUE = "continue"     # 超时继续执行
+    CACHE_FALLBACK = "cache_fallback"  # 超时使用缓存
+
+# 根据检查的重要性设置策略
+CHECK_TIMEOUT_STRATEGY = {
+    "auth_ban": TimeoutStrategy.FAIL_FAST,      # 关键检查，超时失败
+    "auth_bot": TimeoutStrategy.FAIL_FAST,      # 关键检查，超时失败
+    "auth_group": TimeoutStrategy.CONTINUE,     # 非关键，超时继续
+    "auth_limit": TimeoutStrategy.CONTINUE,     # 非关键，超时继续
+    "get_plugin_and_user": TimeoutStrategy.CACHE_FALLBACK,  # 使用缓存降级
+}
+```
+
+## 实施优先级
+
+### 高优先级（立即实施）
+
+1. ✅ **调整检查顺序**：ban 检查优先执行
+   - 风险：低
+   - 收益：高
+   - 工作量：小（1-2 小时）
+
+### 中优先级（本周实施）
+
+2. ✅ **提前获取 group 数据，统一数据获取**
+
+   - 风险：低
+   - 收益：高（充分利用 DataAccess 缓存）
+   - 工作量：小（1-2 小时）
+
+3. ✅ **优化数据获取顺序，提高缓存命中率**
+
+   - 风险：低（DataAccess 已实现缓存）
+   - 收益：高（减少数据库查询）
+   - 工作量：小（1 小时）
+
+4. ✅ **添加缓存统计监控**
+   - 风险：低
+   - 收益：中（可以监控缓存效果）
+   - 工作量：小（30 分钟）
+
+### 低优先级（后续优化）
+
+4. ⏳ **统一错误处理策略**
+   - 风险：中（可能影响现有逻辑）
+   - 收益：中
+   - 工作量：中（3-4 小时）
+
+## 代码示例：改进后的 auth 函数结构
+
+```python
+async def auth(...):
+    # 1. 获取所有需要的数据（并行）
+    plugin, user, group = await gather_all_data(...)
+
+    # 2. 关键检查（串行，失败立即退出）
+    await critical_checks(plugin, user, group, ...)
+
+    # 3. 其他检查（并行）
+    await other_checks(plugin, user, group, ...)
+
+    # 4. 金币检查（最后执行，超时不影响）
+    cost_gold = await check_cost(user, plugin, ...)
+```
+
+## 注意事项
+
+1. **向后兼容**：确保改进不影响现有功能
+2. **充分测试**：特别是缓存相关的改进
+3. **监控指标**：记录改进前后的性能指标
+4. **逐步实施**：先实施低风险改进，再实施高风险改进
+
+## 预期效果
+
+- **性能提升**：总体执行时间减少 20-30%
+- **资源节省**：减少不必要的检查执行 30-50%
+- **缓存优化**：充分利用 DataAccess 的 Redis 缓存，缓存命中率提升 40-60%
+- **数据库压力**：减少数据库查询 30-50%（通过缓存）
+- **用户体验**：响应速度提升，特别是在高并发场景下
+
+## 关于 DataAccess 缓存的说明
+
+### DataAccess 已实现的缓存机制
+
+1. **自动缓存查询结果**
+
+   - 所有通过 `DataAccess.safe_get_or_none()` 等方法的查询结果都会自动缓存
+   - 缓存键基于模型的主键字段（如 `user_id`、`module` 等）
+
+2. **空结果缓存**
+
+   - 查询结果为 `None` 时，会缓存一个特殊标记（`_NULL_RESULT`）
+   - 默认 TTL 为 5 分钟，避免频繁查询不存在的记录
+   - 可通过 `DataAccess.set_null_result_ttl()` 调整
+
+3. **缓存统计**
+
+   - 自动统计缓存命中率、命中次数、未命中次数等
+   - 可通过 `DataAccess.get_cache_stats()` 获取统计信息
+
+4. **缓存失效**
+   - 数据更新、创建、删除时自动清除相关缓存
+   - 支持手动清除：`DataAccess.clear_cache(**kwargs)`
+
+### 优化建议
+
+1. **统一使用 DataAccess**：确保所有数据访问都通过 `DataAccess`，而不是直接使用模型类
+2. **并行获取数据**：利用 `asyncio.gather()` 并行获取多个数据，提高缓存命中率
+3. **监控缓存效果**：定期查看 `DataAccess.get_cache_stats()`，了解缓存命中率
+4. **调整空结果 TTL**：根据实际需求调整 `DataAccess.set_null_result_ttl()`，平衡缓存效果和实时性