Nestjs框架: 基于TypeORM的多租户功能集成

概述

• 我们现在要集成多租户的环境，考虑到使用2个mysql和1个postgresql的数据库
• 现在要集成到 nestjs 的项目中来，考虑到配置通用和性能的处理

配置docker-compose启动服务

docker-compose.multi.yaml

services:mysql:image: mysql:8container_name: mysql_apprestart: alwaysports:- "13306:3306"environment:- MYSQL_ROOT_PASSWORD=123456_mysqlvolumes:# - ./mysql/conf.d:/etc/mysql/conf.d # 默认加载这里的配置- ./docker-dbconfig/mysql/data:/var/lib/mysql- ./docker-dbconfig/mysql/logs:/var/log/mysqlnetworks:- light_networkmysql2:image: mysql:8container_name: mysql_app2restart: alwaysports:- "13307:3306"environment:- MYSQL_ROOT_PASSWORD=123456_mysqlvolumes:# - ./mysql/conf.d:/etc/mysql/conf.d # 默认加载这里的配置- ./docker-dbconfig/mysql/data2:/var/lib/mysql- ./docker-dbconfig/mysql/logs2:/var/log/mysqlnetworks:- light_networkpostgresql:image: postgres:16restart: alwaysenvironment:POSTGRES_PASSWORD: 123456_postgresqlPOSTGRES_DB: testdbPOSTGRES_USER: pguser # 注意，不能用 rootports:- 15432:5432volumes:- ./docker-dbconfig/postgres/data:/var/lib/postgresql/data- ./docker-dbconfig/postgres/logs:/var/log/postgresqlnetworks:- light_networkadminer:image: adminer:5.3.0container_name: adminer_apprestart: alwaysports:- 18080:8080networks:- light_networknetworks:light_network:external: true

执行 $ docker ps

CONTAINER ID   IMAGE                  COMMAND                   CREATED          STATUS         PORTS                                NAMES
6c8879abf8bd   mysql:8                "docker-entrypoint.s…"   10 seconds ago   Up 9 seconds   33060/tcp, 0.0.0.0:13306->3306/tcp   mysql_app
e8dfd71eab60   mysql:8                "docker-entrypoint.s…"   10 seconds ago   Up 9 seconds   33060/tcp, 0.0.0.0:13307->3306/tcp   mysql_app2
5737bcb3c24e   postgres:16            "docker-entrypoint.s…"   10 seconds ago   Up 9 seconds   0.0.0.0:15432->5432/tcp              hello-nest-postgresql-1
af724225edea   adminer:5.3.0          "entrypoint.sh docke…"   10 seconds ago   Up 9 seconds   0.0.0.0:18080->8080/tcp              adminer_app

• 这里能看到4个服务，其中3个数据库mysql, mysql2, postgresql 分别为: 13306, 13307, 15432

1 ) 连接 mysql

• 访问: http://localhost:18080

• 输入

  • 服务器: mysql
  • 用户名: root
  • 密码: 123456_mysql

• 连接上之后，创建数据库 testdb

2 ）连接另一台 mysql2

• 输入

  • 服务器: mysql2
  • 用户名: root
  • 密码: 123456_mysql

• 创建数据库 testdb

3 ) 连接 postgresql

• 系统选择：PostgreSQL

• 输入

  • 服务器: postgresql
  • 用户名: pguser
    • 注意这里不能用 root
  • 密码: 123456_postgresql

• 创建数据库 testdb

配置 TypeORM CLI 命令

• TypeORM 也是支持配置文件的，参考官网 ormconfig
• typeorm cli 和之前使用的 eslint 一样，也是支持配置文件的
• 配置之后，就可以使用 typeorm 提供的 init 方法来初始化数据库了

1 ) 安装依赖和配置

• $ pnpm add dotenv

