NestJS微服务复用Kafka客户端做健康检查启动报错求助

问题背景 #

我用NestJS搭建微服务消费Kafka事件，希望复用main.ts中已创建的Kafka客户端/连接来实现应用健康监控，避免新建重复连接，同时能在应用其他模块中注入该kafkajs客户端实例。

原始main.ts中的Kafka连接代码：

app.connectMicroservice<MicroserviceOptions>({
  transport: Transport.KAFKA,
  options: {
    client: {
      brokers: [process.env.KAFKA_BROKER],
    },
    consumer: {
      groupId: process.env.KAFKA_GROUP_ID,
      retry: { retries: 3 },
    },
  },
}, { inheritAppConfig: true });

为了获取私有consumer，我尝试继承ServerKafka类：

export class ExtendedClientKafka extends ServerKafka {
  constructor(options: KafkaOptions['options']) {
    super(options);
  }

  get publicConsumer() {
    return this.consumer;
  }
}

随后修改main.ts使用这个自定义类：

app.connectMicroservice({
  transport: Transport.KAFKA,
  options: new ExtendedClientKafka({
    client: {
      brokers: [process.env.KAFKA_BROKER],   
    },
    consumer: {
      groupId: process.env.KAFKA_GROUP_ID,
      retry: {
        retries: 3
      },
    },
  })
}, { inheritAppConfig: true });

启动应用时触发错误：

TypeError: Cannot destructure property 'createPartitioner' of '(intermediate value)(intermediate value)(intermediate value)' as it is null.
    at Client.producer 

错误原因 #

直接将ExtendedClientKafka实例传给connectMicroservice的options参数是错误逻辑。NestJS的Kafka微服务配置中，options需要接收的是KafkaOptions['options']格式的配置对象，而非ServerKafka实例。ServerKafka是Nest封装的服务器类，内部会自行初始化客户端，手动实例化传入会破坏其内部初始化流程，导致客户端实例未正确创建，进而触发createPartitioner相关的空值错误。

正确实现方案 #

方案一：创建全局Kafka客户端提供者（推荐） #

通过自定义提供者创建全局共享的Kafka客户端，让微服务和健康监控等模块复用同一个实例：

1. 定义Kafka客户端提供者

// src/kafka/kafka-client.provider.ts
import { Provider } from '@nestjs/common';
import { Kafka, KafkaConfig } from 'kafkajs';
import { ConfigService } from '@nestjs/config';

export const KAFKA_CLIENT = 'KAFKA_CLIENT';

export const KafkaClientProvider: Provider = {
  provide: KAFKA_CLIENT,
  useFactory: (configService: ConfigService) => {
    const kafkaConfig: KafkaConfig = {
      brokers: [configService.get('KAFKA_BROKER')],
    };
    return new Kafka(kafkaConfig);
  },
  inject: [ConfigService],
};

2. 在Module中注册提供者

// src/app.module.ts
import { Module } from '@nestjs/common';
import { ConfigModule } from '@nestjs/config';
import { KafkaClientProvider } from './kafka/kafka-client.provider';

@Module({
  imports: [ConfigModule.forRoot({ isGlobal: true })],
  providers: [KafkaClientProvider],
})
export class AppModule {}

3. 微服务配置复用全局客户端

修改main.ts，在Kafka微服务配置中指定已创建的全局客户端：

// src/main.ts
import { NestFactory } from '@nestjs/core';
import { AppModule } from './app.module';
import { Transport, MicroserviceOptions } from '@nestjs/microservices';
import { ConfigService } from '@nestjs/config';
import { KAFKA_CLIENT } from './kafka/kafka-client.provider';

async function bootstrap() {
  const app = await NestFactory.create(AppModule);
  const configService = app.get(ConfigService);
  const kafkaClient = app.get(KAFKA_CLIENT);

  app.connectMicroservice<MicroserviceOptions>({
    transport: Transport.KAFKA,
    options: {
      client: kafkaClient, // 复用全局Kafka客户端
      consumer: {
        groupId: configService.get('KAFKA_GROUP_ID'),
        retry: { retries: 3 },
      },
    },
  }, { inheritAppConfig: true });

  await app.startAllMicroservices();
  await app.listen(3000);
}
bootstrap();

4. 注入客户端到健康监控或其他服务

现在可以在任意服务中注入该Kafka客户端，例如实现健康监控：

// src/health/kafka.health.ts
import { Injectable, Inject } from '@nestjs/common';
import { HealthIndicator, HealthIndicatorResult } from '@nestjs/terminus';
import { KAFKA_CLIENT } from '../kafka/kafka-client.provider';
import { Kafka } from 'kafkajs';

@Injectable()
export class KafkaHealthIndicator extends HealthIndicator {
  constructor(@Inject(KAFKA_CLIENT) private readonly kafkaClient: Kafka) {}

  async checkHealth(): Promise<HealthIndicatorResult> {
    try {
      // 通过获取元数据测试Kafka连接状态
      await this.kafkaClient.admin().fetchTopicMetadata({ topics: [] });
      return this.getStatus('kafka', true);
    } catch (error) {
      return this.getStatus('kafka', false, { error: error.message });
    }
  }
}

方案二：从微服务实例中获取内部客户端 #

若不想单独创建提供者，也可从启动后的微服务实例中直接获取Nest内部的Kafka客户端/消费者：

// src/main.ts
import { NestFactory } from '@nestjs/core';
import { AppModule } from './app.module';
import { Transport, MicroserviceOptions } from '@nestjs/microservices';
import { ConfigService } from '@nestjs/config';
import { ServerKafka } from '@nestjs/microservices/external/kafka.interface';

async function bootstrap() {
  const app = await NestFactory.create(AppModule);
  const configService = app.get(ConfigService);

  const microservice = app.connectMicroservice<MicroserviceOptions>({
    transport: Transport.KAFKA,
    options: {
      client: {
        brokers: [configService.get('KAFKA_BROKER')],
      },
      consumer: {
        groupId: configService.get('KAFKA_GROUP_ID'),
        retry: { retries: 3 },
      },
    },
  }, { inheritAppConfig: true });

  await app.startAllMicroservices();

  // 获取ServerKafka实例并提取内部客户端/消费者
  const serverKafka = microservice.getServer() as ServerKafka;
  const kafkaClient = serverKafka['client'];
  const kafkaConsumer = serverKafka['consumer'];

  // 可将客户端存入全局容器，供其他模块注入使用
  await app.listen(3000);
}
bootstrap();

注意：此方案依赖Nest内部私有属性，版本升级时可能存在兼容性问题，需谨慎使用。

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