多租户 (Multi-Tenancy) 是 SaaS 的核心架构决策:一套应用代码服务多个客户 (租户/Tenant),但每个租户的数据互相不可见。
想想公寓楼——共享地基和外墙(应用层),但每户有自己的锁和房间(数据层)。区别在于:共享多少、隔离多少。
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 } });
}
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),
])),
]
多租户的核心问题是数据隔离粒度,共有三种主流模型:
每个租户一个完整数据库。物理隔离最强,但运维成本最高。
• 零数据泄露风险(物理隔离)
• 可独立备份/恢复
• 可按租户定制 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]
同一个 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 的数据
所有租户数据在同一张表,通过 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+ 租户 |
| 成本 | $$$ | $$ | $ |
PostgreSQL 的 Row-Level Security (RLS) 是让 Row-Level 模型安全的关键——它在数据库层强制执行租户隔离,即使应用代码有 bug 也不会泄露数据。
-- 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 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-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() → 只返回当前租户
(tenant_id, created_at) 比 (tenant_id) + (created_at) 更快SET LOCAL,会话模式用 SET-- 分区表 + 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);
没有一种模型适合所有场景。真实 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
-- 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';
多租户的隐藏杀手是连接数。每个数据库/Schema 需要独立的连接池,不加控制就会耗尽 PostgreSQL 的 max_connections。
请求到来时才建立到租户 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
}
}
所有租户共享连接池,事务级别切换 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;
-- 事务结束,连接归还池
// 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)
})
Schema 变更是多租户的噩梦——改一个表,需要跑 N 次。以下是三种策略:
# 顺序执行迁移(适合 < 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()))
# 并行迁移(适合 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")
-- 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;
// 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 数据删除 | Schema / Database | 删除整个 Schema 或数据库比逐行删除更可靠 |
| HIPAA | Database-Level | 物理隔离是 HIPAA 最安全的选择 |
| SOC 2 | Schema + RLS | RLS 提供可审计的访问控制 |
| FedRAMP | Database-Level | 美国联邦政府要求物理隔离 |
| ISO 27001 | 任何 + RLS | 需要证明访问控制有效,RLS 可满足 |
Python/Django Schema-Level
最成熟的 Django 多租户方案。自动路由、自动迁移、域名映射。
✅ 自动 Schema 管理
✅ 域名 → 租户映射
✅ 与 Django 生态无缝集成
❌ 仅 Django
❌ Schema 级别为主
pip install django-tenants
TENANT_MODEL = "customers.Client"
TENANT_DOMAIN_MODEL = "customers.Domain"
# 自动创建/删除 Schema
# 自动路由查询
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"
}
}
}
}
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('*')
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 | 混合 | ⭐⭐ | 大规模、合规需求 |
选择 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);
为付费客户引入 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)
}
}
}
引入连接池管理、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)
}
}
// 多租户隔离测试
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)}`)
}
}
多租户不仅是传统 SaaS 的问题——AI 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 计费) |
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+ 人 → 混合模型(按需分配)
| 产品 | 模型 | 原因 |
|---|---|---|
| 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 |
每个租户一个 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 批量执行
每个租户一个 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 (分布式 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]' -- 大客户数据留在美东
tenant_id 列且 NOT NULLcurrent_setting 而非应用变量tenant_id 过滤(即使有 RLS 也带,双层防护)tenant_id 作为前缀列tenant_idtenant_idtenant_id