• 配置 ormconfig.ts

  import { DataSource, DataSourceOptions } from 'typeorm';
  import * as dotenv from 'dotenv';
  import * as fs from 'fs';
  import { TypeOrmModuleOptions } from '@nestjs/typeorm';export function getEnv(env: string): Record<string, unknown> | undefined {if (fs.existsSync(env)) {return dotenv.parse(fs.readFileSync(env));}
  }export function buildConnectionOptions(){const defaultConfig = getEnv('.env');const envConfig = getEnv(`.env.${process.env.NODE_ENV || 'development'}`);const config = {...defaultConfig, ...envConfig };return {type: config['DB_TYPE'],host: config['DB_HOST'],port: config['DB_PORT'],username: config['DB_USERNAME'],password: config ['DB_PASSWORD'],database: config['DB_DATABASE'],entities: [__dirname + '/**/*.entity{.ts,.js}'],synchronize: Boolean(config['DB_SYNC']),autoLoadEntities: Boolean(config['DB_AUTOLOAD']),} as TypeOrmModuleOptions;
  }export default new DataSource({...buildConnectionOptions(),
  } as DataSourceOptions);
  

2 ) 配置脚本并同步

• 配置 package.json 中的 scripts, 添加如下

  "typeorm": "typeorm-ts-node-commonjs -d ormconfig.ts",
  "typeorm:sync": "npm run typeorm schema:sync"
  

• 参考官网 using-cli

• 目前有2个mysql的服务和一个postgresql 的服务

  • mysql 的 .env 配置

    DB_TYPE=mysql
    DB_HOST=localhost
    DB_PORT=13306
    DB_USERNAME=root
    DB_PASSWORD=123456_mysql
    DB_DATABASE=testdb
    DB_AUTOLOAD=true
    DB_SYNC=true
    

  • mysql2 的 .env 配置

    DB_TYPE=mysql
    DB_HOST=localhost
    DB_PORT=13307
    DB_USERNAME=root
    DB_PASSWORD=123456_mysql
    DB_DATABASE=testdb
    DB_AUTOLOAD=true
    DB_SYNC=true
    

  • postgresql 的 .env 配置

    DB_TYPE=postgres
    DB_HOST=localhost
    DB_PORT=15432
    DB_USERNAME=pguser
    DB_PASSWORD=123456_postgresql
    DB_DATABASE=testdb
    DB_AUTOLOAD=true
    DB_SYNC=true
    

• 对于 postgres 数据库，需要安装依赖 $ pnpm add pg

• 切换上面的不同 .env 配置，分别执行 $ pnpm run typeorm:sync

• mysql 的输出

  query: SELECT VERSION() AS `version`
  query: START TRANSACTION
  query: SELECT DATABASE() AS `db_name`
  query: SELECT `TABLE_SCHEMA`, `TABLE_NAME`, `TABLE_COMMENT` FROM `INFORMATION_SCHEMA`.`TABLES` WHERE `TABLE_SCHEMA` = 'testdb' AND `TABLE_NAME` = 'user' UNION SELECT `TABLE_SCHEMA`, `TABLE_NAME`, `TABLE_COMMENT` FROM `INFORMATION_SCHEMA`.`TABLES` WHERE `TABLE_SCHEMA` = 'testdb' AND `TABLE_NAME` = 'user'
  query:SELECT*FROM`INFORMATION_SCHEMA`.`COLUMNS`WHERE`TABLE_SCHEMA` = 'testdb'AND`TABLE_NAME` = 'user'query: SELECT * FROM (SELECT*FROM `INFORMATION_SCHEMA`.`KEY_COLUMN_USAGE` `kcu`WHERE`kcu`.`TABLE_SCHEMA` = 'testdb'AND`kcu`.`TABLE_NAME` = 'user') `kcu` WHERE `CONSTRAINT_NAME` = 'PRIMARY'
  query:SELECT`SCHEMA_NAME`,`DEFAULT_CHARACTER_SET_NAME` as `CHARSET`,`DEFAULT_COLLATION_NAME` AS `COLLATION`FROM `INFORMATION_SCHEMA`.`SCHEMATA`query:SELECT`s`.*FROM (SELECT*FROM `INFORMATION_SCHEMA`.`STATISTICS`WHERE`TABLE_SCHEMA` = 'testdb'AND`TABLE_NAME` = 'user') `s`LEFT JOIN (SELECT*FROM `INFORMATION_SCHEMA`.`REFERENTIAL_CONSTRAINTS`WHERE`CONSTRAINT_SCHEMA` = 'testdb'AND`TABLE_NAME` = 'user') `rc`ON`s`.`INDEX_NAME` = `rc`.`CONSTRAINT_NAME`AND`s`.`TABLE_SCHEMA` = `rc`.`CONSTRAINT_SCHEMA`WHERE`s`.`INDEX_NAME` != 'PRIMARY'AND`rc`.`CONSTRAINT_NAME` IS NULLquery:SELECT`kcu`.`TABLE_SCHEMA`,`kcu`.`TABLE_NAME`,`kcu`.`CONSTRAINT_NAME`,`kcu`.`COLUMN_NAME`,`kcu`.`REFERENCED_TABLE_SCHEMA`,`kcu`.`REFERENCED_TABLE_NAME`,`kcu`.`REFERENCED_COLUMN_NAME`,`rc`.`DELETE_RULE` `ON_DELETE`,`rc`.`UPDATE_RULE` `ON_UPDATE`FROM (SELECT*FROM `INFORMATION_SCHEMA`.`KEY_COLUMN_USAGE` `kcu`WHERE`kcu`.`TABLE_SCHEMA` = 'testdb'AND`kcu`.`TABLE_NAME` = 'user') `kcu`INNER JOIN (SELECT*FROM `INFORMATION_SCHEMA`.`REFERENTIAL_CONSTRAINTS`WHERE`CONSTRAINT_SCHEMA` = 'testdb'AND`TABLE_NAME` = 'user') `rc`ON`rc`.`CONSTRAINT_SCHEMA` = `kcu`.`CONSTRAINT_SCHEMA`AND`rc`.`TABLE_NAME` = `kcu`.`TABLE_NAME`AND`rc`.`CONSTRAINT_NAME` = `kcu`.`CONSTRAINT_NAME`query: SELECT * FROM `INFORMATION_SCHEMA`.`COLUMNS` WHERE `TABLE_SCHEMA` = 'testdb' AND `TABLE_NAME` = 'typeorm_metadata'
  query: COMMIT
  Schema synchronization finished successfully.
  

  • 上面执行了很多 query 查询，它利用了 ormconfig.ts 中的配置信息同步到服务器侧上去
  • typeorm 就是使用上面的配置，连接到远程数据库，并且把所有表信息都同步到数据库中

