NestJS请求作用域下如何替代ModuleInit初始化Kafka消费者？

方案1：拆分依赖，解耦请求上下文与消费者逻辑 #

Kafka消费者是后台常驻服务，本身不依附于任何HTTP请求，直接依赖REQUEST作用域服务本身就违背了两者的设计逻辑。你可以把RequestScopedService中不需要请求上下文的核心业务逻辑抽离成单例服务，让Kafka消费者依赖这个单例；而请求相关的逻辑留在原请求作用域服务中，仅供HTTP请求场景使用。

1.1 抽离单例共享服务 #

@Injectable()
export class SharedBusinessService {
  // 存放原RequestScopedService中不依赖HTTP请求的业务逻辑
  processMessageContent(content: string) {
    // 示例处理逻辑
    return content.toUpperCase();
  }
}

1.2 修改请求作用域服务 #

import { Injectable, Scope, Inject } from '@nestjs/common';
import { REQUEST } from '@nestjs/core';
import { Request } from 'express';
import { SharedBusinessService } from './shared-business.service';

@Injectable({ scope: Scope.REQUEST })
export class RequestScopedService {
  constructor(
    @Inject(REQUEST) private request: Request,
    private sharedService: SharedBusinessService // 依赖单例共享服务
  ) {}

  // 仅保留需要HTTP请求上下文的逻辑
  getCurrentRequestUserId() {
    return this.request.user?.id;
  }
}

1.3 修改Kafka消费者 #

@Injectable()
export class KafkaConsumer implements OnModuleInit {
  constructor(
    private sharedService: SharedBusinessService // 依赖单例共享服务
  ) {}

  async onModuleInit() {
    console.log('初始化Kafka消费者');
    const consumer = kafka.consumer({ groupId: 'my-group' })
    await consumer.connect()

    await consumer.subscribe({ topics: ['topic-A'] })
    await consumer.run({
        eachMessage: async ({ message }) => {
            console.log('处理消息');
            const rawContent = message.value.toString();
            // 使用单例服务处理业务逻辑
            const processedContent = this.sharedService.processMessageContent(rawContent);
            console.log('处理后内容:', processedContent);
        },
    })
  }
}

方案2：通过消息传递上下文信息，避免依赖请求作用域服务 #

如果业务需要在Kafka消息处理中用到HTTP请求相关的信息，不要直接依赖RequestScopedService，而是在生产消息时把需要的上下文信息（如用户ID、请求ID）放到消息的headers或payload中，消费者直接从消息里提取使用即可。

2.1 生产者端（HTTP请求场景） #

@Injectable()
export class KafkaProducerService {
  constructor(private kafka: Kafka, @Inject(REQUEST) private request: Request) {}

  async sendMessage(data: any) {
    const producer = this.kafka.producer();
    await producer.connect();
    
    await producer.send({
      topic: 'topic-A',
      messages: [
        {
          value: JSON.stringify(data),
          headers: {
            userId: this.request.user?.id?.toString(), // 把请求用户ID放入消息头
            requestId: this.request.headers['x-request-id']?.toString()
          }
        }
      ]
    });
    
    await producer.disconnect();
  }
}

2.2 消费者端 #

@Injectable()
export class KafkaConsumer implements OnModuleInit {
  async onModuleInit() {
    console.log('初始化Kafka消费者');
    const consumer = kafka.consumer({ groupId: 'my-group' })
    await consumer.connect()

    await consumer.subscribe({ topics: ['topic-A'] })
    await consumer.run({
        eachMessage: async ({ message }) => {
            console.log('处理消息');
            // 从消息头提取上下文信息
            const userId = message.headers?.userId?.toString();
            const requestId = message.headers?.requestId?.toString();
            
            // 直接使用这些信息处理业务
            console.log(`处理用户${userId}的消息，请求追踪ID：${requestId}`);
        },
    })
  }
}

方案3：手动创建请求上下文（不推荐） #

如果以上两种方案都无法满足需求，你可以通过ModuleRef手动创建RequestScopedService的实例，但需要模拟一个HTTP请求上下文。这种方式仅作为权宜之计，因为它违背了请求作用域服务的设计初衷，容易引发逻辑混乱。

示例代码：

import { Injectable, OnModuleInit, ModuleRef } from '@nestjs/common';
import { ExecutionContext } from '@nestjs/common/interfaces';

@Injectable()
export class KafkaConsumer implements OnModuleInit {
  constructor(private moduleRef: ModuleRef) {}

  async onModuleInit() {
    console.log('初始化Kafka消费者');
    const consumer = kafka.consumer({ groupId: 'my-group' })
    await consumer.connect()

    await consumer.subscribe({ topics: ['topic-A'] })
    await consumer.run({
        eachMessage: async () => {
            console.log('处理消息');
            // 模拟HTTP请求上下文
            const mockContext: ExecutionContext = {
              switchToHttp: () => ({
                getRequest: () => ({
                  user: { id: 'mock-kafka-user-001' },
                  headers: { 'x-request-id': 'kafka-message-12345' }
                })
              })
            } as any;

            // 动态获取请求作用域服务实例
            const requestScopedService = await this.moduleRef.resolve(RequestScopedService, mockContext);
            // 使用服务逻辑
            console.log('模拟用户ID:', requestScopedService.getCurrentRequestUserId());
        },
    })
  }
}

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