🏢 多租户架构

Schema 隔离 / Row-Level Security / Database-Level 隔离——如何在共享与隔离之间找到最佳平衡

01 什么是多租户

多租户 (Multi-Tenancy) 是 SaaS 的核心架构决策:一套应用代码服务多个客户 (租户/Tenant),但每个租户的数据互相不可见。

想想公寓楼——共享地基和外墙(应用层),但每户有自己的锁和房间(数据层)。区别在于:共享多少、隔离多少。

为什么这很重要? 多租户直接决定你的成本结构、运维复杂度、安全合规和扩展上限。选错了,后面每加一个客户都像拆房子重盖。
用户请求 租户识别 (Subdomain/Header/JWT) 租户上下文注入 数据隔离层 返回租户数据

租户识别的三种方式

🌐 子域名

acme.myapp.com → 解析子域名 → 查租户表

用户感知强 品牌定制

需要通配符 DNS + TLS 证书

// Next.js middleware 识别租户
export function middleware(req) {
  const host = req.headers.get('host');
  const subdomain = host.split('.')[0];
  if (subdomain === 'www' || subdomain === 'app') return NextResponse.next();
  // 查询租户
  const tenant = await db.tenant.findUnique({ where: { subdomain } });
  if (!tenant) return NextResponse.redirect('/404');
  // 注入到请求头
  req.headers.set('x-tenant-id', tenant.id);
  return NextResponse.next({ request: { headers: req.headers } });
}
📎 请求头 / JWT

X-Tenant-Id: t_abc123 或 JWT claim

实现简单 需防篡改

URL 无感知,不适合品牌定制场景

// FastAPI 中间件提取租户
from fastapi import Request, HTTPException

async def tenant_middleware(request: Request, call_next):
    # 从 JWT claims 提取
    tenant_id = request.state.user.get('tenant_id')
    if not tenant_id:
        raise HTTPException(403, "No tenant context")
    request.state.tenant_id = tenant_id
    response = await call_next(request)
    return response
🔀 路径前缀

/t/acme/dashboard → URL 路径包含租户标识

调试友好 URL 较长

# Django URL 配置
urlpatterns = [
    path('t/<slug:tenant_slug>/', include([
        path('dashboard/', views.dashboard),
        path('settings/', views.settings),
    ])),
]

02 三种隔离模型:深度对比

多租户的核心问题是数据隔离粒度,共有三种主流模型:

模型 1:Database-Level(独立数据库)

每个租户一个完整数据库。物理隔离最强,但运维成本最高。

✅ 优势

• 零数据泄露风险(物理隔离)
• 可独立备份/恢复
• 可按租户定制 Schema
• 合规性最强(HIPAA/FINRA)
• 故障隔离(一个 DB 坏不影响其他)
• 数据迁移简单(导出整个库)

❌ 劣势

• 连接池爆炸(1000 租户 = 1000 连接)
• 迁移成本 O(N):改表要跑 N 次
• 运维复杂度高(监控/升级/备份×N)
• 跨租户查询极难
• 基础设施成本高
• 新租户开通需创建库

-- 为每个租户创建独立数据库
CREATE DATABASE tenant_acme;
CREATE DATABASE tenant_globex;

-- 应用层连接路由
class TenantRouter:
    def get_connection(self, tenant_id: str):
        db_name = f"tenant_{tenant_id}"
        return connections[db_name]

模型 2:Schema-Level(共享数据库,独立 Schema)

同一个 PostgreSQL 数据库内,每个租户一个 Schema。中等隔离,中等成本。

✅ 优势

• 逻辑隔离(租户间不可见)
• 单连接池管理
• 比独立库运维简单
• 可按租户定制 Schema
• 备份粒度灵活
• 迁移比独立库快(同实例)

❌ 劣势

• Schema 漂移风险
• 迁移仍需 N 次(SET search_path)
• 连接数仍较高
• 跨租户查询需 UNION ALL
• 单点故障(PG 实例挂了全挂)
• Schema 数量上限(~1000 性能退化)

-- PostgreSQL Schema 隔离
CREATE SCHEMA tenant_acme;
CREATE SCHEMA tenant_globex;

-- 每个租户的表结构完全相同
CREATE TABLE tenant_acme.users (id UUID PRIMARY KEY, name TEXT, email TEXT);
CREATE TABLE tenant_globex.users (id UUID PRIMARY KEY, name TEXT, email TEXT);