• mysql2 的输出

  query: SELECT VERSION() AS `version`
  query: START TRANSACTION
  query: SELECT DATABASE() AS `db_name`
  query: SELECT `TABLE_SCHEMA`, `TABLE_NAME`, `TABLE_COMMENT` FROM `INFORMATION_SCHEMA`.`TABLES` WHERE `TABLE_SCHEMA` = 'testdb' AND `TABLE_NAME` = 'user' UNION SELECT `TABLE_SCHEMA`, `TABLE_NAME`, `TABLE_COMMENT` FROM `INFORMATION_SCHEMA`.`TABLES` WHERE `TABLE_SCHEMA` = 'testdb' AND `TABLE_NAME` = 'user'
  query: SELECT * FROM `INFORMATION_SCHEMA`.`COLUMNS` WHERE `TABLE_SCHEMA` = 'testdb' AND `TABLE_NAME` = 'typeorm_metadata'
  creating a new table: testdb.user
  query: CREATE TABLE `user` (`id` int NOT NULL AUTO_INCREMENT, `username` varchar(255) NOT NULL, `password` varchar(255) NOT NULL, PRIMARY KEY (`id`)) ENGINE=InnoDB
  query: COMMIT
  Schema synchronization finished successfully.
  

• postgresql的输出

  query: SELECT version()
  query: SELECT * FROM current_schema()
  query: START TRANSACTION
  query: SELECT * FROM current_schema()
  query: SELECT * FROM current_database()
  query: SELECT "table_schema", "table_name", obj_description(('"' || "table_schema" || '"."' || "table_name" || '"')::regclass, 'pg_class') AS table_comment FROM "information_schema"."tables" WHERE ("table_schema" = 'public' AND "table_name" = 'user') OR ("table_schema" = 'public' AND "table_name" = 'user')
  query: SELECT * FROM "information_schema"."tables" WHERE "table_schema" = 'public' AND "table_name" = 'typeorm_metadata'
  creating a new table: public.user
  query: CREATE TABLE "user" ("id" SERIAL NOT NULL, "username" character varying NOT NULL, "password" character varying NOT NULL, CONSTRAINT "PK_cace4a159ff9f2512dd42373760" PRIMARY KEY ("id"))
  query: COMMIT
  Schema synchronization finished successfully.
  

