如何强制 observables 按顺序执行？

Posted 2023-02-23

【中文标题】如何强制 observables 按顺序执行？

【英文标题】：How to force observables to execute in sequence?

【发布时间】：2017-09-06 06:53:28

【问题描述】：

我正在从 Promise 世界转移到 Observable 世界。我通常用 Promise 做的一件事就是把一系列任务串联起来，让它们按顺序运行。例如，我有三个任务：printLog1() 打印 1 到控制台，printLog23() 打印 2 和 3 到控制台，printLog4() 打印 4。

当我想打印 1-2-3-4 时，我会写一个类似的承诺链

printLog1()
  .then(() => 
    printLog23();
  )
  .then(() => 
    printLog4();
  );

现在我想要与 Observable 相同的功能，我可以将 printLog() 函数重写为 Observable 之类的

printLog1 = Rx.Observabale.of(1).map((i) => console.log(i));
printLog23 = Rx.Observabale.of(2, 3).map((i) => console.log(i));
printLog4 = Rx.Observabale.of(4).map((i) => console.log(i));

然后我有三个向控制台发出不同值的可观察对象。我如何链接它们，以便这三个可观察对象按顺序运行并打印1-2-3-4？

【参考方案1】：

如果您想确保发射的顺序与您指定源 Observable 的顺序相同，您可以使用 concat 或 concatMap 运算符。

concat* 运算符仅在前一个 Observable 完成后订阅 Observable（它也适用于 Promises，请参阅 http://reactivex.io/rxjs/class/es6/MiscJSDoc.js~ObservableInputDoc.html）。

在您的情况下，它看起来如下所示：

import  concat  from 'rxjs'; // Note, concat from 'rxjs', is not the same as concat from 'rxjs/operators'

concat(printLog1, printLog23, printLog4);

... 或 concatMap 如果对一个 Promise 的请求取决于前一个 Promise 的响应：

printLog1.pipe(
  concatMap(response => ...),
  concatMap(response => ...),
);

...或者当订单无关紧要时，您可以使用 merge 立即订阅所有 Observables/Promises 并在它们到达时重新发送它们的结果：

merge(printLog1, printLog23, printLog4);

2019 年 1 月：针对 RxJS 6 更新

【讨论】：

我按照const $printLog1 = from(someFunc1Async()) 和const $printLog2 = from(someFunc2Async()) 测试了这个方法，然后尝试使用concat 失败。上面带有from 的分配会立即引起调用。也许很明显，但希望对其他人有所帮助。

你好！谢谢！我在 2 个 observables 上做了merge，我得到了两个响应，但我不知道哪个是哪个。有什么办法吗？

我不认为这个例子中管道中的concatMap 是绝对必要的——可以使用mergeMap 来获得相同的结果。每次调用printLog1 时，它都会返回一个新的可观察对象，并且每个后续映射运算符都将是一个新调用——因此不会有要取消的内部可观察对象。如果地图位于主题的pipe 中，则使用concatMap 是有意义的 - 并且正在调用printLogSubject.next() 来启动序列