microservices - Nestjs kafka实现

Question

我已经阅读了 nestjs 微服务和 kafka 文档，但我无法理解其中的一些内容。如果你能帮助我，我将非常感激。因此，正如文档所说，我必须在 main.ts 文件中创建一个微服务，如下所示:

const app = await NestFactory.createMicroservice<MicroserviceOptions>(AppModule, {
  transport: Transport.KAFKA,
  options: {
    client: {
      brokers: ['localhost:9092'],
    }
  }
});

await app.listen(() => console.log('app started'));

然后有一个kafkaModule文件是这样的:

@Module({
  imports: [
    ClientsModule.register([
      {
        name: 'HERO_SERVICE',
        transport: Transport.KAFKA,
        options: {
          client: {
            clientId: 'hero',
            brokers: ['localhost:9092'],
          },
          consumer: {
            groupId: 'hero-consumer'
          }
        }
      },
    ]),
  ]
})
export class KafkaModule implements OnModuleInit {
  constructor(@Inject('HERO_SERVICE') private readonly clientService: KafkaClient)
  async onModuleInit() {
    await this.clientService.connect();
  }
}

我首先想不通的是 createMicroservice 的第一个参数有什么用？ (我通过了 AppModule 和 KafkaModule，两者都工作正常。知道 kafkaModule 是在 appModule 导入的)

另一件事是，据我了解，微服务部分和 main.ts 文件中的配置用于订阅 MessagePattern 或 EventPattern 装饰器中使用的主题，并使用 kafkaModule 中描述的 kafkaClient向不同主题发送消息。

这里的问题是，如果我之前所说的是真的，那么如果 clientModule 没有指定作为消费者工作，为什么会使用默认的 groupId。奇怪的是我找不到使用 clientModule 从任何主题获取任何消息的解决方案。我现在正在做的是在每个文件中使用不同的组 ID，这样它们就不会发生任何冲突。

Answer 1

createMicroservice 的第一个参数，当你想消费来自特定主题的消息时，它将帮助指导消费者如何连接到 Kafka。

示例:我们想从主题中获取消息:test01

我们如何申报？

import {Controller} from '@nestjs/common'
import {MessagePattern, Payload} from '@nestjs/microservices'

@Controller('sync')
export class SyncController {
  @MessagePattern('test01')
  handleTopicTest01(@Payload() message: Sync): any {
  // Handle your message here
  }
}

第二个 block 用作非消费者的生产者。当应用程序想要向特定主题发送消息时，clientModel 将支持这一点。

@Get()
sayHello() {
    return this.clientModule.send('say.hello', 'hello world')
}