• 注意，可见上面这么搞，还是有些麻烦的，后续可以把不同数据库配置都写上

• .env 文件区分不同的库，写一个前缀作为区分，之后 ormconfig.ts 中的配置也基于命令参数做动态处理

3 ) 刷新UI管理界面

• 刷新: http://localhost:18080/?server=mysql&username=root&db=testdb
• 发现同步过来了
• 关于后续的数据库迁移相关的文档 using-cli.html#run-migrations
• 里面还有如何恢复迁移, 后续有需求，继续看文档

4 ）现在我们要分别在三个数据库中创建数据了

• mysql 的 testdb 中手动创建一条数据
  • {username: 'mysql', password: '123456'}
• mysql2 的 testdb 中手动创建一条数据
  • {username: 'mysql2', password: '123456'}
• postgresql 的 testdb 中手动创建一条数据
  • {username: 'postgresql', password: '123456'}

关于useFactory和dataSourceFactory

• 官方文档里 关于 database#custom-datasource-factory

• 这里 useFactory 和 dataSourceFactory 二者有什么关系呢？ 参考下面的官方代码

  TypeOrmModule.forRootAsync({imports: [ConfigModule],inject: [ConfigService],// Use useFactory, useClass, or useExisting// to configure the DataSourceOptions.useFactory: (configService: ConfigService) => ({type: 'mysql',host: configService.get('HOST'),port: +configService.get('PORT'),username: configService.get('USERNAME'),password: configService.get('PASSWORD'),database: configService.get('DATABASE'),entities: [],synchronize: true,}),// dataSource receives the configured DataSourceOptions// and returns a Promise<DataSource>.dataSourceFactory: async (options) => {const dataSource = await new DataSource(options).initialize();return dataSource;},
  });
  

• useFactory 用于产生 DataSource 的选项（options）

• 而 dataSourceFactory 则实际用于创建并返回数据库连接的实例

• 它的具体运作方式

  • 使用 useSource 创建 DataSourceOptions
  • 接着用 DataSourceFactory 响应返回一个全新的 DataSource
  • 这个 DataSource 就是用来连接具体数据库的

• 在此，我们可以考虑做一层小优化, 在响应返回 DataSource 时

• 用一个全局的私有变量存储该 DataSource 实例

• 这样，当下次有 options 传入时，就无需再次创建新的 DataSource，即手动维护其连接实例

• 注意，这里，不同类型数据库每个new出来的客户端都会有各自的连接池

• 我们这里要对这个 dataSource 进行管理

设计租户接口和标识

• get 请求
• 路径 /multi
• headers 参数
  • x-tenant-id: mysql 这是租户1
  • x-tenant-id: mysql2 这是租户2
  • x-tenant-id: postgresql 这是租户3

集成多租户服务

1 ） 新建 typeorm/typeorm-config.service.ts

import { Inject } from '@nestjs/common';
import { REQUEST } from '@nestjs/core';
import { TypeOrmModuleOptions, TypeOrmOptionsFactory } from '@nestjs/typeorm';
import { ConfigService } from '@nestjs/config';export class TypeOrmConfigService implements TypeOrmOptionsFactory {constructor(@Inject(REQUEST) private request: Request,private configService: ConfigService) {}createTypeOrmOptions( connectionName?: string ): TypeOrmModuleOptions | Promise<TypeOrmModuleOptions> {const headers = this.request.headers;const tenantId = headers['x-tenant-id'];const { configService } = this;// 默认 mysql 的const envConfig = {type: configService.get<string>('DB_TYPE'),host: configService.get<string>('DB_HOST'),port: configService.get<number>('DB_PORT'),username: configService.get<string>('DB_USERNAME'),password: configService.get<string>('DB_PASSWORD'),database: configService.get<string>('DB_DATABASE'),autoLoadEntities: Boolean(configService.get<string | boolean>('DB_AUTOLOAD', false)),tenantId, // 额外参数};let config: Record<string, any> = {};switch(tenantId) {// mysql2 的case 'mysql2':config = { port: 13307 };break;// postgres 的case 'postgresql':config = {type: 'postgres',port: 15432,username: 'pguser',password: '123456_postgresql',database: 'testdb',}break;}const finalConfig = Object.assign(envConfig, config) as TypeOrmModuleOptions;console.log('~ finalConfig: ', finalConfig);return finalConfig;}
}

• 注意，这里有很多硬编码的东西，后面会进行配置优化，这里仅作演示举例

2 ) 配置 app.controller.ts