-- 应用层自动切换 Schema
SET search_path TO tenant_acme;
SELECT * FROM users;  -- 只看到 acme 的数据

模型 3:Row-Level(共享表,tenant_id 列)

所有租户数据在同一张表,通过 tenant_id 列区分。最简单,但隔离最弱。

✅ 优势

• 最简单:一张表搞定
• 迁移一次:改表只跑一次
• 连接池最小
• 跨租户查询天然支持
• 运维成本最低
• 新租户零成本(插一行记录)

❌ 劣势

• 遗漏 tenant_id = 数据泄露
• 索引膨胀(所有租户数据混合)
• 查询性能随数据量退化
• 难以按租户备份/恢复
• 定制化困难(Schema 统一)
• 合规风险(共享资源)

-- Row-Level:所有租户共享一张表
CREATE TABLE users (
  id UUID PRIMARY KEY,
  tenant_id UUID NOT NULL REFERENCES tenants(id),
  name TEXT,
  email TEXT
);

-- 每次查询必须带 tenant_id
SELECT * FROM users WHERE tenant_id = 't_acme';

-- 索引优化:分区索引
CREATE INDEX idx_users_tenant ON users(tenant_id);
CREATE INDEX idx_users_tenant_email ON users(tenant_id, email);

三种模型全景对比

维度 Database-Level Schema-Level Row-Level
隔离强度 物理隔离 逻辑隔离 应用层隔离
运维复杂度 🔴 高 🟡 中 🟢 低
迁移成本 O(N) 跑 N 次 O(N) SET search_path O(1) 跑一次
连接池 N 个池 1 池 + search_path 1 池
跨租户查询 🔴 极难 🟡 UNION ALL 🟢 天然支持
数据泄露风险 🟢 极低 🟡 低 🔴 取决于代码质量
租户定制 🟢 完全自由 🟢 Schema 级定制 🔴 困难
备份/恢复粒度 🟢 租户级 🟡 Schema 级 🔴 全库级
适用规模 <100 租户 100-1000 租户 1000+ 租户
成本 $$$ $$ $
常见误区: "Row-Level 不安全"——实际上,PostgreSQL 的 RLS (Row-Level Security) 可以在数据库层强制隔离,即使应用代码遗漏了 tenant_id 过滤,数据库也会拦截。关键是你必须启用 RLS,不能只靠应用层。

03 PostgreSQL RLS:Row-Level 的救星

PostgreSQL 的 Row-Level Security (RLS) 是让 Row-Level 模型安全的关键——它在数据库层强制执行租户隔离,即使应用代码有 bug 也不会泄露数据。

RLS 完整配置

-- 1. 启用 RLS
ALTER TABLE users ENABLE ROW LEVEL SECURITY;
ALTER TABLE projects ENABLE ROW LEVEL SECURITY;
ALTER TABLE invoices ENABLE ROW LEVEL SECURITY;

-- 2. 创建策略:只允许看到自己租户的数据
CREATE POLICY tenant_isolation ON users
  USING (tenant_id = current_setting('app.current_tenant')::UUID);

CREATE POLICY tenant_isolation ON projects
  USING (tenant_id = current_setting('app.current_tenant')::UUID);

CREATE POLICY tenant_isolation ON invoices
  USING (tenant_id = current_setting('app.current_tenant')::UUID);

-- 3. 应用层设置租户上下文(每个请求开始时)
SET app.current_tenant = '550e8400-e29b-41d4-a716-446655440000';

-- 4. 查询自动过滤!不需要 WHERE tenant_id = ...
SELECT * FROM users;  -- 只返回当前租户的用户!

-- 5. 即使恶意查询也会被拦截
SELECT * FROM users WHERE email = 'admin@other-tenant.com';  -- 返回空!

应用层集成(Prisma + Node.js)

// Prisma Client Extension 实现 RLS
import { PrismaClient } from '@prisma/client'

const prisma = new PrismaClient()

export function createTenantClient(tenantId: string) {
  return prisma.$extends({
    query: {
      async $allOperations({ args, query, model }) {
        // 注入 tenant_id 到 where 条件
        if (model !== 'Tenant') {
          args.where = { ...args.where, tenantId }
        }
        return query(args)
      }
    },
    client: {
      async $setTenantContext() {
        // 设置 PostgreSQL session 变量
        await prisma.$executeRawUnsafe(
          `SET app.current_tenant = '${tenantId}'`
        )
      }
    }
  })
}

