如何有效地链接 RxJS Observables答案

【问题标题】：How to I effectively chain RxJS Observables如何有效地链接 RxJS Observables
【发布时间】：2020-08-07 21:41:12
【问题描述】：

我知道这个问题已经在这里以各种形式被问过几次，但我正在努力让它完全按照我的意愿去做。

为了提供一些上下文，我在 NodeJS 中有一个文件上传服务，它将图像存储在一个临时文件夹中。然后我有一些功能可以将图像调整为常规网络大小和缩略图，然后保存到上传文件夹，然后使用另一个功能来删除临时文件。所有这三个函数都返回 Observables。

我想要的是按顺序运行这些可观察对象，例如

observableA().subscribe((result) => //do something here ) // Resize the photo
observableB().subscribe((result) => //do something here ) // Create a thumbnail
observableC().subscribe((result) => //do something here ) // Delete the temporary file

我想确保 A 和 B 在执行 C 之前完成（删除临时文件夹）。我不想嵌套 Observables。 A 和 B 都返回一些我想要的数据。

哪个是实现这一目标的最佳 rxJS 运算符？

【问题讨论】：

使用rxjs的combine操作符
检查concat()运营商。

标签： javascript node.js typescript rxjs observable

【解决方案1】：

您可以使用forkJoin 组合前 2 个可观察对象（它们将并行运行）。然后使用switchMap 切换到第三个 observable。

import { of, timer, forkJoin } from "rxjs";
import { switchMap, delay, map } from "rxjs/operators";

const observableA = of("resized photo").pipe(delay(300));
const observableB = of("thumbnail").pipe(delay(250));
const observableC = of("temp deleted").pipe(delay(100));

const result = forkJoin(observableA, observableB).pipe(
  switchMap(([resultA, resultB]) => {
    console.log(resultA, resultB);
    console.log('now to observableC');
    return observableC.pipe(map(resultC => [resultA, resultB, resultC]));
  })
);
result.subscribe(x => console.log(x));

在stakblitz 上查看（别忘了打开控制台查看console.log 输出）。

RxJS 文档也有一个方便的页面，可帮助您为工作选择正确的运算符：operator-decision-tree

【讨论】：

【解决方案2】：

您应该使用flatMap，在某种意义上then 用于可观察对象，记住flatMap 实际上执行了一个功能。

this._myService.do1()
  .flatMap(function(x){return functionReturningObservableOrPromise(x)})
  .flatMap(... return observable here tooo)
  .subscribe(get your data)

【讨论】：

尝试此操作时出现此错误：“Observable”类型上不存在属性“flatMap”。
@RobBailey 试试mergeMap
你必须从import 'rxjs/add/operator/mergeMap'导入它

【解决方案3】：

要保持顺序，您应该使用 concatMap 运算符，并使用 tap 运算符来实现如下所示的副作用：

observableA()
  .pipe(
    concatMap(() => observable().pipe(tap(() => /* add some sides effects for this obs. */ )),
    tap(() => /* or add some sides effects here or somewhere else */ )),
    concatMap(() => observableC())
  )
  .subscribe();

【讨论】：