import { Controller, Get } from '@nestjs/common';
import { InjectRepository } from '@nestjs/typeorm';
import { User } from './user/user.entity';
import { Repository } from 'typeorm';@Controller()
export class AppController {constructor(@InjectRepository(User) private userRepository: Repository<User>,) {}@Get('/multi')async getMulti(): Promise<any> {const rs = await this.userRepository.find();return rs;}
}

3 ) 配置 app.module.ts

import { Module } from '@nestjs/common';
import { AppController } from './app.controller';
import { AppService } from './app.service';
import { ConfigModule, ConfigService } from '@nestjs/config';
import { TypeOrmModule, TypeOrmModuleOptions } from '@nestjs/typeorm';
import { DataSource } from 'typeorm';
import { User } from './user/user.entity';
import { TypeOrmConfigService } from './typeorm/typeorm-config.service';const connections = new Map<string, DataSource>();@Module({imports: [// 1. 下面这个后续可以封装一个新的模块，来匹配 .env 和 其他配置ConfigModule.forRoot({  // 配置环境变量模块envFilePath: '.env', // 指定环境变量文件路径isGlobal: true, // 全局可用}),// 2. 集成 TypeormTypeOrmModule.forRootAsync({useClass: TypeOrmConfigService,dataSourceFactory: async (options) => {console.log('connections', connections.keys());const tenantId = options?.['tenantId'] ?? 'mysql';if (tenantId && connections.has(tenantId)) {console.log('reuse');return connections.get(tenantId)!;}console.log('new dataSource');const dataSource = await new DataSource(options!).initialize();connections.set(tenantId, dataSource);return dataSource;},inject:[],extraProviders: [],}),TypeOrmModule.forFeature([User]),],controllers: [AppController],providers: [AppService,// 这里把connections作为全局变量供AppService中使用{provide: 'TYPEORM_CONNECTIONS',useValue: connections,}],
})export class AppModule {}

• 注意，这里的 connections 是一个map用于优化 DataSource 实例的
• 并且这个 connections 后续会被 AppService 使用

4 ) 配置 app.service.ts

import { Inject, OnApplicationShutdown } from '@nestjs/common';
import { DataSource } from 'typeorm';export class AppService implements OnApplicationShutdown {constructor(@Inject('TYPEORM_CONNECTIONS') private connections: Map<string, DataSource>,){}onApplicationShutdown(singal) {console.log('shutdown singal: ', singal);if (this.connections.size > 0) {for(const key of this.connections.keys()) {this.connections.get(key).destroy();}}}
}

• 这里继承了 OnApplicationShutdown 这个 生命周期api, 在异常或关闭时销毁所有数据库实例

5 ）配置 main.ts

import { NestFactory } from '@nestjs/core';
import { AppModule } from './app.module';async function bootstrap() {const app = await NestFactory.create(AppModule);app.enableShutdownHooks(); // 注意这里await app.listen(process.env.PORT ?? 3000);
}
bootstrap();

• 这里开启 app.enableShutdownHooks(); 相关生命周期钩子

6 ) 恢复 .env 的配置

DB_TYPE=mysql
DB_HOST=localhost
DB_PORT=13306
DB_USERNAME=root
DB_PASSWORD=123456_mysql
DB_DATABASE=testdb
DB_AUTOLOAD=true
DB_SYNC=true

• 目前这里是一个配置，后续可以配置多个，使用前缀区分
• 当然，相关程序也需要重新调整

测试效果

1 ）测试租户1

• 请求

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

• 响应

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

2 ) 测试租户2

• 请求

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

• 响应

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

3 ) 测试租户3

• 请求

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

• 响应

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

4 ）综上

• 可以看到，请求头不同，获取的内容也不同
• 从控制台输出的 finalConfig 也可以看出，调用了不同数据库
• 并且可以看到实例基于Map实现了缓存功能
• 关闭程序，也可看到输出了 singal 后面也进行了实例的销毁
• 目前多租户的雏形已经实现了

5 ）数据库配置相关的优化

• 参考后续文章