// 使用
const tenantPrisma = createTenantClient('t_acme')
await tenantPrisma.$setTenantContext()
const users = await tenantPrisma.user.findMany()  // 自动过滤

Django + RLS

# django-tenants 实现 Schema 隔离 + RLS
# settings.py
TENANT_MODEL = "tenants.Tenant"
TENANT_DOMAIN_MODEL = "tenants.Domain"

DATABASES = {
    'default': {
        'ENGINE': 'django_tenants.postgresql_backend',
        'NAME': 'myapp',
    }
}

# models.py
from django_tenants.models import TenantMixin, DomainMixin

class Tenant(TenantMixin):
    name = models.CharField(max_length=100)
    paid_until = models.DateField(null=True)
    on_trial = models.BooleanField(default=True)

    auto_create_schema = True  # 自动创建 Schema

class Domain(DomainMixin):
    pass

# 租户自动路由
# 访问 acme.myapp.com → 自动切换到 schema tenant_acme
# 代码里直接写 User.objects.all() → 只返回当前租户

RLS 性能优化

⚡ RLS 性能关键点
-- 分区表 + RLS 组合拳(大数据量最佳实践)
CREATE TABLE invoices (
  id UUID DEFAULT gen_random_uuid(),
  tenant_id UUID NOT NULL,
  amount DECIMAL(10,2),
  created_at TIMESTAMPTZ DEFAULT now()
) PARTITION BY LIST (tenant_id);

-- 为大租户创建独立分区
CREATE TABLE invoices_t_acme PARTITION OF invoices
  FOR VALUES IN ('550e8400-e29b-41d4-a716-446655440000');

-- 小租户合并在 default 分区
CREATE TABLE invoices_default PARTITION OF invoices DEFAULT;

-- RLS 仍然生效,但分区让查询只扫描目标分区
ALTER TABLE invoices ENABLE ROW LEVEL SECURITY;
CREATE POLICY tenant_isolation ON invoices
  USING (tenant_id = current_setting('app.current_tenant')::UUID);

04 混合模型:现实中的最优解

没有一种模型适合所有场景。真实 SaaS 通常采用混合模型——根据租户等级选择不同隔离策略。

按租户等级分层

租户等级 隔离模型 典型客户 月费
Free/Solo Row-Level + RLS 个人用户、小团队 $0-29
Pro/Small Row-Level + RLS + 分区 小公司 $29-99
Business Schema-Level 中型企业 $99-499
Enterprise Database-Level 大企业/合规需求 $499+
// 混合模型:租户路由器
class HybridTenantResolver {
  private strategies = {
    'row-level':  new RowLevelStrategy(),
    'schema':     new SchemaLevelStrategy(),
    'database':   new DatabaseLevelStrategy(),
  }

  async resolve(tenantId: string): Promise<TenantContext> {
    const tenant = await this.getTenantConfig(tenantId)

    const strategy = this.strategies[tenant.isolationLevel]

    return {
      tenantId,
      isolationLevel: tenant.isolationLevel,
      connection: await strategy.getConnection(tenantId),
      queryRunner: strategy.createQueryRunner(tenantId),
    }
  }
}

// 使用:中间件自动注入
app.use(async (req, res, next) => {
  const tenantId = extractTenantId(req)
  req.tenantCtx = await resolver.resolve(tenantId)
  next()
})

租户升级:Row → Schema → Database 迁移

1. 创建目标 Schema/DB 2. 运行 Migration 3. 数据迁移 (pg_dump/pg_restore) 4. 双写验证 5. 切换路由 6. 清理旧数据
-- 租户升级脚本:Row → Schema
-- Step 1: 创建 Schema
CREATE SCHEMA tenant_acme;

-- Step 2: 复制表结构
CREATE TABLE tenant_acme.users (LIKE public.users INCLUDING ALL);
CREATE TABLE tenant_acme.projects (LIKE public.projects INCLUDING ALL);

-- Step 3: 迁移数据
INSERT INTO tenant_acme.users SELECT * FROM public.users WHERE tenant_id = 't_acme';
INSERT INTO tenant_acme.projects SELECT * FROM public.projects WHERE tenant_id = 't_acme';

-- Step 4: 验证数据一致性
SELECT COUNT(*) FROM tenant_acme.users;
-- vs
SELECT COUNT(*) FROM public.users WHERE tenant_id = 't_acme';

-- Step 5: 更新租户配置(应用层路由切换)
UPDATE tenants SET isolation_level = 'schema' WHERE id = 't_acme';

