如何在RxJS中实现async/await逻辑？附序列执行场景需求

需求 #

• 存在一个返回结果的Observable（obs1）
• 依赖该Observable的返回值，按顺序执行另外两个Observable（文件上传、用户信息更新）
• 待两个Observable都执行完成后，基于obs1的返回值返回最终结果

原代码问题 #

原代码使用map操作符，会直接同步返回结果，不会等待内部的put和patch请求完成，同时也无法保证这两个请求的执行顺序，导致业务逻辑不符合预期。

正确实现 #

使用concatMap订阅初始Observable，内部通过concat保证两个请求依次执行，待所有操作完成后返回初始结果：

updateAvatar(
  @Headers() headers: { authorization: string },
  @CurrentUserId() currentUserId: string,
  @UploadedFile() avatarFile: Express.Multer.File,
): Observable<{ avatarUrl: string }> {
  const obs1 = this.queue.send(
    PostCommands.uploadImage,
    new UploadImageCommandRequst(
      currentUserId,
      avatarFile.originalname,
      'avatars',
      true,
    ),
  );

  return obs1.pipe(
    concatMap(({ imageUploadUrl, imageUrl }) => {
      // 先执行文件上传请求
      const uploadObs = this.httpService.put(imageUploadUrl, avatarFile.buffer);
      // 再执行用户信息更新请求
      const updateProfileObs = this.httpService.patch(
        `${this.config.GATEWAY_URL}/user/profile`,
        { avatarUrl: imageUrl },
        { headers: { authorization: headers.authorization } },
      );

      // 用concat保证顺序执行，最后映射回需要的结果
      return concat(uploadObs, updateProfileObs).pipe(
        // 取最后一个请求的完成信号，确保所有操作都已完成
        last(),
        map(() => ({ avatarUrl: imageUrl }))
      );
    })
  );
}

关键说明 #

• concatMap：订阅初始Observable，将其结果映射为新的Observable序列，保证内部Observable依次执行（无并发）
• concat：严格按传入顺序执行Observable，前一个完成后才触发下一个
• last()：忽略中间请求结果，仅取最后一个请求的完成信号，确保所有操作都已完成
• 最终返回的imageUrl来自初始Observable的结果，符合需求要求

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