microservices - 带有 1 个生产者和 2 个消费者的 Nestjs 基于事件的消息传递

Question

使用 nestjs 微服务，您可以发送消息和 接收 结果使用基于请求/响应的方法。这是通过 @MessagePattern 的组合实现的和 client.send('my_pattern', myData) .有关示例，请参阅 nest 文档:https://docs.nestjs.com/microservices/basics#request-response和 https://docs.nestjs.com/microservices/basics#sending-messages .
如何在基于事件的方法中接收结果？
假设您有一个用户微服务和一个 auth 微服务。每当创建用户时，您还希望创建一个身份验证主题(保存用户名和密码的哈希值，以便用户可以使用 api 请求登录身份验证微服务而不是用户服务)。
auth/auth.controller.ts

  @EventPattern('EVT_USER_CREATED')
  public async handleUserCreated(data: any): Promise<AuthSubject> {
    if (!data.username || !data.password) {
      throw new RpcException('Auth subject must supply username and password');
    } 
    const newSubject: CreateAuthSubject = {
      username: data.username,
      password: data.password,
      email: data.email ?? '',
    };
    const sub = await this.subjectService.create(subject);
    return sub;
  }

用户/user.controller.ts

  @Post('')
  @ApiBody({ type: CreateUser })
  @ApiCreatedResponse({ type: User })
  public async create(@Body() user: CreateUser): Promise<User> {
    const newUser = await this.userService.create(user);

    this.userQueue
      .emit<any, CreateUser>('EVT_USER_CREATED', user)
      .subscribe((res) => {
        console.log(res);   // undefined
      });

    return newUser;
  }

为了验证我的设置没有错误，我更改了 @EventPattern至 @MessagePattern和 this.userQueue.emit<...至 this.userQueue.send<... .它起作用了，即 res 是一个有效的身份验证主题，具有预期的用户名和密码。但是，使用此问题中概述的基于事件的方法 res是 总是 undefined (无论身份验证 Controller handleUserCreated 是否返回或抛出)。
最终我想实现以下目标:如果另一个微服务需要处理“EVT_USER_CREATED”事件，我只需添加一个 @EventPattern('EVT_USER_CREATED') public async handleUserCreated方法到其 Controller 。可观察的 this.userQueue.emit<any, CreateUser>('EVT_USER_CREATED', user)然后将收到两个结果:每个微服务消费用户创建的事件一次。
所以假设我添加了第三个微服务:负责保存支付信息、订单历史等的客户微服务。就像它订阅“EVT_USER_CREATED”的 auth 服务一样。
客户/customer.controller.ts

  @EventPattern('EVT_USER_CREATED')
  public async handleUserCreated(data: any): Promise<AuthSubject> {    
    const customer = await this.customerService.create(data);
    return customer ;
  }

现在通过上述设置，微服务认证和客户将交替接收事件:如果用户微服务发出用户的创建，只有认证服务会对其作出 react 并从帽子用户创建认证主题。不会为该用户创建客户。对于在用户微服务中创建的下一个用户，只会创建一个客户，但不会创建身份验证主题。第三个创建的用户将再次被 auth 微服务消费，但不会被客户微服务消费。等等。

                                           -- auth microservice 
                                         /
user microservice --- message broker ---
                                         \
                                           -- customer microservice

总结一下:我如何实现图中所示的消息传递架构，这样我只需要一个 emit(...)调用 user.controller.ts并且，这样我在对 emit(...) 的订阅中重新获得了两个回复。称呼？

Answer 1

可能有点晚了，但我为 future 的用户留下了我的贡献。
对于这种强基于事件的架构，建议使用Kafka这样的消息代理，不同的服务可以订阅不同的主题，甚至不同的服务也可以监听同一个主题，这样你就可以使用react以不同的方式处理同一个事件。
Kafka 还提供了许多优势，当您想扩展微服务时将非常有用，此外，它还支持 Nest。
https://docs.nestjs.com/microservices/kafka
您应该记住的是，在使用消息传递系统时，通信通常是完全异步的，这是一个重要的细节，因为按照您提供的案例，不仅足以让您通过 kafka 将消息发送到另一个微服务以验证用户的凭据，但有必要向客户端返回响应。对于这种情况，nest 提供的工具是有限的，因为使用 @MessagePattern 装饰器，我们可以在 Controller 中接收来自 kafka 的消息，但我们不能等待响应(同一主题或其他主题)并确认成功响应到客户要求。在这种情况下，您可以方便地使用 Kafkajs ( https://kafka.js.org/ ) 创建自己的传输器或自己的 kafka 客户端，以便在必要时保留用户的请求，直到收到另一个主题的确认。
我在某些论坛中看到的一个解决方案是将一个对象保存在内存中(键/值)，其中包含与用户 ID 关联的请求的“响应”对象，这样您就可以通过 kafka 或任何方式发送消息具有特定操作的其他代理，并接收另一个 Controller 的确认，检索用户的“请求”对象，然后发送响应(Request.send(...))。这不是理想的解决方案，但它适用于某些情况。
例如:

@Controller('users')
class MyController {
    constructor(private kafkaClient: KafkaClient) {}
    private users: new Map<string, Request>() // Or private users: any = {}

    onModuleInit() {
        // By default, NestJs create topic "users-topic.reply"
        this.kafkaClient.subscribeToResponseOf('users-topic')
    }

    @Post('auth')
    authUser(@Req req: Request, @Res res: Response): void {
        const userData = req.body;
        const { id } = userData;

        // Save request object in private variable;
        this.users.set(id, res); // or this.users[id] = res;

        // Send kafka message to validate user data
        this.kafkaClient.emit('users-topic')
    }

    @MessagePattern('users-topic.reply')
    authUser(@Payload() data: any): any {
        if (data.message === 'success') {
            const res:Request = this.users.get(data.id); // or this.users[data.id];
            
            // Clean users object
            this.users.remove(data.id); // or this.users[data.id] = null;

            // Response client request
            res.send('Authentication successfully')
        }
    }
}