-- Step 6: 清理(确认无误后)
DELETE FROM public.users WHERE tenant_id = 't_acme';
DELETE FROM public.projects WHERE tenant_id = 't_acme';

05 连接池与路由策略

多租户的隐藏杀手是连接数。每个数据库/Schema 需要独立的连接池,不加控制就会耗尽 PostgreSQL 的 max_connections

三种连接路由模式

🔗 按需连接 (On-Demand)

请求到来时才建立到租户 DB 的连接

节省连接 冷启动延迟

// 连接池管理器
class ConnectionPoolManager {
  private pools = new Map<string, Pool>()
  private maxPools = 50  // 最多 50 个活跃池

  async getPool(tenantId: string): Promise<Pool> {
    if (this.pools.has(tenantId)) {
      return this.pools.get(tenantId)!
    }

    if (this.pools.size >= this.maxPools) {
      // LRU 淘汰最久未用的池
      const [oldest] = this.pools.keys()
      await this.pools.get(oldest)!.end()
      this.pools.delete(oldest)
    }

    const pool = new Pool({
      host: getDbHost(tenantId),
      database: `tenant_${tenantId}`,
      max: 5,          // 每池最多 5 连接
      idleTimeoutMillis: 60000,
    })
    this.pools.set(tenantId, pool)
    return pool
  }
}
⚡ PgBouncer 事务路由

所有租户共享连接池,事务级别切换 search_path

连接最少 需 SET LOCAL

# PgBouncer 配置
[databases]
tenant_* = host=127.0.0.1 port=5432 dbname=%b

[pgbouncer]
pool_mode = transaction   # 事务级路由
max_client_conn = 1000
default_pool_size = 20
reserve_pool_size = 5

# 应用层:每个事务开始时设置
BEGIN;
SET LOCAL search_path = tenant_acme;
SELECT * FROM users;
COMMIT;
-- 事务结束,连接归还池

Schema-Level 连接路由中间件

// Express 中间件:Schema 路由
import { Pool } from 'pg'

const pool = new Pool({ max: 20 })

export async function tenantMiddleware(req, res, next) {
  const tenantId = req.tenant.id
  const client = await pool.connect()

  try {
    // 切换 search_path 到租户 Schema
    await client.query(`SET search_path TO tenant_${tenantId}, public`)
    req.db = client
    next()
  } catch (err) {
    client.release()
    next(err)
  }
}

// 路由处理器
app.get('/api/users', tenantMiddleware, async (req, res) => {
  const users = await req.db.query('SELECT * FROM users')
  req.db.release()  // 归还连接
  res.json(users.rows)
})

06 数据迁移:多租户的最大运维挑战

Schema 变更是多租户的噩梦——改一个表,需要跑 N 次。以下是三种策略:

策略 1:顺序迁移(简单但慢)

# 顺序执行迁移(适合 < 50 租户)
import asyncio
from sqlalchemy.ext.asyncio import create_async_engine

async def migrate_all_tenants(tenant_ids: list[str]):
    for tenant_id in tenant_ids:
        engine = create_async_engine(
            f"postgresql+asyncpg://user:pass@localhost/tenant_{tenant_id}"
        )
        async with engine.begin() as conn:
            await conn.run_sync(Base.metadata.create_all)
            await conn.execute(text("""
                ALTER TABLE users ADD COLUMN IF NOT EXISTS phone TEXT;
            """))
        await engine.dispose()
        print(f"Migrated tenant {tenant_id}")

asyncio.run(migrate_all_tenants(get_all_tenant_ids()))

策略 2:并行迁移(快但需锁控制)

# 并行迁移(适合 50-500 租户)
import asyncio
from concurrent.futures import ThreadPoolExecutor

async def migrate_tenant(tenant_id: str, migration_sql: str):
    engine = create_async_engine(
        f"postgresql+asyncpg://user:pass@localhost/tenant_{tenant_id}",
        pool_size=2
    )
    try:
        async with engine.begin() as conn:
            await conn.execute(text(migration_sql))
    finally:
        await engine.dispose()

async def migrate_all_parallel(tenant_ids: list[str], batch_size=10):
    migration_sql = "ALTER TABLE users ADD COLUMN IF NOT EXISTS phone TEXT;"

    for i in range(0, len(tenant_ids), batch_size):
        batch = tenant_ids[i:i + batch_size]
        await asyncio.gather(*[
            migrate_tenant(tid, migration_sql) for tid in batch
        ])
        print(f"Batch {i // batch_size + 1} done")

