订阅列表 + 使用记录查询优化

版本: 1.0.0
更新日期: 2026-02-08
状态: ✅ 已完成

本文档记录订阅列表查询与使用记录的优化方案和实施情况。

📚 目录

1. 优化场景
2. 优化方案
3. 实施详情
4. 性能提升

优化场景

问题描述

在订阅列表 API (/billing/subscriptions) 中，如果需要显示每个订阅的使用记录统计信息（如 Token 使用量、调用次数、余额等），可能会产生 N+1 查询问题：

# ❌ 潜在的 N+1 查询问题
subscriptions = await get_subscriptions()  # 1 次查询
for subscription in subscriptions:
    usage = await get_user_usage(subscription.user_id)  # N 次查询
    balance = await get_user_balance(subscription.user_id)  # N 次查询

当前实现

文件:

• shared/src/api/admin/billing_api.py - 订阅列表 API
• private/platform/src/api_admin/billing_api.py - 订阅列表 API

当前状态:

• ✅ 订阅列表查询已使用 LEFT JOIN 获取用户信息（无 N+1 问题）
• ✅ 使用统计和余额信息已使用批量查询（已优化）

优化方案

方案：批量获取使用统计和余额信息

使用 ANY(:user_ids) 和 GROUP BY 一次性获取所有订阅的使用统计和余额信息。

优化前（潜在问题）:

subscriptions = await get_subscriptions()  # 1 次查询
for subscription in subscriptions:
    usage = await get_user_usage(subscription.user_id)  # N 次查询
    balance = await get_user_balance(subscription.user_id)  # N 次查询

优化后:

# 1. 获取订阅列表
subscriptions = await get_subscriptions()  # 1 次查询

# 2. 批量获取使用统计（1 次查询）
user_ids = [s.user_id for s in subscriptions]
usage_stats = await batch_get_usage_stats(user_ids)  # 1 次批量查询

# 3. 批量获取余额信息（1 次查询）
balance_stats = await batch_get_balance_stats(user_ids)  # 1 次批量查询

# 4. 合并数据
for subscription in subscriptions:
    subscription.usage = usage_stats.get(subscription.user_id, {})
    subscription.balance = balance_stats.get(subscription.user_id, {})

总查询次数: 从 N+1 次减少到 3 次

实施详情

1. ✅ 订阅列表 API 优化

文件:

• shared/src/api/admin/billing_api.py:280-322
• private/platform/src/api_admin/billing_api.py:280-322

优化内容:

1. 在获取订阅列表后，批量查询所有订阅的使用统计
2. 批量查询所有订阅的余额信息
3. 将统计信息添加到订阅对象中

代码实现:

# 优化：批量获取所有订阅的使用统计（避免 N+1 查询）
user_ids = [str(s.user_id) for s in subscriptions]
usage_stats = {}
if user_ids:
    # 批量获取 Token 使用统计
    usage_sql = text("""
        SELECT 
            user_id,
            COALESCE(SUM(total_tokens), 0) as total_tokens_used,
            COUNT(*) as total_calls,
            MAX(created_at) as last_used_at
        FROM token_usage_logs
        WHERE user_id = ANY(:user_ids)
        GROUP BY user_id
    """)
    usage_result = await session.execute(usage_sql, {"user_ids": user_ids})
    for stat_row in usage_result.fetchall():
        usage_stats[str(stat_row.user_id)] = {
            "total_tokens_used": stat_row.total_tokens_used or 0,
            "total_calls": stat_row.total_calls or 0,
            "last_used_at": (
                stat_row.last_used_at.isoformat()
                if stat_row.last_used_at
                else None
            ),
        }

# 批量获取 Token 余额信息
balance_stats = {}
if user_ids:
    balance_sql = text("""
        SELECT 
            user_id,
            monthly_quota,
            used_this_month,
            overage_this_month,
            overage_cost_cents
        FROM user_token_balance
        WHERE user_id = ANY(:user_ids)
    """)
    balance_result = await session.execute(
        balance_sql, {"user_ids": user_ids}
    )
    for balance_row in balance_result.fetchall():
        balance_stats[str(balance_row.user_id)] = {
            "monthly_quota": balance_row.monthly_quota or 0,
            "used_this_month": balance_row.used_this_month or 0,
            "overage_this_month": balance_row.overage_this_month or 0,
            "overage_cost_cents": balance_row.overage_cost_cents or 0,
        }

