Nestjs框架: 基于Prisma的多租户功能集成和优化

概述

• 多租户系统指一个系统可以为多个客户（Tenant）提供服务，每个客户的数据相互隔离。常见做法是每个租户使用独立的数据库或共享数据库但通过 schema 或前缀隔离
• 这里，我们演示一下基于Prisma集成mysql和postgresql的多种数据库类型多数据库的多租户模式
• Prisma 内置数据库的驱动，它会根据 schema 文件中的 provider 在 generate 的时候
• 产生何种类型的 Prisma client, 需要根据不同的 provider 生成不同的 client
• 以便同一个 Prisma module 来访问不同类型数据库，这点和 typeorm 非常类似，只不过后者是外置驱动
• prisma 这里麻烦一点儿，参考 generate命令 就是根据 generator 和 data model 来产生对应的 Prisma Client
• 默认产生的位置在 node_module/@prisma/client中，我们现在要把不同数据库的client产生到指定的目录

产生不同的数据库client

• 在 generator 文档中，有一个字段 output ，默认情况下会产生于 node_modules/.prisma/client 中
  • 参考文档 prisma-schema-reference#fields-for-prisma-client-js-provider
• 我们将不同数据库客户端都产生于 src/prisma/clients 中

1 ) 配置 mysql

src/prisma/schema.prisma 下

generator client {provider = "prisma-client-js"output   = "./clients/mysql"
}datasource db {provider = "mysql"url      = env("DATABASE_URL")
}

注意，上面 .env 中的 DATABASE_URL 中配置的是有效的mysql链接, 如：

DATABASE_URL="mysql://root:123456_mysql@localhost:13306/testdb"

执行 $ pnpx prisma generate 此时就在 ./clients/mysql 目录中产生了

2 ) 配置 postgresql

src/prisma/schema.prisma 下

generator client {provider = "prisma-client-js"output   = "./clients/postgresql"
}datasource db {provider = "postgresql"url      = env("DATABASE_URL")
}

对应的 DATABASE_URL 修改为

DATABASE_URL="postgresql://pguser:123456_postgresql@localhost:15432/testdb"

• 执行 $ pnpx prisma generate 此时就在 ./clients/postgresql 目录中产生了

• 这里，如果重新 $ npm run start:dev 会有报错，这是临时产生的，先不要管它

  PrismaClientInitializationError: error: Error validating datasource `db`: the URL must start with the protocol `mysql://`.
  

• 注意：目前2个租户都是使用 一个 DATABASE_URL 这是为了生成客户端时候用的，之后，程序写到多租户的时候，就可以替换掉了

• 在 Prisma 多租户方案中，生成时使用的 URL 仅作为占位符，无需真实有效，运行时才会动态连接真实数据库 URL

prisma generate 生成原理

• 生成阶段（prisma generate）

  • Prisma 客户端生成过程仅解析 Schema 结构（表、字段、关系等），不验证数据库连接有效性
    占位 URL 只需满足格式规范（如 postgresql://dummy:dummy@localhost:5432/dummy），不要求实际可访问。

• 运行时阶段（动态连接）

  • 通过 PrismaClient 构造函数动态注入真实 URL：

    new PrismaClient({ datasources: { db: { url: realTenantUrl } } })
    

  • 每个租户独立实例化客户端，实现隔离连接

注意事项

• 占位 URL 的格式要求

  • 必须包含协议前缀（如 postgresql:// 或 mysql://），否则生成可能报错。
  • 示例：

    # 有效占位（符合 URL 格式）
    DATABASE_URL="mysql://dummy:dummy@dummy:3306/dummy"
    

• 生成与运行时的环境变量分离

  • 生成时：依赖 .env 中的占位 URL（或命令行临时指定）。
  • 运行时：从独立配置源（如数据库、配置中心）读取租户真实 URL，避免硬编码

方案优势与风险

优势	风险
一套 Schema 支持异构数据库（MySQL/PostgreSQL）	占位 URL 格式错误会导致生成失败
避免生成时依赖真实数据库环境	运行时需确保租户 URL 可访问且权限正确
动态扩展租户无需重新生成客户端	多客户端实例可能增加内存开销（需监控）