策略 3:Schema-Level 用模板 Schema

-- Schema-Level 迁移:使用模板 Schema
-- 1. 在模板上开发迁移
CREATE SCHEMA IF NOT EXISTS template_schema;

-- 2. 迁移模板
ALTER TABLE template_schema.users ADD COLUMN phone TEXT;

-- 3. 生成迁移脚本给所有租户
DO $$
DECLARE
  schema_name TEXT;
BEGIN
  FOR schema_name IN
    SELECT nspname FROM pg_namespace
    WHERE nspname LIKE 'tenant_%' AND nspname != 'template_schema'
  LOOP
    EXECUTE format('ALTER TABLE %I.users ADD COLUMN IF NOT EXISTS phone TEXT', schema_name);
    RAISE NOTICE 'Migrated %', schema_name;
  END LOOP;
END $$;
最佳实践:维护一个 schema_versions 表,记录每个租户的迁移版本号。这样可以在后台逐步迁移,不需要一次性跑完所有租户。
-- schema_versions 表
CREATE TABLE schema_versions (
  tenant_id UUID REFERENCES tenants(id),
  version INT NOT NULL DEFAULT 0,
  migrated_at TIMESTAMPTZ DEFAULT now(),
  PRIMARY KEY (tenant_id, version)
);

-- 查询哪些租户需要迁移
SELECT t.id, t.name
FROM tenants t
LEFT JOIN schema_versions sv ON t.id = sv.tenant_id AND sv.version = 42
WHERE sv.version IS NULL OR sv.version < 42;

07 安全与合规要点

数据泄露的常见路径

🚨 5 大数据泄露场景

防御:缓存 Key 设计

// Redis 缓存 Key 必须包含 tenant_id
class TenantCache {
  private prefix(tenantId: string, key: string): string {
    return `t:${tenantId}:${key}`
  }

  async get(tenantId: string, key: string) {
    return redis.get(this.prefix(tenantId, key))
  }

  async set(tenantId: string, key: string, value: any, ttl = 300) {
    return redis.set(this.prefix(tenantId, key), JSON.stringify(value), 'EX', ttl)
  }
}

// ❌ 危险!缓存 key 没有租户隔离
await redis.set('user:settings', data)

// ✅ 安全
await redis.set(`t:${tenantId}:user:settings`, data)

防御:搜索隔离

// Elasticsearch / Meilisearch 租户隔离
// 方案 1: 索引级隔离 (大租户)
const indexName = `tenant_${tenantId}_products`

// 方案 2: 文档级过滤 (小租户,更经济)
const results = await meilisearch.index('products').search(query, {
  filter: `tenant_id = ${tenantId}`
})

// 方案 3: Meilisearch 多租户 Token (最安全)
const searchToken = generateTenantToken({
  searchRules: { filter: `tenant_id = ${tenantId}` },
  apiKey: masterKey,
  expiresAt: new Date(Date.now() + 3600000)
})

合规:GDPR / HIPAA 要求

合规要求 推荐隔离模型 关键措施
GDPR 数据删除 Schema / Database 删除整个 Schema 或数据库比逐行删除更可靠
HIPAA Database-Level 物理隔离是 HIPAA 最安全的选择
SOC 2 Schema + RLS RLS 提供可审计的访问控制
FedRAMP Database-Level 美国联邦政府要求物理隔离
ISO 27001 任何 + RLS 需要证明访问控制有效,RLS 可满足

08 框架与工具选型

全栈多租户框架

🔶 django-tenants

Python/Django Schema-Level

最成熟的 Django 多租户方案。自动路由、自动迁移、域名映射。

✅ 自动 Schema 管理
✅ 域名 → 租户映射
✅ 与 Django 生态无缝集成

❌ 仅 Django
❌ Schema 级别为主

pip install django-tenants

TENANT_MODEL = "customers.Client"
TENANT_DOMAIN_MODEL = "customers.Domain"

# 自动创建/删除 Schema
# 自动路由查询

🔷 hasura/multi-tenant

GraphQL Row-Level

Hasura 的 RLS 集成。Header 传入 x-hasura-tenant-id,自动过滤。

✅ 零代码 API 隔离
✅ RLS 自动配置
✅ 实时订阅隔离

❌ 绑定 Hasura
❌ 复杂查询受限

# Hasura RLS 自动配置
# metadata.json
{
  "type": "set_permission",
  "args": {
    "table": "users",
    "role": "tenant_user",
    "permission": {
      "filter": {
        "tenant_id": "X-Hasura-Tenant-Id"
      }
    }
  }
}

