计算链中的其他运算符，而不是 combineLatest 以避免冗余计算

Question

我已经使用 RxJS 成功地将更大的 Excel 计算迁移到 JS。当任何 Excel 公式使用多个输入时，任何输入数据都表示为 Observable，随后的计算通过 .map 和 .combineLatest 实现。

这很好用，除了一个问题。这是一个简化的示例：

三个输入（a$=1、b$=2、c$=3）用于两个不同的计算（第一个是$ab = $a+$b = 3，第二个是$bc = $b+$c = 5）作为计算最终结果的中间步骤$abbc = $ab + $bc = 8 .

当 $b 现在更新/发出一个新的输入值（例如 4）时，$abbc 被计算两次 - 第一次是在更新 $ab 时（导致错误的结果 $abbc=10），然后是在 $bc 是更新，得到正确的结果12。

虽然最终结果是正确的，但中间计算既错误又多余。有没有办法只在$b 更新时执行最后一次计算 - 同时在a$ 或c$ 更新时仍然更新计算（这将排除 zip 运算符）。我知道这个示例显然可以简化为省略中间步骤并直接从$a、$b 和$c 计算$abbc - 但在真实的示例中这是不可能/可行的。

这是 JSbin 中的运行示例：https://jsbin.com/pipiyodixa/edit?js,console

Answer 1

这里的问题是 RxJS 的行为在设计上是正确的。

它确实应该首先更新b => ab => abbc 然后bc => abbc。它处理流中的值。

你想要的是处理“层”中的值。a、b、c，然后是ac、bc，然后计算最终值abbc。

我能想到的唯一方法是利用 JavaScript 执行上下文和 setTimeout(() => {}, 0) 的技巧。这样你就不会安排任何超时（实际上是真正的timeout might be > 0），而只是在 JavaScript 完成当前执行后在另一个执行上下文中运行闭包。

不好的是，为了避免多次重新发出值，您需要缓存更多（因为merge()）：

var b2$ = b$.cache(1);

var ab$ = a$
  .combineLatest(b2$, (a, b) => a + b)
  .do(x => console.log('$a + $b = ' + x))
  .cache(1);

var bc$ = b2$
  .combineLatest(c$, (b, c) => b + c)
  .do(x => console.log('$b + $c = ' + x))
  .cache(1);

var abbc$ = new Rx.Observable.merge(ab$, bc$)
  .auditTime(0)
  .withLatestFrom(ab$, bc$, (_, ab, bc) => ab + bc)
  .do(x => console.log('$ab + $bc = ' + x));

console.log("Initial subscription:")
abbc$.subscribe();

b$.next(4);

auditTime() 运算符在这里是最重要的。它触发withLatestFrom() 更新它的值，而当它第一次被ab$ 或bc$ 触发时，它会忽略所有连续的发射，直到这个闭包执行结束（这就是setTimeout() 技巧）。

观看现场演示：https://jsbin.com/zoferid/edit?js,console

另外，如果您添加a$.next(5);，最终计算只执行一次（可能是好是坏:)）。

我不知道这是否能解决你的问题，因为正如我所说，RxJS 就是这样工作的。我还没有在任何更复杂的示例上对其进行测试，所以也许这不是您最终可以使用的方式。

请注意，cache() 运算符已在 RC.1 中删除，目前没有替代品：https://github.com/ReactiveX/rxjs/blob/master/CHANGELOG.md

Answer 2

基于@martin 的（正确）答案，我创建了一个执行 combineLatestDelayed 的 rxjs 运算符：

Rx.Observable.prototype.combineLatestDelayed = function(b$, cb) {
  var a2$ = this.cache(1);
  var b2$ = b$.cache(1);
  return a2$
    .merge(b2$)
    .auditTime(0)
    .withLatestFrom(a2$, b2$, (_, a, b) => cb(a,b));
}

这样，我只需要将原来的.combineLatest调用替换为.combineLatestDelayed即可：

var a$ = new Rx.BehaviorSubject(1).do(x => console.log("Emitting $a=" + x)); 
var b$ = new Rx.BehaviorSubject(2).do(x => console.log("Emitting $b=" + x)); var b1$ = b$.cache(1);
var c$ = new Rx.BehaviorSubject(3).do(x => console.log("Emitting $c=" + x));

var ab$ = a$
  .combineLatestDelayed(b1$, (a, b) => a + b)
  .do(x => console.log('ab$: $a + $b = ' + x))

var bc$ = b1$
  .combineLatestDelayed(c$, (b, c) => b + c)
  .do(x => console.log('bc$: $b + $c = ' + x))

var abbc$ = ab$
  .combineLatestDelayed(bc$, (ab, bc) => ab + bc)
  .do(x => console.log('$abbc: $ab + $bc = ' + x));

console.log("Initial subscription:")
abbc$.subscribe();

setTimeout(() => b$.next(4), 100);

完整的JS Bin here