在NestJS+Neo4J中，如何让RxJS流操作全完成后再执行下一步？

我在NestJS项目中结合使用RxJS与Neo4J，要求每一步操作必须完全完成后，下一步才能成功执行，因此希望为以下代码中的各步骤实现类似Promise.all的效果。但问题在于，tap操作符似乎无法等待所有Promise完成，我不确定该如何实现这一点。

loadBooks() {
    const booksObservable =
      this.booksService.getBooksFromAPI();

      booksObservable
      .pipe(
        mergeMap((response) => response.data),
        tap((protocol) => {
          return from(
            this.booksService.find(book.name).then((bookNode) => {
              if (!bookNode) {
                return from(
                  this.bookService.create({
                    name: book.name
                  }),
                );
              }
            }),
          );
        }),
        groupBy((book) => book.category),
        tap((categoryName) => {
          return from(
            this.bookCategoryService
              .find(categoryName.key)
              .then((categoryNode) => {
                if (!categoryNode) {
                  return from(
                    this.bookCategoryService.create({
                      name: categoryName.key,
                    }),
                  );
                }
              }),
          );
        }),
        mergeMap((group) => group),
        tap((book) => {
          let bookId: string;
          let categoryId: string;
          return from(
            this.bookService
              .find(book.name)
              .then(async (bookNode) => {
                if (!bookNode) {
                  throw new NotFoundError(
                    `Could not find book by name ${book.name}`,
                  );
                }
                bookId = bookNode.getId();
                await this.bookCategoryService
                  .find(book.category)
                  .then(async (categoryNode) => {
                    if (!categoryNode) {
                      throw new NotFoundError(
                        `Could not find category by name ${protocol.category}`,
                      );
                    }
                    categoryId = categoryNode.getId();
                    await this.bookService.relateToCategory(
                      bookId,
                      categoryId,
                    );
                  });
              }),
          );
        }),
      )
      .subscribe(() => {
        return 'done';
      });

如何才能让tap中的操作针对流中的每个项目都完全完成后，再执行管道中的下一个函数？

tap操作符仅用于执行副作用，不会改变流的传递逻辑，也不会等待内部的Promise/Observable完成。要实现串行等待的效果，必须用mergeMap/switchMap/exhaustMap这类映射操作符——它们会将异步操作转换成流的一部分，等待异步操作完成后再把值传递到下一个操作符。

具体修改步骤： #

1. 替换所有tap为mergeMap：每个需要等待的异步逻辑都用mergeMap包裹，确保异步操作完成后再执行下一步。
2. 修正代码变量错误：比如第一个tap里的protocol参数应为book，最后一个tap里的protocol.category应为book.category。
3. 简化异步逻辑：把嵌套的then/await改成清晰的async/await写法，避免嵌套层级过深。

修改后的代码如下：

loadBooks() {
  const booksObservable = this.booksService.getBooksFromAPI();

  booksObservable
    .pipe(
      mergeMap((response) => response.data),
      // 替换tap为mergeMap，等待书籍查询/创建完成
      mergeMap(async (book) => {
        const bookNode = await this.booksService.find(book.name);
        if (!bookNode) {
          await this.bookService.create({ name: book.name });
        }
        return book; // 传递book到下一个操作符
      }),
      groupBy((book) => book.category),
      // 替换tap为mergeMap，等待分类查询/创建完成
      mergeMap(async (categoryGroup) => {
        const categoryNode = await this.bookCategoryService.find(categoryGroup.key);
        if (!categoryNode) {
          await this.bookCategoryService.create({ name: categoryGroup.key });
        }
        return categoryGroup; // 传递分组到下一个操作符
      }),
      mergeMap((group) => group),
      // 替换tap为mergeMap，等待关联操作完成
      mergeMap(async (book) => {
        const bookNode = await this.bookService.find(book.name);
        if (!bookNode) {
          throw new NotFoundError(`Could not find book by name ${book.name}`);
        }
        const bookId = bookNode.getId();

        const categoryNode = await this.bookCategoryService.find(book.category);
        if (!categoryNode) {
          throw new NotFoundError(`Could not find category by name ${book.category}`);
        }
        const categoryId = categoryNode.getId();

        await this.bookService.relateToCategory(bookId, categoryId);
        return book;
      }),
    )
    .subscribe({
      next: () => console.log('单个项目处理完成'),
      complete: () => console.log('所有书籍处理完成'),
      error: (err) => console.error('处理出错:', err),
    });
}

关键说明： #

• mergeMap的核心作用：订阅内部的Promise/Observable，等待其完成后将结果传递到下一个操作符，确保每一步异步操作完成后再执行后续逻辑。
• async/await简化：RxJS会自动将async函数转换成Observable，无需手动用from()包裹Promise，代码更简洁易读。
• 保持流的传递：每个mergeMap必须返回当前处理的对象（如book、categoryGroup），确保后续操作符能拿到对应数据继续处理。

内容的提问来源于stack exchange，提问作者duck degen