# 合并统计信息到订阅对象
for s in subscriptions:
    subscription_data = {
        # ... 订阅基本信息 ...
        "usage": usage_stats.get(
            str(s.user_id),
            {
                "total_tokens_used": 0,
                "total_calls": 0,
                "last_used_at": None,
            },
        ),
        "balance": balance_stats.get(
            str(s.user_id),
            {
                "monthly_quota": 0,
                "used_this_month": 0,
                "overage_this_month": 0,
                "overage_cost_cents": 0,
            },
        ),
    }

性能提升

查询次数对比

查询分解:

• 获取订阅列表: 1 次（LEFT JOIN 用户信息）
• 批量获取使用统计: 1 次（GROUP BY user_id）
• 批量获取余额信息: 1 次（WHERE user_id = ANY(:user_ids)）

响应时间对比

• 优化前: 100ms + N × 10ms × 2（假设每次查询 10ms，需要查询使用统计和余额）
• 优化后: 100ms + 50ms + 30ms（批量查询）
• 提升: 随着订阅数量增加，性能提升显著

数据库压力

• 优化前: N+1 次数据库查询，高并发时压力大
• 优化后: 3 次数据库查询，压力显著降低

API 响应格式

优化后的 API 响应包含使用统计和余额信息：

{
  "success": true,
  "subscriptions": [
    {
      "id": 1,
      "user_id": "uuid",
      "email": "user@example.com",
      "plan_type": "pro",
      "status": "active",
      "start_date": "2026-01-01",
      "end_date": "2026-12-31",
      "auto_renew": true,
      "usage": {
        "total_tokens_used": 125000,
        "total_calls": 342,
        "last_used_at": "2026-02-08T10:30:00"
      },
      "balance": {
        "monthly_quota": 100000,
        "used_this_month": 25000,
        "overage_this_month": 0,
        "overage_cost_cents": 0
      }
    }
  ]
}

相关优化

1. ✅ 订阅列表 + 用户信息

文件: shared/src/api/admin/billing_api.py:289-298

状态: ✅ 已优化

• 使用 LEFT JOIN 一次性获取用户信息
• 无 N+1 问题

2. ✅ 子账号列表 + 配额信息

文件: private/platform/src/users/router.py:1865-1904

状态: ✅ 已优化

• 使用 LEFT JOIN 一次性获取配额信息
• 无 N+1 问题

最佳实践

1. 批量查询原则

当需要为多个对象获取关联数据时，优先使用批量查询：

# ✅ 好的做法：批量查询
user_ids = [subscription.user_id for subscription in subscriptions]
usage_stats = await batch_get_usage_stats(user_ids)

# ❌ 不好的做法：循环查询
for subscription in subscriptions:
    usage = await get_usage_stats(subscription.user_id)

2. 使用 GROUP BY 聚合

对于统计信息，使用 GROUP BY 一次性获取：

# ✅ 好的做法：使用 GROUP BY
SELECT user_id, SUM(total_tokens), COUNT(*)
FROM token_usage_logs
WHERE user_id = ANY(:user_ids)
GROUP BY user_id

# ❌ 不好的做法：分开查询
# 查询1: 总使用量
# 查询2: 调用次数

3. 合并多个统计查询

如果可能，将多个统计查询合并为一个：

# ✅ 好的做法：合并查询（如果可能）
SELECT 
    user_id,
    SUM(total_tokens) as total_tokens,
    COUNT(*) as total_calls,
    MAX(created_at) as last_used_at
FROM token_usage_logs
WHERE user_id = ANY(:user_ids)
GROUP BY user_id

# ⚠️ 注意：如果查询表不同，分开查询也是可以的
# 但都要使用批量查询，避免循环

相关资源

• N+1 查询优化指南
• 性能优化文档
• 用户列表 + 对话历史优化
• 专家列表 + 使用统计优化

文档版本: 1.0.0
最后更新: 2026-02-08