🟢 Supabase RLS

PostgreSQL Row-Level

Supabase 原生 RLS + JWT 集成。最简单的多租户起步方案。

✅ RLS 开箱即用
✅ JWT → RLS 自动映射
✅ 客户端 SDK 直接用

❌ Schema-Level 需自建
❌ 连接池管理需注意

-- Supabase RLS 策略
CREATE POLICY "tenant_isolation" ON users
  USING (
    tenant_id = (auth.jwt() ->> 'tenant_id')::uuid
  );

-- 前端直接查询,自动过滤
const { data } = await supabase
  .from('users')
  .select('*')

🟣 Citadel (Ruby)

Ruby/Rails Schema-Level

Rails 生态的 Apartment 替代品。Schema 级隔离 + 迁移管理。

✅ Rails 原生集成
✅ 迁移自动分发
✅ Sidekiq 租户感知

❌ Ruby only
❌ Apartment 已归档,Citadel 较新

工具矩阵

工具 语言 隔离模型 成熟度 适用场景
django-tenants Python Schema ⭐⭐⭐⭐⭐ Django 全栈 SaaS
Supabase RLS Any Row + RLS ⭐⭐⭐⭐ 快速原型、小中型 SaaS
Hasura Any Row + RLS ⭐⭐⭐⭐ GraphQL API
Citadel/ro Ruby Schema ⭐⭐⭐ Rails SaaS
Prisma + RLS TypeScript Row + RLS ⭐⭐⭐ Next.js 全栈
自定义 (Hybrid) Any 混合 ⭐⭐ 大规模、合规需求

09 实战:从 0 开始的多租户架构

阶段 1:MVP(0-100 租户)

🎯 目标:最快上线

选择 Row-Level + PostgreSQL RLS,一张表搞定,成本最低。

-- MVP 多租户最小实现
-- 1. 租户表
CREATE TABLE tenants (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  name TEXT NOT NULL,
  subdomain TEXT UNIQUE NOT NULL,
  plan TEXT NOT NULL DEFAULT 'free',
  created_at TIMESTAMPTZ DEFAULT now()
);

-- 2. 所有业务表加 tenant_id
CREATE TABLE projects (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  tenant_id UUID NOT NULL REFERENCES tenants(id),
  name TEXT NOT NULL,
  created_at TIMESTAMPTZ DEFAULT now()
);

-- 3. 启用 RLS
ALTER TABLE projects ENABLE ROW LEVEL SECURITY;
CREATE POLICY tenant_isolation ON projects
  USING (tenant_id = current_setting('app.current_tenant')::UUID);

-- 4. 必须的索引
CREATE INDEX idx_projects_tenant ON projects(tenant_id);
CREATE INDEX idx_projects_tenant_created ON projects(tenant_id, created_at DESC);

阶段 2:增长(100-1000 租户)

📈 目标:性能 + 大客户

为付费客户引入 Schema-Level,免费用户继续 Row-Level。

// 租户路由器 v2
class TenantRouter {
  getConnection(tenant: Tenant) {
    switch (tenant.isolationLevel) {
      case 'row':
        // 共享连接池 + SET app.current_tenant
        return this.sharedPool.withSetTenant(tenant.id)
      case 'schema':
        // 共享连接池 + SET search_path
        return this.sharedPool.withSchema(`tenant_${tenant.id}`)
      case 'database':
        // 独立连接池
        return this.tenantPools.getOrCreate(tenant.id)
    }
  }
}

阶段 3:规模化(1000+ 租户)

🏢 目标:弹性 + 合规

引入连接池管理、Schema 模板、自动迁移、分库分表。

// 规模化架构
class ScaledTenantArchitecture {
  // 1. PgBouncer 事务级路由
  // 2. Schema 模板自动迁移
  // 3. 大租户独立实例
  // 4. 分区表 + RLS

  async handleRequest(req) {
    const tenant = await this.resolveTenant(req)

    // 连接池路由
    const pool = this.poolRouter.route(tenant)

    // 设置租户上下文
    const client = await pool.connect()
    await this.setContext(client, tenant)

    return { client, tenant }
  }

  // Schema 模板迁移
  async migrateTemplate(migration: Migration) {
    // 1. 在 template_schema 上测试
    await this.testMigration(migration)

    // 2. 后台逐步迁移所有 Schema
    await this.backgroundMigrate(migration)

    // 3. 监控迁移进度
    await this.monitorProgress(migration)
  }
}

