NestJS中Kafka客户端依赖注入错误排查求助

错误核心是Nest无法为CampaignProducer注入依赖的Kafka实例：

• CampaignProducer被标记为@Injectable()，Nest会自动尝试注入其构造函数中的参数，但Kafka是第三方类，并未注册为Nest容器内的Provider。
• 你在MicroserviceClientModule中直接声明CampaignProducer为Provider，但未提供Kafka实例的注入来源。

方案一：通过KafkaService获取Producer实例（推荐） #

此方案复用你已有的KafkaService对Producer生命周期的管理，无需修改原有核心逻辑：

1. 修改MicroserviceClientModule，移除CampaignProducer的Provider声明：

@Module({
    imports: [KafkaModule],
    // 移除CampaignProducer，无需Nest单独创建它
})
export class MicroserviceClientModule {}

2. 在业务服务中注入KafkaService获取Producer：

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

    async sendCampaignMsg(content: string) {
        const producer = this.kafkaService.getCampaignProducer();
        await producer.sendCampaignMessage(content);
    }
}

方案二：将Kafka实例注册为Nest Provider（符合DI规范） #

适合需要直接注入CampaignProducer的场景，需调整多处代码：

1. 修改KafkaModule，注册Kafka实例为Provider：

@Module({
    providers: [
        KafkaService,
        // 用工厂函数创建Kafka实例并注册为Provider
        {
            provide: 'KAFKA_INSTANCE',
            useFactory: () => new Kafka({
                clientId: 'your-client-id',
                brokers: ['localhost:9092'],
            }),
        },
    ],
    exports: [KafkaService, 'KAFKA_INSTANCE'],
})
export class KafkaModule {}

2. 修改CampaignProducer，指定注入的Kafka实例Token：

@Injectable()
export class CampaignProducer {
    private producer: Producer;
    private readonly topic = 'campaign-topic';

    constructor(@Inject('KAFKA_INSTANCE') private readonly kafka: Kafka) {
        this.producer = this.kafka.producer();
    }

    async sendCampaignMessage(message: string): Promise<void> {
        await this.producer.connect();
        await this.producer.send({
            topic: this.topic,
            messages: [{ value: message }],
        });
    }
}

3. 修改KafkaService，通过DI注入Kafka和CampaignProducer：

@Injectable()
export class KafkaService implements OnModuleInit {
    constructor(
        @Inject('KAFKA_INSTANCE') private readonly kafka: Kafka,
        private readonly campaignProducer: CampaignProducer,
    ) {
        this.campaignConsumer = new CampaignConsumer(this.kafka);
    }

    async onModuleInit(): Promise<void> {
        await this.campaignConsumer.consumeCampaignMessages();
    }

    getCampaignConsumer(): CampaignConsumer {
        return this.campaignConsumer;
    }

    getCampaignProducer(): CampaignProducer {
        return this.campaignProducer;
    }
}

4. 更新KafkaModule，添加CampaignProducer到Providers并导出：

@Module({
    providers: [
        KafkaService,
        {
            provide: 'KAFKA_INSTANCE',
            useFactory: () => new Kafka({
                clientId: 'your-client-id',
                brokers: ['localhost:9092'],
            }),
        },
        CampaignProducer,
        CampaignConsumer, // 若Consumer也需要DI，同理添加
    ],
    exports: [KafkaService, 'KAFKA_INSTANCE', CampaignProducer],
})
export class KafkaModule {}

5. 在MicroserviceClientModule中直接注入CampaignProducer：

@Injectable()
export class YourBusinessService {
    constructor(private readonly campaignProducer: CampaignProducer) {}

    async sendCampaignMsg(content: string) {
        await this.campaignProducer.sendCampaignMessage(content);
    }
}

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