• 生产实践提示：

  • 通过 prisma validate 命令可提前验证 Schema 正确性，无需真实数据库
  • 运行时建议增加租户 URL 的格式校验逻辑（如正则匹配），防止配置错误导致连接失败。

• 总结

  • 占位 URL 的核心价值：
  • 解耦生成时与运行时的关注点
    • 生成时：仅关注 Schema 结构兼容性（如避免 MySQL/PostgreSQL 特有语法）。
    • 运行时：通过租户标识动态路由连接，实现多租户隔离。

• 有些业务场景，可能支持百级别租户混合使用 MySQL/PostgreSQL，如果需要进一步优化可结合连接池参数调优，或引入熔断机制（如 Hystrix）处理数据库故障

使用生成的客户端

• $ pnpm add prisma-mysql@link:./prisma/clients/mysql
• $ pnpm add prisma-postgresql@link:./prisma/clients/postgresql

关键输出信息

prisma-mysql <- prisma-client-e6c9963f1991a4598b0cb949accc2f3a50f445a316da9910f0b1811a3ad47379 6.12.0 <- prisma/clients/mysql

和

prisma-postgresql <- postgresql 0.0.0 <- prisma/clients/postgresql

可以看到 package.json 中多出来了 2 个依赖

"prisma-mysql": "link:prisma/clients/mysql",
"prisma-postgresql": "link:prisma/clients/postgresql",

Prisma 多租户集成初步规划

在 src/database/prisma/ 下

src/database/prisma/
|-- prisma-core.module.ts
|-- prisma-options.interface.ts
|-- prisma.constants.ts
|-- prisma.module.ts
|-- prisma.service.ts
|-- prisma.utils.ts

这里将 prisma module 拆分细化

1 ） prisma.constants.ts

export const PRISMA_CONNECTION_NAME = Symbol('PRISMA_CONNECTION_NAME');
export const PRISMA_MODULE_OPTIONS = Symbol('PRISMA_MODULE_OPTIONS');
export const PRISMA_CONNECTIONS = Symbol('PRISMA_CONNECTIONS');
// 这个配置模拟调接口/读数据库获取的
export const tenantMap = new Map([['1', 'T1'],['2', 'T2']
]);
export const defaultTenant = tenantMap.values().next().value; // 第一个

• 定义抽离出来的可用变量

2 ）prisma-options.interface.ts

import { Prisma } from '@prisma/client';
import { ModuleMetadata, Type } from '@nestjs/common';export interface PrismaModuleOptions {url?: string;datasourceUrl?: string;name?: string;options?: Prisma.PrismaClientOptions;retryAttempts?: number;retryDelay?: number;connectionFactory?: (connection: any, clientClass: any) => any;connectionErrorFactory?: (error: Prisma.PrismaClientKnownRequestError,) => Prisma.PrismaClientKnownRequestError;
}export interface PrismaOptionsFactory {createPrismaModuleOptions(): | Promise<PrismaModuleOptions> | PrismaModuleOptions;
}export type PrismaModuleFactoryOptions = Omit<PrismaModuleOptions, 'name'>; // 排除name属性，为了避免可能得冲突export interface PrismaModuleAsyncOptions extends Pick<ModuleMetadata, 'imports'> {name?: string;useExisting?: Type<PrismaOptionsFactory>;useClass?: Type<PrismaOptionsFactory>;useFactory?: (...args: any[]) => Promise<PrismaModuleFactoryOptions> | PrismaModuleFactoryOptions;inject?: any[];
}

• 定义后续可能用到的类型

3 ）prisma.utils.ts