10 多租户测试:防止数据泄露

租户隔离测试(最关键的测试)

// 多租户隔离测试
import { describe, it, expect, beforeEach } from 'vitest'

describe('Tenant Isolation', () => {
  let tenantA, tenantB

  beforeEach(async () => {
    // 创建两个租户
    tenantA = await createTenant({ name: 'Acme Corp' })
    tenantB = await createTenant({ name: 'Globex Inc' })

    // 各创建数据
    await createProject(tenantA, { name: 'Project A1' })
    await createProject(tenantA, { name: 'Project A2' })
    await createProject(tenantB, { name: 'Project B1' })
  })

  it('tenant A cannot see tenant B data', async () => {
    const projects = await listProjects(tenantA)
    expect(projects).toHaveLength(2)
    expect(projects.map(p => p.name)).not.toContain('Project B1')
  })

  it('tenant B cannot see tenant A data', async () => {
    const projects = await listProjects(tenantB)
    expect(projects).toHaveLength(1)
    expect(projects.map(p => p.name)).not.toContain('Project A1')
  })

  it('cannot access resource by ID cross-tenant', async () => {
    const projectA = await getProject(tenantA, projectA1Id)
    expect(projectA).toBeDefined()

    // 尝试用租户 B 的身份访问租户 A 的项目
    const project = await getProject(tenantB, projectA1Id)
    expect(project).toBeNull()  // 或抛出 403
  })

  it('search results are tenant-scoped', async () => {
    const results = await searchProjects(tenantA, 'Project')
    expect(results.every(r => r.tenantId === tenantA.id)).toBe(true)
  })

  it('cache keys include tenant_id', async () => {
    const cache = new TenantCache()
    await cache.set(tenantA.id, 'settings', { theme: 'dark' })
    const result = await cache.get(tenantB.id, 'settings')
    expect(result).toBeNull()
  })
})

自动化泄露检测

// CI 集成:自动检测租户泄露
async function leakDetectionTest() {
  const tenants = await getAllTenantIds()
  const failures = []

  for (const tenantId of tenants) {
    // 1. 列出所有 API 端点
    const endpoints = getApiEndpoints()

    for (const endpoint of endpoints) {
      const response = await fetch(endpoint, {
        headers: { 'Authorization': `Bearer ${getToken(tenantId)}` }
      })

      const data = await response.json()

      // 2. 递归检查所有字段值
      const otherTenantIds = tenants.filter(t => t !== tenantId)
      const found = findValues(data, otherTenantIds)

      if (found.length > 0) {
        failures.push({
          tenant: tenantId,
          endpoint,
          leaked: found,
        })
      }
    }
  }

  if (failures.length > 0) {
    throw new Error(`Tenant leak detected: ${JSON.stringify(failures)}`)
  }
}

11 与 Agent 架构的关联

多租户不仅是传统 SaaS 的问题——AI Agent 平台同样需要多租户隔离。

🤖 Agent 场景下的多租户

对话隔离:每个租户的 Agent 对话历史不能泄露给其他租户

工具权限隔离:租户 A 的 Agent 不能调用租户 B 的 API Key

知识库隔离:RAG 的向量数据按租户隔离

执行沙箱隔离:Agent 代码执行环境按租户隔离

Agent 层 隔离需求 推荐方案 参考项目
对话历史 严格隔离 Row-Level + RLS (session.tenant_id) OpenHands (Event 隔离)
API Key / 凭证 绝对隔离 加密存储 + 租户级密钥 Hermes (Credential Pool)
向量数据 (RAG) 搜索级隔离 tenant_id metadata 过滤 GPT Researcher (VectorStore)
代码执行 沙箱级隔离 租户级 Docker 容器 OpenHands (Docker Sandbox)
Agent 配置 租户定制 Schema-Level (允许不同配置) OpenClaw (Per-tenant config)
费用/配额 精确计费 Row-Level (usage_logs.tenant_id) OpenClaw (Token 计费)
关键洞察:Agent 平台的多租户比传统 SaaS 更复杂,因为需要同时隔离数据、凭证、执行环境、费用四个维度。传统 SaaS 通常只需隔离数据。具体参考 安全机制模块记忆系统模块

12 决策树:如何选择

