📋 本文概览
学习目标：
- 理解多租户架构的核心概念和设计模式
- 掌握三种主流数据隔离策略的实现
- 学会使用PostgreSQL Row-Level Security（RLS）
- 实现细粒度的资源配额管理系统
- 构建完整的租户计费统计模块
- 掌握多租户系统的安全最佳实践
技术栈：
- PostgreSQL（数据库 + RLS）
- SQLAlchemy（ORM + 多租户支持）
- FastAPI（租户上下文管理）
- Redis（配额缓存）
- Celery（后台计费任务）
- Python（核心业务逻辑）
预计阅读时间：75分钟
前置知识要求：
- 熟悉PostgreSQL数据库
- 掌握SQLAlchemy ORM
- 了解FastAPI框架（参考第3篇）
- 理解用户认证系统（参考第6篇）
- 了解工作流执行引擎（参考第4-5篇）
🎯 业务场景
为什么需要多租户架构？
在SaaS（Software as a Service）模式下，多租户架构是必不可少的。它允许单个应用实例服务于多个客户（租户），同时保证数据隔离和资源公平分配。
场景1：数据隔离需求
问题： - 公司A和公司B都使用QuantumFlow - 公司A不能看到公司B的工作流 - 公司B不能访问公司A的执行记录 - 数据泄露会导致严重的法律和商业后果 传统单租户方案： - 为每个客户部署独立系统 - 成本高昂（服务器、维护、升级） - 无法规模化 多租户解决方案： - 单一代码库，多个租户 - 数据库级别隔离 - 应用层访问控制 - 成本降低80%+场景2：资源配额管理
需求： - Free用户：每月100次执行 - Pro用户：每月10,000次执行 - Enterprise用户：无限执行 挑战： - 如何实时统计使用量？ - 如何防止超额使用？ - 如何优雅地限流？ - 如何处理配额耗尽？ 解决方案： - Redis实时计数 - 中间件拦截超额请求 - 友好的配额提示 - 自动升级引导场景3：计费统计
需求： - 按执行次数计费 - 按节点数量计费 - 按存储空间计费 - 生成月度账单 挑战： - 如何准确统计？ - 如何处理时区？ - 如何防止重复计费？ - 如何生成发票？ 解决方案： - 事件驱动计费 - 幂等性保证 - 定时任务对账 - Stripe集成业界解决方案对比
| 隔离策略 | 优势 | 劣势 | 成本 | 适用场景 |
|---|---|---|---|---|
| 独立数据库 | 完全隔离、易于迁移 | 成本高、维护复杂 | 高 | 大客户、合规要求 |
| 共享数据库+独立Schema | 平衡性好、中等隔离 | 连接数限制 | 中 | 中型客户 |
| 共享数据库+共享Schema | 成本低、易扩展 | 隔离弱、查询复杂 | 低 | 小客户、初创 |
| 混合模式 | 灵活、可定制 | 架构复杂 | 中 | 多层级客户 |
🏗️ 架构设计
整体架构图
graph TB subgraph "客户端层" CLIENT_A[租户A客户端] CLIENT_B[租户B客户端] CLIENT_C[租户C客户端] end subgraph "API网关层" GATEWAY[API Gateway] TENANT_RESOLVER[租户识别] QUOTA_CHECK[配额检查] end subgraph "应用层" APP[FastAPI应用] TENANT_CTX[租户上下文] QUOTA_MGR[配额管理器] BILLING[计费引擎] end subgraph "数据隔离层" subgraph "策略1: 独立数据库" DB_A[(租户A数据库)] DB_B[(租户B数据库)] end subgraph "策略2: 共享数据库+RLS" SHARED_DB[(共享数据库)] RLS[Row-Level Security] end end subgraph "缓存层" REDIS_QUOTA[配额缓存] REDIS_TENANT[租户配置] end subgraph "后台任务" CELERY[Celery Worker] BILLING_TASK[计费任务] QUOTA_RESET[配额重置] end CLIENT_A --> GATEWAY CLIENT_B --> GATEWAY CLIENT_C --> GATEWAY GATEWAY --> TENANT_RESOLVER TENANT_RESOLVER --> QUOTA_CHECK QUOTA_CHECK --> APP APP --> TENANT_CTX TENANT_CTX --> QUOTA_MGR TENANT_CTX --> BILLING TENANT_CTX --> DB_A TENANT_CTX --> DB_B TENANT_CTX --> SHARED_DB SHARED_DB --> RLS QUOTA_MGR --> REDIS_QUOTA TENANT_CTX --> REDIS_TENANT BILLING --> CELERY CELERY --> BILLING_TASK CELERY --> QUOTA_RESET style TENANT_RESOLVER fill:#3B82F6 style RLS fill:#10B981 style QUOTA_MGR fill:#F59E0B style BILLING fill:#EF4444核心模块说明
1. 租户识别层
- 从请求中提取租户标识（域名/Header/Token）
- 加载租户配置（计划、配额、设置）
- 注入租户上下文到请求生命周期
2. 数据隔离层
- 独立数据库：每个租户独立的PostgreSQL实例
- 共享数据库+RLS：使用PostgreSQL行级安全策略
- 动态连接路由：根据租户选择数据库连接
3. 配额管理层
- 实时配额检查：Redis计数器
- 配额耗尽处理：友好提示+升级引导
- 配额重置：定时任务按周期重置
4. 计费统计层
- 事件采集：工作流执行、节点运行、存储使用
- 计费计算：按量计费、阶梯定价
- 账单生成：月度汇总、发票导出
数据流图
sequenceDiagram participant Client as 客户端 participant Gateway as API网关 participant Resolver as 租户识别 participant Quota as 配额检查 participant App as 应用层 participant DB as 数据库 participant Redis as Redis participant Billing as 计费引擎 Client->>Gateway: 请求（带租户标识） Gateway->>Resolver: 识别租户 alt 从域名识别 Resolver->>Resolver: 解析子域名 else 从Header识别 Resolver->>Resolver: 读取X-Tenant-ID else 从Token识别 Resolver->>Resolver: 解析JWT中的tenant_id end Resolver->>Redis: 加载租户配置 Redis-->>Resolver: 返回配置 Resolver->>Quota: 检查配额 Quota->>Redis: 获取当前使用量 Redis-->>Quota: 返回使用量 alt 配额充足 Quota->>App: 允许请求 App->>DB: 执行业务逻辑（带租户过滤） DB-->>App: 返回结果 App->>Redis: 增加使用量 App->>Billing: 记录计费事件 App-->>Client: 返回响应 else 配额耗尽 Quota-->>Client: 返回403（配额耗尽） end Billing->>Billing: 异步计费💻 代码实现
1. 租户模型定义
# models/tenant.py from sqlalchemy import Column, String, Integer, Boolean, DateTime, JSON, Enum as SQLEnum from sqlalchemy.orm import relationship from datetime import datetime from enum import Enum from database import Base class TenantPlan(str, Enum): """租户计划枚举""" FREE = "free" PRO = "pro" ENTERPRISE = "enterprise" class TenantStatus(str, Enum): """租户状态枚举""" ACTIVE = "active" SUSPENDED = "suspended" CANCELLED = "cancelled" class IsolationStrategy(str, Enum): """数据隔离策略枚举""" DEDICATED_DB = "dedicated_db" # 独立数据库 SHARED_DB_RLS = "shared_db_rls" # 共享数据库+RLS SHARED_DB_FILTER = "shared_db_filter" # 共享数据库+应用过滤 class Tenant(Base): """ 租户模型 存储租户的基本信息、计划、配额、计费配置等 """ __tablename__ = "tenants" # 基本信息 id = Column(String(36), primary_key=True) name = Column(String(255), nullable=False) slug = Column(String(100), unique=True, nullable=False, index=True) domain = Column(String(255), unique=True, nullable=True) # 计划与状态 plan = Column(SQLEnum(TenantPlan), default=TenantPlan.FREE, nullable=False) status = Column(SQLEnum(TenantStatus), default=TenantStatus.ACTIVE, nullable=False) # 数据隔离策略 isolation_strategy = Column( SQLEnum(IsolationStrategy), default=IsolationStrategy.SHARED_DB_RLS, nullable=False ) database_url = Column(String(500), nullable=True) # 独立数据库URL # 配额配置（JSON字段） quota_config = Column(JSON, default={ "workflows": 10, # 工作流数量限制 "executions_per_month": 100, # 每月执行次数 "nodes_per_workflow": 20, # 每个工作流节点数 "storage_mb": 100, # 存储空间（MB） "concurrent_executions": 1, # 并发执行数 }) # 当前使用量（JSON字段） current_usage = Column(JSON, default={ "workflows": 0, "executions_this_month": 0, "storage_mb": 0, "concurrent_executions": 0, }) # 计费配置 billing_email = Column(String(255), nullable=True) stripe_customer_id = Column(String(100), nullable=True) stripe_subscription_id = Column(String(100), nullable=True) # 特性开关（JSON字段） features = Column(JSON, default={ "custom_connectors": False, "advanced_analytics": False, "priority_support": False, "sso": False, "audit_logs": False, }) # 元数据 metadata = Column(JSON, default={}) created_at = Column(DateTime, default=datetime.utcnow, nullable=False) updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow) # 关系 users = relationship("User", back_populates="tenant") workflows = relationship("Workflow", back_populates="tenant") def __repr__(self): return f"<Tenant {self.slug} ({self.plan})>" def has_quota(self, resource: str, amount: int = 1) -> bool: """ 检查是否有足够的配额 Args: resource: 资源类型（workflows/executions_per_month等） amount: 需要的数量 Returns: bool: 是否有足够配额 """ quota = self.quota_config.get(resource, 0) usage = self.current_usage.get(resource, 0) # Enterprise计划无限配额 if self.plan == TenantPlan.ENTERPRISE: return True return (usage + amount) <= quota def increment_usage(self, resource: str, amount: int = 1): """增加使用量""" if resource not in self.current_usage: self.current_usage[resource] = 0 self.current_usage[resource] += amount def reset_monthly_usage(self): """重置月度使用量""" self.current_usage["executions_this_month"] = 0 def get_quota_percentage(self, resource: str) -> float: """获取配额使用百分比""" if self.plan == TenantPlan.ENTERPRISE: return 0.0 quota = self.quota_config.get(resource, 0) usage = self.current_
