NestJS全局动态模块KafkaService注入失败问题及解决思路

我有一个包含KafkaService的全局动态模块，希望在AuthenticationService中使用该服务，但NestJS无法完成KafkaService的注入。

KafkaModule（全局动态模块） #

import { DynamicModule, Global, Module } from '@nestjs/common';
import { KafkaService } from './kafka.service';
import { KafkaConfig } from './kafka.types';

@Global()
@Module({})
export class KafkaModule {
  static register(kafkaConfig: KafkaConfig): DynamicModule {
    return {
      module: KafkaModule,
      global: true,
      providers: [
        {
          provide: KafkaService,
          useValue: new KafkaService(kafkaConfig),
        },
      ],
      exports: [KafkaService],
    };
  }
}

AppModule（模块注册） #

import { Module } from '@nestjs/common';
import { KafkaModule } from './common/kafka/kafka.module';
import { AuthenticationModule } from './modules/authentication.module';

@Module({
  imports: [
    KafkaModule.register({
      clientId: 'abc',
      brokers: ['localhost:9092'],
      groupId: 'def',
    }),
    AuthenticationModule,
  ],
})
export class AppModule {}

AuthenticationService（注入尝试） #

import { Injectable } from '@nestjs/common';
import { SubscribeTo } from 'src/common/kafka/kafka.decorator';
import { KafkaService } from 'src/common/kafka/kafka.service';
import { Constant } from 'src/constant';

type LoginPayload = {
  username: string;
  password: string;
};

@Injectable()
export class AuthenticationService {
  constructor(private kafkaService: KafkaService) {}

  @SubscribeTo(Constant.TOPIC_LOGIN)
  async login(payload: LoginPayload) {
    console.log(this.kafkaService);
  }
}

问题排查结果 #

经排查发现，装饰器的执行时机早于服务注入：@SubscribeTo(Constant.TOPIC_LOGIN)装饰器会将函数保存到变量中供后续调用，但装饰器运行时，服务尚未注入到类中。

因此计划不再使用装饰器订阅主题，同时修改Kafka模块，使消费者不在模块初始化阶段启动。

具体解决方案 #

1. 修改KafkaService，延迟消费者启动并提供手动订阅方法

将消费者初始化和启动逻辑从构造函数中分离，改为通过订阅方法触发：

import { Injectable, Inject } from '@nestjs/common';
import { KafkaConfig } from './kafka.types';
import { Kafka, Consumer, EachMessagePayload } from 'kafkajs';

@Injectable()
export class KafkaService {
  private kafka: Kafka;
  private consumer: Consumer;
  private subscriptions = new Map<string, (payload: any) => Promise<void>>();
  private isConsumerRunning = false;

  constructor(@Inject('KAFKA_CONFIG') private readonly kafkaConfig: KafkaConfig) {
    this.kafka = new Kafka(kafkaConfig);
    this.consumer = this.kafka.consumer({ groupId: kafkaConfig.groupId });
  }

  // 手动订阅主题并绑定处理函数
  async subscribe(topic: string, handler: (payload: any) => Promise<void>) {
    this.subscriptions.set(topic, handler);
    
    if (!this.consumer.connected()) {
      await this.consumer.connect();
    }
    
    await this.consumer.subscribe({ topic, fromBeginning: true });
    
    // 仅首次订阅时启动消费者消息循环
    if (!this.isConsumerRunning) {
      this.isConsumerRunning = true;
      await this.consumer.run({
        eachMessage: async ({ topic, message }: EachMessagePayload) => {
          const handler = this.subscriptions.get(topic);
          if (handler) {
            const payload = JSON.parse(message.value.toString());
            await handler(payload);
          }
        },
      });
    }
  }

  // 可选：提供断开连接方法
  async disconnect() {
    await this.consumer.disconnect();
    this.isConsumerRunning = false;
  }
}

2. 调整KafkaModule，让NestJS管理服务实例

避免手动实例化KafkaService，改为通过依赖注入传递配置：

import { DynamicModule, Global, Module } from '@nestjs/common';
import { KafkaService } from './kafka.service';
import { KafkaConfig } from './kafka.types';

@Global()
@Module({})
export class KafkaModule {
  static register(kafkaConfig: KafkaConfig): DynamicModule {
    return {
      module: KafkaModule,
      global: true,
      providers: [
        {
          provide: 'KAFKA_CONFIG',
          useValue: kafkaConfig,
        },
        KafkaService,
      ],
      exports: [KafkaService],
    };
  }
}

3. 修改AuthenticationService，使用生命周期钩子完成订阅

移除装饰器，在服务初始化完成后手动调用订阅方法：

import { Injectable, OnModuleInit } from '@nestjs/common';
import { KafkaService } from 'src/common/kafka/kafka.service';
import { Constant } from 'src/constant';

type LoginPayload = {
  username: string;
  password: string;
};

@Injectable()
export class AuthenticationService implements OnModuleInit {
  constructor(private readonly kafkaService: KafkaService) {}

  // 模块初始化完成后执行订阅
  async onModuleInit() {
    // 使用bind确保this指向当前服务实例
    await this.kafkaService.subscribe(Constant.TOPIC_LOGIN, this.login.bind(this));
  }

  // 普通处理方法，无需装饰器
  async login(payload: LoginPayload) {
    console.log(this.kafkaService); // 现在可正常访问KafkaService实例
    // 编写登录业务逻辑
  }
}

内容的提问来源于stack exchange，提问作者Rahandi Noor Pasha