typescript - Nestjs微服务中的动态kafka主题名称

Question

在 Nestjs 中，我使用 kafka 作为消息代理并设置主题名称，如下所示：

@MessagePattern('topic-name')
async getNewRequest(@Payload() message: any): Promise<void> {
  // my code goes here
}

有没有办法从配置服务模块中读取 kafka 主题名称？

Answer 1

我通过创建一个新的自定义装饰器来处理这个问题。

export function KafkaTopic(variable: string | keyof AppConfig): any {
  return (
    target: any,
    key: string | symbol,
    descriptor: PropertyDescriptor,
  ) => {
    Reflect.defineMetadata(
      KAFKA_TOPIC_METADATA,
      variable,
      descriptor.value,
    );
    return descriptor;
  };

然后用 MessagePattern 动态替换它并从 appConfig 设置主题名称：

@Injectable()
export class KafkaDecoratorProcessorService {
  constructor(
    private readonly LOG: Logger,
    private readonly appConfig: AppConfig,
  ) {
  }

  processKafkaDecorators(types: any[]) {
    for (const type of types) {
      const propNames = Object.getOwnPropertyNames(type.prototype);
      for (const prop of propNames) {
        const propValue = Reflect.getMetadata(
          KAFKA_TOPIC_METADATA,
          Reflect.get(type.prototype, prop),
        );

        if (propValue) {
          const topic = this.appConfig[propValue];
          this.LOG.log(`Setting topic ${topic} for ${type.name}#${prop}`);
          Reflect.decorate(
            [MessagePattern(topic)],
            type.prototype,
            prop,
            Reflect.getOwnPropertyDescriptor(type.prototype, prop),
          );
        }
      }
    }
  }
}

这是在 main.ts 文件中运行 processKafkaDecorators 的方法：

const app = await NestFactory.create(AppModule);
  app
    .get(KafkaDecoratorProcessorService)
    .processKafkaDecorators([AppController]);

  app.connectMicroservice({
    transport: Transport.KAFKA,
    ...
   })

请注意，您必须在连接微服务之前运行它。并像这样使用它：

@KafkaTopic('KAFKA_TOPIC_BOOK_UPDATE')
  async processMessage(
    @Payload() { value: payload }: { value: BookUpdateModel },
  ) {
    ...
  }

资源