基于RabbitMQ消息动态配置NestJS多Mongo连接方案问询

需求背景 #

我全网查找过相关实现方案但没有结果，参考动态切换数据库连接的思路，希望在NestJS项目中，根据RabbitMQ Broker传来的Payload动态建立多个MongoDB连接。

实现方案 #

和HTTP请求拦截动态切换连接的核心思路一致，但因为是消息驱动场景，需要在消息消费流程中处理动态连接，利用Mongoose的多连接特性管理连接池，避免重复创建连接。

1. 调整全局MongoDB配置

修改AppModule，移除预先初始化的全局Mongo连接，改为提供动态连接管理服务：

@Module({
  imports: [
    ConfigModule.forRoot({
      isGlobal: true,
      load: [envConfig]
    }),
    // 移除全局Mongoose初始化，改为动态创建
    // MongooseModule.forRootAsync({
    //   useClass: MongoConfig,
    //   inject: [ConfigService]
    // })
  ],
  controllers: [AppController],
  providers: [AppService, MongoConnectionService],
  exports: [MongoConnectionService]
})
export class AppModule {}

2. 实现动态Mongo连接管理服务

创建MongoConnectionService，负责根据Payload中的信息创建/复用Mongo连接：

import { Injectable } from '@nestjs/common';
import { ConfigService } from '@nestjs/config';
import { createConnection, Connection, MongooseModuleOptions } from 'mongoose';

@Injectable()
export class MongoConnectionService {
  private connections = new Map<string, Connection>();
  private baseMongoConfig: MongooseModuleOptions;

  constructor(private readonly config: ConfigService) {
    // 读取基础Mongo配置（比如主机、端口前缀）
    this.baseMongoConfig = {
      uri: this.config.get('mongo').baseUri // 示例：mongodb://localhost:27017/
    };
  }

  async getConnection(payloadDbInfo: { dbName: string; uri?: string }): Promise<Connection> {
    const connKey = payloadDbInfo.uri || `${this.baseMongoConfig.uri}${payloadDbInfo.dbName}`;
    
    // 连接已存在则直接返回
    if (this.connections.has(connKey)) {
      return this.connections.get(connKey);
    }

    // 构建最终连接配置
    const connOptions: MongooseModuleOptions = {
      uri: payloadDbInfo.uri || `${this.baseMongoConfig.uri}${payloadDbInfo.dbName}`,
      // 可添加额外配置：连接超时、认证信息、连接池参数等
    };

    // 创建新连接并缓存
    const connection = await createConnection(connOptions.uri);
    this.connections.set(connKey, connection);
    
    return connection;
  }
}

3. 在RabbitMQ消息处理器中使用动态连接

修改MessagingService，在消费消息时根据Payload获取对应Mongo连接，再执行数据库操作：

import { Injectable } from '@nestjs/common';
import { RabbitSubscribe } from '@golevelup/nestjs-rabbitmq';
import { MongoConnectionService } from './mongo-connection.service';
import { BusinessModel } from './schemas/business.schema'; // 替换为你的数据模型

@Injectable()
export class MessagingService {
  constructor(private readonly mongoConnService: MongoConnectionService) {}

  @RabbitSubscribe({
    exchange: 'your-target-exchange',
    routingKey: 'your-routing-key',
    queue: 'your-consume-queue'
  })
  async handleMessage(payload: { dbName: string; businessData: any }) {
    // 根据Payload获取对应Mongo连接
    const connection = await this.mongoConnService.getConnection({ dbName: payload.dbName });
    // 基于当前连接获取数据模型
    const Model = connection.model(BusinessModel.name, BusinessModel.schema);
    
    // 执行具体数据库操作（示例：插入数据）
    await Model.create(payload.businessData);
  }
}

4. 调整MessagingModule

确保MessagingModule引入动态连接服务：

@Module({
  imports: [
    RabbitMQModule.forRootAsync(RabbitMQModule, {
      imports: [ConfigModule],
      useClass: AmqpConfig,
      inject: [ConfigService],
    })
  ],
  controllers: [MessagingController],
  providers: [MessagingService, ConfigService, MongoConnectionService],
  exports: [MessagingService, RabbitMQModule]
})
export class MessagingModule {}

关键注意事项 #

• 连接缓存：通过Map维护连接池，避免为相同数据库重复创建连接，降低资源消耗。
• 模型绑定：必须基于当前动态连接注册/获取模型，不能使用全局默认连接的模型实例。
• 资源清理：可根据业务需求添加连接过期清理逻辑，比如定期检查空闲连接并关闭，防止连接泄漏。

内容的提问来源于stack exchange，提问作者Andy95