3、ES模块封装：创建ElasticsearchModule、动态模块配置、连接池管理、健康检查

在 NestJS 中集成 Elasticsearch 的最佳实践是将客户端连接、配置与生命周期管理封装为一个独立的动态模块。本节将带领你从零构建 ElasticsearchModule，涵盖动态模块注册、连接池调优以及健康检查机制，确保搜索服务在生产环境中具备高可用性与可观测性。

3.1 模块架构概览

下图展示了 ElasticsearchModule 的核心组件及其交互关系：

3.2 动态模块配置

NestJS 动态模块允许在导入时传递配置，实现多环境适配。我们提供两种注册方式：同步配置（forRoot）与异步配置（forRootAsync）。

// elasticsearch.module.ts
import { DynamicModule, Module, Provider } from '@nestjs/common';
import { ElasticsearchService } from './elasticsearch.service';
import {
  ElasticsearchModuleOptions,
  ElasticsearchModuleAsyncOptions,
} from './interfaces';

@Module({})
export class ElasticsearchModule {
  static forRoot(options: ElasticsearchModuleOptions): DynamicModule {
    return {
      module: ElasticsearchModule,
      providers: [
        {
          provide: 'ELASTICSEARCH_OPTIONS',
          useValue: options,
        },
        ElasticsearchService,
      ],
      exports: [ElasticsearchService],
    };
  }

  static forRootAsync(options: ElasticsearchModuleAsyncOptions): DynamicModule {
    const asyncProvider: Provider = {
      provide: 'ELASTICSEARCH_OPTIONS',
      useFactory: options.useFactory,
      inject: options.inject || [],
    };
    return {
      module: ElasticsearchModule,
      imports: options.imports || [],
      providers: [asyncProvider, ElasticsearchService],
      exports: [ElasticsearchService],
    };
  }
}

3.3 连接池管理

Elasticsearch 官方客户端默认使用 连接池（Connection Pool） 来复用 TCP 连接，减少握手开销。在生产环境中，需要根据节点数量与并发量调整以下参数：

• maxRetries：请求失败时的最大重试次数，默认 3。
• requestTimeout：单个请求超时时间（毫秒），默认 30000。
• pingTimeout：节点健康检查超时，默认 3000。
• sniffOnStart：启动时是否嗅探集群所有节点，默认 false。
• sniffInterval：定期嗅探间隔（毫秒），默认 60000。
• resurrectStrategy：死节点复活策略，可选 ping / optimistic / none。

// elasticsearch.service.ts
import { Injectable, Inject, OnModuleDestroy } from '@nestjs/common';
import { Client } from '@elastic/elasticsearch';
import { ElasticsearchModuleOptions } from './interfaces';

@Injectable()
export class ElasticsearchService implements OnModuleDestroy {
  private readonly client: Client;

  constructor(
    @Inject('ELASTICSEARCH_OPTIONS')
    private readonly options: ElasticsearchModuleOptions,
  ) {
    this.client = new Client({
      node: this.options.node,
      maxRetries: this.options.maxRetries ?? 3,
      requestTimeout: this.options.requestTimeout ?? 30000,
      pingTimeout: this.options.pingTimeout ?? 3000,
      sniffOnStart: this.options.sniffOnStart ?? false,
      sniffInterval: this.options.sniffInterval ?? 60000,
      resurrectStrategy: this.options.resurrectStrategy ?? 'ping',
      // 连接池大小控制
      connectionPool: {
        maxRetries: 3,
        resurrectStrategy: 'ping',
      },
    });
  }

  getClient(): Client {
    return this.client;
  }

  async onModuleDestroy() {
    await this.client.close();
  }
}

3.4 健康检查

健康检查是微服务架构中不可或缺的环节。我们通过 @nestjs/terminus 集成 Elasticsearch 健康指示器，或手动实现 Ping 探针。

3.4.1 自定义健康检查方法

// elasticsearch.service.ts（续）
export class ElasticsearchService implements OnModuleDestroy {
  // ... 省略构造与 getClient

  async healthCheck(): Promise<{ status: string; nodes?: number }> {
    try {
      const info = await this.client.info();
      return {
        status: 'ok',
        nodes: info.body?.number_of_nodes ?? 1,
      };
    } catch (error) {
      return {
        status: 'error',
        nodes: 0,
      };
    }
  }
}

3.4.2 集成 Terminus 健康端点

// health.controller.ts
import { Controller, Get } from '@nestjs/common';
import {
  HealthCheckService,
  HealthCheck,
  MicroserviceHealthIndicator,
} from '@nestjs/terminus';
import { ElasticsearchService } from './elasticsearch.service';

@Controller('health')
export class HealthController {
  constructor(
    private health: HealthCheckService,
    private elasticsearchService: ElasticsearchService,
  ) {}

  @Get()
  @HealthCheck()
  check() {
    return this.health.check([
      async () => {
        const result = await this.elasticsearchService.healthCheck();
        return {
          elasticsearch: {
            status: result.status === 'ok' ? 'up' : 'down',
            nodes: result.nodes,
          },
        };
      },
    ]);
  }
}

3.5 完整模块配置参数表

3.6 模块使用示例

// app.module.ts
import { Module } from '@nestjs/common';
import { ConfigModule, ConfigService } from '@nestjs/config';
import { ElasticsearchModule } from './elasticsearch/elasticsearch.module';
import { HealthController } from './health.controller';

@Module({
  imports: [
    ConfigModule.forRoot(),
    ElasticsearchModule.forRootAsync({
      imports: [ConfigModule],
      useFactory: (configService: ConfigService) => ({
        node: configService.get('ES_NODE') || 'http://localhost:9200',
        maxRetries: 3,
        requestTimeout: 30000,
        sniffOnStart: true,
        sniffInterval: 60000,
        auth: {
          username: configService.get('ES_USERNAME'),
          password: configService.get('ES_PASSWORD'),
        },
      }),
      inject: [ConfigService],
    }),
  ],
  controllers: [HealthController],
})
export class AppModule {}