如何在一项服务中使用多个ClientKafka？

Question

我在一项服务中使用多个clientkafka的问题有问题，这是我的实现：

@Controller()
export class ApiController implements OnModuleInit {
  constructor(
    @Inject("ACCOUNT_SERVICE") private readonly accountKafkaClient: ClientKafka,
    @Inject("WORKSPACE_SERVICE") private readonly workspaceKafkaClient: ClientKafka
  ) { }

  async onModuleInit() {
    const requestPatterns = [
      'topic'
    ];
    requestPatterns.forEach((pattern) => {
      this.accountKafkaClient.subscribeToResponseOf(`account.${pattern}`);
    });
    await this.accountKafkaClient.connect();
  }

  async onModuleDestroy() {
    await this.accountKafkaClient.close();
  }

  @Get()
  async sendMessage() {
    const data = {
      msg: "account.topic"
    }

    const kafkaResponse = this.accountKafkaClient.send<any>('account.topic', JSON.stringify(data));
    const response = await firstValueFrom(kafkaResponse);

    const kafkaResponse2 = this.workspaceKafkaClient.send<any>('workspace.topic', JSON.stringify(response )) //THIS IS NOT RUNNING, WORKSPACE_SERVICE NOT RECEIVE ANY MESSAGE

    return await firstValueFrom(kafkaResponse2);
  }
}

有人可以告诉我为什么workspacekafkaclient没有将任何消息发送到workspace_service微服务？我尝试在onModule中传递此客户端... accountkafkaclient之类的功能，但这对我没有帮助，

这也是我在模块中的设置：

@Module({
  imports: [
    ClientsModule.register([
      {
        name: 'ACCOUNT_SERVICE',
        transport: Transport.KAFKA,
        options: {
          client: {
            clientId: 'account_service',
            brokers: ['localhost:29092'],
          },
          consumer: {
            groupId: 'account-consumer',
          },
        },
      },
      {
        name: 'WORKSPACE_SERVICE',
        transport: Transport.KAFKA,
        options: {
          client: {
            clientId: 'workspace_service',
            brokers: ['localhost:29092'],
          },
          consumer: {
            groupId: 'workspace-consumer',
          },
        },
      },
    ]),
  ],
  controllers: [ApiController],
  providers: [
    ApiService,
    // KafkaProducerProvider,
  ],
})
export class ApiModule {}

感谢您的任何帮助！

Answer 1

每个应用程序您只需要一个生产者客户端，但是Kafka生产商从未立即将数据发送给经纪人。

您需要将它们冲洗才能发生，这是等待FirstValuefrom（...）应该这样做的事情，但是您尚未显示该方法。

否则，您似乎正在尝试从一个主题中获得回复，以发送到另一个主题，这是消费者应使用的方法，而不是在一个生产者请求中阻止。