🌳 多租户模型选择决策树
START
  │
  ├─ Q1: 是否有合规要求 (HIPAA/FedRAMP/金融监管)?
  │   └─ YES → Database-Level(物理隔离)
  │   └─ NO  → 继续
  │
  ├─ Q2: 是否需要租户级 Schema 定制?
  │   └─ YES → Schema-Level
  │   └─ NO  → 继续
  │
  ├─ Q3: 租户数量预期?
  │   ├─ < 100  → Schema-Level(运维可控,隔离好)
  │   ├─ 100-1000 → Row-Level + RLS(混合模型)
  │   └─ 1000+   → Row-Level + RLS + 分区
  │
  ├─ Q4: 是否需要跨租户查询(分析/管理后台)?
  │   └─ YES → Row-Level 优先(天然支持)
  │   └─ NO  → Schema/DB-Level 也可以
  │
  └─ Q5: 团队规模和运维能力?
      ├─ 1-3 人  → Row-Level + RLS(最简单)
      ├─ 4-10 人 → 可选 Schema-Level
      └─ 10+ 人  → 混合模型(按需分配)

经典 SaaS 产品的选择

产品 模型 原因
Notion Row-Level 数百万 workspace,需要跨租户搜索
Slack Schema-Level 团队隔离需求强,租户数量可控
Salesforce Database-Level (Enterprise) 企业客户合规要求 + 定制化
GitHub 混合模型 免费用户 Row-Level,Enterprise Database-Level
Vercel Row-Level 团队/项目数量大,需要灵活跨查询
Supabase Row-Level + RLS 自己就用 RLS,也推荐用户用 RLS

13 替代方案与前沿趋势

Serverless Database 的多租户

🟢 Turso (libSQL)

每个租户一个 SQLite 数据库文件,按需创建。

极致隔离 边缘部署

// Turso 多租户
import { createClient } from '@libsql/client'

function getTenantClient(tenantId: string) {
  return createClient({
    url: `libsql://tenant-${tenantId}-myapp.turso.io`,
    authToken: process.env.TURSO_TOKEN
  })
}

// 每个租户独立数据库
// Schema 迁移用 Turso Platform API 批量执行
🔵 Cloudflare D1

每个租户一个 D1 数据库,Worker 边缘执行。

全球分布 大小限制

// D1 多租户
export default {
  async fetch(request, env) {
    const tenantId = getTenantId(request)
    const db = env.DB  // 绑定到租户 D1

    // 根据租户 ID 选择 D1 数据库
    const d1Binding = `DB_${tenantId}`
    const db = env[d1Binding]

    return db.prepare('SELECT * FROM users')
      .all()
  }
}

向量数据库多租户

// 向量数据库多租户方案对比

// Pinecone: 命名空间级隔离
const index = pinecone.index('documents')
const ns = index.namespace(`tenant_${tenantId}`)
await ns.upsert([{ id: 'doc1', values: [0.1, 0.2], metadata: { tenantId } }])
const results = await ns.query({ vector: [0.1, 0.2], topK: 10 })

// Qdrant: Payload 过滤
await qdrant.search('documents', {
  vector: [0.1, 0.2],
  filter: { must: [{ key: 'tenant_id', match: { value: tenantId } }] }
})

// ChromaDB: 集合级隔离
const collection = chroma.getCollection({
  name: `tenant_${tenantId}_docs`
})

新兴趋势:Citus / CockroachDB

🔮 分布式数据库的多租户

Citus (分布式 PostgreSQL):按 tenant_id 自动分片,Row-Level 但有分布式性能。

-- Citus 分布式表
SELECT create_distributed_table('users', 'tenant_id');
SELECT create_distributed_table('projects', 'tenant_id');

-- 大租户自动独立分片
-- 小租户合并分片
-- 查询自动路由到正确节点

CockroachDB:多区域部署 + Zone 配置,按租户控制数据驻留地。

-- CockroachDB 区域配置
ALTER RANGE tenants CONFIGURE ZONE USING
  constraints = '[+region=us-east]'  -- 大客户数据留在美东

14 上线 Checklist

✅ 多租户上线前必检

15 关联阅读

🏛️ SaaS 架构模式

多租户是架构决策的一部分,结合微服务/Serverless 模式选择

🗄️ 数据库选型

不同数据库的多租户支持能力差异很大

🔐 认证授权

RBAC + 多租户 = 细粒度权限控制

🛡️ Agent 安全机制

Agent 平台的多租户安全需要额外隔离维度

🧠 Agent 记忆系统

对话历史和知识库的租户隔离实践

🔌 API 设计

多租户 API 的认证和路由设计