如何同步 RxJS 更新以便中间值不会通过流传递？

Question

在我的系统中，我有一个源，两个将源映射到新值的“步骤”，然后是组合这两个步骤以创建最终值的总和。该系统的初始运行如我所愿，生成了 3 的单个总和。

\n\n

var source = new Rx.BehaviorSubject(0);    \nvar stepOne = source.map(function (value) {\n    return value + 1;\n});\nvar stepTwo = source.map(function (value) {\n    return value + 2;\n});    \nvar sum = Rx.Observable.combineLatest(\n    stepOne,\n    stepTwo,\n    function (s1, s2) {\n        console.log(\'calc sum: \' + (s1 + s2));\n        return s1 + s2;\n    }).subscribe(function (sum) {\n    });\n

\n\n

输出：

\n\n

> calc sum: 3\n

\n\n

但是，如果我为源输入一个新值，我会得到如下两个结果：

\n\n

source.onNext(1);\n\n> calc sum: 4\n> calc sum: 5\n

\n\n

第一个是新的源值通过系统的一部分时的中间结果\xe2\x80\xa6，然后当所有值完成传播时得到最终结果。

\n\n

所以我的问题是，推荐的配置方式是什么，以便推送到源中的新值将原子地通过系统并仅生成一个总和结果？

\n\n

谢谢！

\n

Answer 1

这就是combineLatest工作原理，它确实令人困惑，因为正如您所指出的，它允许这些暂时不一致的状态。需要学习的关键combineLatest是，每当它的任何一个源发出一个新项目时，它都会发出一个新项目，并且它是部分这样做的，它没有任何类型的“等待”机制。

在图中，http://rxmarbles.com/#combineLatest。

您可能想要的是zip运算符。Zip 等待其输入发出相互匹配的项目。换句话说，一旦所有输入中的所有第 n 个项目都已发出，zip 的输出就会发出其第 n 个项目。它非常适合这种菱形情况，您已经source生成stepOne和 ，stepTwo并且想要组合stepOne和stepTwo。

在图中，http://rxmarbles.com/#zip。

请记住，zip假设输入具有相同的发射频率。在其他情况下，当步骤一和步骤二中的项目具有不同的发射频率时，您可能希望将它们组合起来。那么你需要使用combineLatest.