import { retry, timer, throwError, catchError } from 'rxjs';export const PROTOCALREGEX = /^(.*?):\/\//;export function getDBType(url: string) {const matches = url.match(PROTOCALREGEX);const protocol = matches ? matches[1] : 'file';return protocol === 'file' ? 'sqlite' : protocol;
}export function handleRetry(retryAttempts: number, retryDelay: number) {return (source) =>source.pipe(retry({count: retryAttempts < 0 ? Infinity : retryAttempts, // 默认 count 是 正数，为了让 -1 这类值也能运行, 并无限重试delay: (error, retryCount) => {const attempts = retryAttempts < 0 ? Infinity : retryAttempts;if (retryCount <= attempts) {console.error(`Unable to connect to the database. Retrying (${retryCount})...`,error.stack,);return timer(retryDelay);} else {return throwError(() => new Error('Reached max retries'));}},}),catchError(error => {console.error(`Failed to connect to the database after retries ${retryAttempts} times`, error.stack || error,);return throwError(() => error);})) 
}

• 定义工具类，主要用于处理数据库连接的协议解析和重试逻辑
• 它结合了 RxJS（响应式编程库）的功能，实现了在数据库连接失败时的自动重试机制，并在重试失败后抛出错误

4 ） prisma.module.ts

import { Module, DynamicModule } from '@nestjs/common';
import { PrismaModuleOptions, PrismaModuleAsyncOptions } from './prisma-options.interface';
import { PrismaCoreModule } from './prisma-core.module';@Module({})
export class PrismaModule {static forRoot(options: PrismaModuleOptions): DynamicModule;static forRoot(url: string): DynamicModule;static forRoot(url: string, name: string): DynamicModule;static forRoot(arg: PrismaModuleOptions | string, ...args): DynamicModule {let _options: PrismaModuleOptions;if (args?.length) {_options = { url: arg, name: args[0]} as PrismaModuleOptions;} else if (typeof arg === 'string') {_options = { url: arg };} else {_options = arg;}return {module: PrismaModule,imports: [PrismaCoreModule.forRoot(_options)],}}static forRootAsync(options: PrismaModuleAsyncOptions) {return {module: PrismaModule,imports: [PrismaCoreModule.forRootAsync(options)],providers: [],exports: [],}}
}

• 它是一个模块封装，使用 @Module({}) 装饰器声明。

• 它本身是一个无内部逻辑的模块（@Module({}) 是空的），但通过静态方法 forRoot 和 forRootAsync 提供了动态模块的构建能力，用于在应用启动时配置 Prisma 数据库连接。

  @Module({})
  export class PrismaModule {static forRoot(...): DynamicModule;static forRootAsync(...): DynamicModule;
  }
  

• forRoot() 方法（同步配置）该方法用于同步配置 Prisma 模块的数据库连接。它有多个函数签名重载，支持灵活的参数传递方式：

  • forRoot(options: PrismaModuleOptions)
  • forRoot(url: string)
  • forRoot(url: string, name: string)
  • 支持多种传参方式：
    • 全部参数传入一个对象（推荐）
    • 只传入数据库连接 URL
    • 同时传入 URL 和一个命名标识（name）
  • 内部将参数统一转换为 PrismaModuleOptions 类型，然后交给 PrismaCoreModule.forRoot() 处理
  • 返回一个动态模块，注册 PrismaModule 并导入 PrismaCoreModule，实现核心功能注入

• forRootAsync() 方法（异步配置）用于异步配置 Prisma 模块，通常用于需要依赖注入（DI）或异步加载配置（如从配置文件或服务中获取数据库连接信息）的场景

  static forRootAsync(options: PrismaModuleAsyncOptions) {return {module: PrismaModule,imports: [PrismaCoreModule.forRootAsync(options)],providers: [],exports: [],}
  }
  

• 通常包含以下字段（由 NestJS 惯例和 Prisma 模块设计决定）：

  • useFactory：用于异步生成配置的工厂函数
  • inject：注入的依赖项数组
  • imports：其他模块导入（用于获取依赖）
  • extraProviders：额外提供者（如服务类）

• 依赖模块：PrismaCoreModule

  • PrismaCoreModule 是实际处理 Prisma 初始化逻辑的模块。它提供了两个静态方法：
    • forRoot()：同步初始化 Prisma 客户端
    • forRootAsync()：异步初始化 Prisma 客户端
  • 这个模块才是真正负责创建 Prisma 实例、注册服务、管理连接池等核心任务的地方

• 接口依赖类型说明

  • PrismaModuleOptions：配置数据库连接的接口，通常包括：
    • url: 数据库连接字符串
    • name: 模块实例的标识名（用于多数据库连接场景）
  • PrismaModuleAsyncOptions：异步初始化配置接口，支持 DI 和异步加载

5 ）prisma-core.module

import { Global, Module, OnApplicationShutdown, Provider, Type } from '@nestjs/common';
import { PrismaModuleOptions, PrismaModuleAsyncOptions, PrismaOptionsFactory } from './prisma-options.interface';
import { PrismaClient as MySQLClient } from 'prisma-mysql';
import { PrismaClient as PgClient } from 'prisma-postgresql';
import { getDBType, handleRetry } from './prisma.utils';
import { PRISMA_CONNECTION_NAME, PRISMA_MODULE_OPTIONS, PRISMA_CONNECTIONS } from './prisma.constants';
import { lastValueFrom, defer, catchError } from 'rxjs';
import { DynamicModule } from '@nestjs/common';@Global()
@Module({})
export class PrismaCoreModule implements OnApplicationShutdown {private static connections: Record<string, any> = {};/*** 应用程序关闭时触发，用于优雅关闭 Prisma 客户端连接*/async onApplicationShutdown(signal?: string): Promise<void> {if (PrismaCoreModule.connections && Object.keys(PrismaCoreModule.connections.length > 0)) {for (const key of Object.keys(PrismaCoreModule.connections)) {const connection = PrismaCoreModule.connections[key];if (connection && typeof connection.$disconnect === 'function') {connection.$disconnect();}}}}static forRoot(_options: PrismaModuleOptions): any {const { url, options = {}, name,retryAttempts = 10,retryDelay = 3000,connectionFactory,connectionErrorFactory,} = _options;let newOptions = {datasourceUrl: url};if (!Object.keys(options).length) {newOptions = { ...newOptions, ...options };}let _prismaClient;const dbType = getDBType(url!);if (dbType === 'mysql') {_prismaClient = MySQLClient;} else if (dbType === 'postgresql') {_prismaClient = PgClient;} else {throw new Error(`Unsupported database type: ${dbType}`);}const providerName = name || PRISMA_CONNECTION_NAME;const prismaConnectionErrorFactory = connectionErrorFactory || ((err) => err);const prismaConnectionFactory = connectionFactory ||(async (clientOptions) => await new _prismaClient(clientOptions));const prismaClientProvider: Provider = {provide: providerName,useFactory: async () => {if (this.connections[url!]) {return this.connections[url!];}// 加入错误重试const client = await prismaConnectionFactory(newOptions, name!);this.connections[url!] = client;return lastValueFrom(defer(async () => await client.$connect()).pipe(handleRetry(retryAttempts, retryDelay),catchError(err => { throw prismaConnectionErrorFactory(err) }))).then(() => client);},};const connectionsProvider = {provide: PRISMA_CONNECTIONS,useValue: this.connections};return {module: PrismaCoreModule,providers: [prismaClientProvider, connectionsProvider],exports: [prismaClientProvider, connectionsProvider],};}static forRootAsync(_options: PrismaModuleAsyncOptions): DynamicModule {const providerName = _options.name || PRISMA_CONNECTION_NAME;const prismaClientProvider: Provider = {provide: providerName,useFactory: (prismaModuleOptions: PrismaModuleOptions) => {const {url, options = {},retryAttempts = 10,retryDelay = 3000,connectionFactory,connectionErrorFactory,} = prismaModuleOptions;let newOptions = {datasourceUrl: url};if (!Object.keys(options).length) {newOptions = { ...newOptions, ...options };}let _prismaClient;const dbType = getDBType(url!);if (dbType === 'mysql') {_prismaClient = MySQLClient;} else if (dbType === 'postgresql') {_prismaClient = PgClient;} else {throw new Error(`Unsupported database type: ${dbType}`);}const prismaConnectionErrorFactory = connectionErrorFactory || ((err) => err);const prismaConnectionFactory = connectionFactory ||(async (clientOptions) => await new _prismaClient(clientOptions));return lastValueFrom(defer(async () => {const url = newOptions.datasourceUrl;if (this.connections[url!]) {return this.connections[url!];}const client = await prismaConnectionFactory(newOptions,_prismaClient);this.connections[url!] = client;return client;}).pipe(handleRetry(retryAttempts, retryDelay),catchError(err => { throw prismaConnectionErrorFactory(err) })));},inject: [PRISMA_MODULE_OPTIONS]}const asyncProviders = this.createAsyncProviders(_options);const connectionsProvider = {provide: PRISMA_CONNECTIONS,useValue: this.connections};return {module: PrismaCoreModule,providers: [...asyncProviders, prismaClientProvider, connectionsProvider],exports: [prismaClientProvider, connectionsProvider],}}private static createAsyncProviders(options: PrismaModuleAsyncOptions) {if (options.useExisting || options.useFactory) {return [this.createAsyncOptionsProviders(options)]}const useClass = options.useClass as Type<PrismaOptionsFactory>return [this.createAsyncOptionsProviders(options),{provide: useClass,useClass,}]}// 创建 PRISMA_MODULE_OPTIONS的Providerprivate static createAsyncOptionsProviders(options: PrismaModuleAsyncOptions):Provider {if (options.useFactory) {return {provide: PRISMA_MODULE_OPTIONS,useFactory: options.useFactory,inject: options.inject || [],}}const inject = [(options.useClass || options.useExisting) as Type<PrismaOptionsFactory>,]return {provide: PRISMA_MODULE_OPTIONS,inject,useFactory: async (optionsFactory: PrismaOptionsFactory) => optionsFactory.createPrismaModuleOptions(),}}
}

• 用于封装并统一管理 Prisma Client 的连接逻辑，适配多种数据库类型（如 MySQL、PostgreSQL），并支持同步和异步配置

• 该 PrismaCoreModule 是 NestJS 中的 全局模块（@Global()），用于集中管理 Prisma 客户端的连接、配置、重试、错误处理、连接池管理等逻辑，确保整个应用中对数据库的访问是统一、高效和稳定的

• 静态连接池管理

  private static connections: Record<string, any> = {};
  

  • 作用：作为模块内的静态属性，用于缓存已实例化的 Prisma 客户端实例，避免重复连接，提升性能
  • 生命周期：模块加载时初始化，应用关闭时通过 onApplicationShutdown 优雅销毁

• 应用关闭时自动断开连接

  async onApplicationShutdown(signal?: string): Promise<void>
  

  • 功能：实现 OnApplicationShutdown 生命周期接口，确保应用关闭前释放所有数据库连接
  • 细节：遍历 connections 对象，调用每个连接的 $disconnect() 方法
  • 意义：防止连接泄漏，提升系统健壮性

• 同步注册：forRoot()

  static forRoot(_options: PrismaModuleOptions): DynamicModule
  

  • 功能：用于同步配置 Prisma 模块的入口方法
  • 参数解析：
    • url: 数据库连接 URL
    • options: 额外 Prisma 客户端配置
    • retryAttempts, retryDelay: 重试机制配置
    • connectionFactory: 自定义连接创建逻辑
    • connectionErrorFactory: 自定义错误处理逻辑

• 核心逻辑：

  • 根据 URL 判断数据库类型（MySQL / PostgreSQL）
  • 实例化对应的 Prisma Client
  • 使用 defer + handleRetry + catchError 实现连接失败的重试逻辑
  • 提供可注入的 Prisma 客户端服务（providerName）供其他模块使用

• 异步注册：forRootAsync()

  static forRootAsync(_options: PrismaModuleAsyncOptions): DynamicModule
  

  • 功能：用于异步方式配置 Prisma 模块，更适用于需要依赖注入的场景
  • 支持方式：
    • useClass: 提供一个实现 PrismaOptionsFactory 接口的类
    • useFactory: 自定义异步工厂函数
    • useExisting: 使用一个已存在的服务
  • 优势：支持异步初始化，如从数据库或远程服务加载配置

• 创建异步依赖项：createAsyncProviders() 与 createAsyncOptionsProviders()

  • 作用：构建异步配置所需的依赖注入提供者（Providers）
  • createAsyncOptionsProviders()：
    • 若使用 useFactory，则直接注册该工厂
    • 若使用类（useClass 或 useExisting），则注入该类并调用 createPrismaModuleOptions() 方法获取配置

• 关键技术点与设计思想

  • 1. 支持多数据库类型

       const dbType = getDBType(url);
       

       • 实现方式：通过 URL 判断数据库类型（mysql、postgresql）
       • 扩展性：未来可轻松支持更多数据库类型（如 SQLite、MongoDB 等）
       • 容错机制：不支持的数据库类型抛出错误
  • 2. 重试机制

       defer(...).pipe(handleRetry(...), catchError(...))
       

       • 实现方式：通过 RxJS 的 defer() 和自定义 handleRetry() 操作符实现连接失败重试。
       • 配置项：retryAttempts 和 retryDelay 可由用户配置。
       • 优势：提高连接的健壮性，应对网络波动或数据库短暂不可用的场景
• 3. 自定义连接与错误处理
  • connectionFactory：允许用户自定义 Prisma Client 的创建逻辑
  • connectionErrorFactory：允许用户自定义连接失败的错误响应逻辑
  • 灵活性：极大增强了模块的可定制性，满足不同业务场景需求
• 4. 模块设计模式：单例 + 依赖注入
  • 模块使用了 NestJS 的 DI（依赖注入）机制，支持同步与异步配置
  • 所有 Prisma 客户端以单例形式缓存，提升性能，避免资源浪费
  • 提供 @Inject(providerName) 方式注入 Prisma 客户端到服务中

6 ) prisma.service.ts

import { Injectable, Inject, OnModuleInit } from '@nestjs/common';
import { PrismaOptionsFactory, PrismaModuleOptions } from './prisma-options.interface';
import { REQUEST } from '@nestjs/core';
import { Request } from 'express';
import { tenantMap, defaultTenant } from './prisma.constants';
import { ConfigService } from '@nestjs/config';@Injectable()
export class PrismaService implements OnModuleInit, PrismaOptionsFactory {constructor(@Inject(REQUEST) private request: Request,private configService: ConfigService) {}createPrismaModuleOptions(): PrismaModuleOptions | Promise<PrismaModuleOptions> {const headers = this.request.headers;const tenantId = headers['x-tenant-id'] as string || 'default';if (tenantId && !tenantMap.has(tenantId)) {throw new Error('invalid tenantId');}const t_prefix = !tenantId ? defaultTenant : tenantMap.get(tenantId);const db_url = this.configService.get<string>(`${t_prefix}_DATABASE_URL`);return { url: db_url };}async onModuleInit() {}
}

• 从请求头中获取 x-tenant-id，若不存在则使用默认租户（如 ‘default’）
• 校验该租户是否存在于 tenantMap（一个 Map 结构，用于映射租户 ID 与数据库标识）
• 获取对应租户的环境变量前缀（如 TENANT1_DATABASE_URL）
• 返回 Prisma 模块所需的连接 URL

应用封装服务

上面核心模块设计好了，之后，就可以开始应用了，上面提供了 forRoot 的方式，也提供了 forRootAsync 动态配置和注入的方式，现在我们使用后者的方式因为更方便

1 ）app.module.ts

import { Module } from '@nestjs/common';
import { AppController } from './app.controller';
import { ConfigModule, ConfigService } from '@nestjs/config';
import { PrismaModule } from './database/prisma/prisma.module';
import { PrismaService } from './database/prisma/prisma.service';const connections = new Map<string, DataSource>();@Module({imports: [// 1. 下面这个后续可以封装一个新的模块，来匹配 .env 和 其他配置ConfigModule.forRoot({  // 配置环境变量模块envFilePath: '.env', // 指定环境变量文件路径isGlobal: true, // 全局可用}),// 2. 集成 PrismaPrismaModule.forRootAsync({name: 'prismaClient',useClass: PrismaService})],controllers: [AppController]
})export class AppModule {}

这里用 forRootAsync 来演示, forRoot的形式在此不演示了

2 ) app.controller.ts

import { Controller, Get, Inject } from '@nestjs/common';
import { PrismaClient } from '@prisma/client';@Controller()
export class AppController {constructor(@Inject('prismaClient') private readonly prismaClientService: PrismaClient,) {}@Get('/multi-prisma')async getMultiPrisma(): Promise<any> {const rs = await this.prismaClientService.user.findMany({});return rs;}
}

测试结果

1 ) 测试1

请求

	curl --request GET \--url http://localhost:3000/multi-prisma \--header 'x-tenant-id: 1'

响应

[{"id": 1,"username": "mysql","password": "123456"}
]

2 ）测试2

请求

curl --request GET \--url http://localhost:3000/multi-prisma \--header 'x-tenant-id: 2'

响应

[{"id": 1,"username": "postgresql","password": "123456"}
]