如何避免 Rx 中的故障答案

【问题标题】：how to avoid glitches in Rx如何避免 Rx 中的故障
【发布时间】：2014-03-11 17:39:55
【问题描述】：

与其他“FRP”库不同，Rx 不能防止故障：使用时间不匹配的数据调用回调。有没有解决这个问题的好方法？

例如，假设我们从单个流派生了一系列昂贵的计算（例如，我们在下面进行排序或 ajax 获取，而不是 _.identity）。我们做 distinctUntilChanged 以避免重新计算昂贵的东西。

sub = new Rx.Subject();
a = sub.distinctUntilChanged().share();
b = a.select(_.identity).distinctUntilChanged().share();
c = b.select(_.identity).distinctUntilChanged();
d = Rx.Observable.combineLatest(a, b, c, function () { return _.toArray(arguments); });
d.subscribe(console.log.bind(console));
sub.onNext('a');
sub.onNext('b');

第二个事件最终会导致一些故障状态：我们输出了三个事件，而不是一个，这会浪费大量 cpu，并且需要我们明确地解决不匹配的数据。

这个特殊的例子可以通过删除 distinctUntilChanged 来解决，如果输入没有改变，可以编写一些古怪的 scan() 函数来传递先前的结果。然后你可以压缩结果，而不是使用 combineLatest。这很笨拙，但可行。

但是，如果任何地方都存在异步，例如ajax 调用，然后 zip 不起作用：ajax 调用将同步（如果缓存）或异步完成，因此您不能使用 zip。

编辑

尝试用一个更简单的例子来阐明期望的行为：

你有两个流，a 和 b。 b 取决于 a。 b 是异步的，但浏览器可能会缓存它，因此它可以独立于 a 更新，也可以与 a 同时更新。因此，浏览器中的特定事件可能会导致以下三种情况之一：更新； b 更新； a 和 b 都更新。期望的行为是在所有三种情况下都只调用一次回调（例如渲染方法）。

zip 不起作用，因为当 a 或 b 单独触发时，我们不会从 zip 中得到回调。 combineLatest 不起作用，因为当 a 和 b 同时触发时，我们会得到两个回调。

【问题讨论】：

你到底想在这里实现什么？从技术上讲，这正是它所要做的。你的预期结果是什么？为什么您认为 zip 不适用于同步与异步？这与 zip 如何使用事件无关，除非您订阅了一个 hot observable，在这种情况下您可能需要重放最后一个值。
在 FRP 系统中（参见 flajax、Elliot 的论文等），您将得到一个事件，一个事件输入，并且没有时间不匹配的数据。第二个 onNext 将触发 console.log 一次，而不是三次，并且 a、b 和 c 在第二个事件上都将是“b”。 zip 不适用于异步，因为它等待来自所有流的值。如果只有一个更新，则 zip 不输出任何内容。
仅供参考，这里也是在 Bacon.js 中实现的。没有故障。 jsfiddle.net/DeUeh
OP，我建议使用油门并忽略 a 如果 b 很快回来（比如在 10 毫秒内）。所以a.merge(b).throttle(10)。或者，您可以使用bufferWithTime，如果您想将a 的事件与b 的事件展平以形成单个对象。
但是根据定义，绝对没有办法说“如果 b 将被缓存，则不要使用 a”，直到 b 已经返回，或者如果您通过某些外部资源知道 b 肯定会被缓存。这不是 Rx 中的故障。这是不确定性的本质。

标签： javascript reactive-programming reactive-extensions-js rxjs

【解决方案1】：

概念

a 和 b 都更新

a 和 b 都是可观察的，在 Rx 中不作为原语存在。

没有可以定义的无损通用运算符来决定它何时收到来自a 的通知是应该将它传递到下游还是推迟到它收到来自b 的通知。 Rx 中的通知本身并不携带“两种”语义，或任何超出 Rx 语法的语义。

此外，Rx 的串行合同防止操作员利用重叠通知来实现此目标。（尽管我怀疑依赖竞争条件并不是您想要的方法。）

请参阅Rx Design Guidelines 中的 §§4.2、6.7。

因此，我上面所说的“没有可以定义的无损通用运算符...”的意思是，给定两个带有独立通知的 observable a 和 b，任何尝试决定何时执行的运算符接收来自a 或b 的通知是否必须立即推送或等待“其他”值，必须依赖于任意时间。这是猜测。所以这个假设的运算符必须要么丢弃值（例如，DistinctUntilChanged 或 Throttle），要么丢弃时间（例如，Zip 或 Buffer），尽管可能两者兼而有之。

因此，如果代理有能力单独推送a，或者单独推送b，或者a和b一起作为通知单元，那么开发者有责任具体化这个通知单位自己。

需要一个三态类型：a |乙 | {a,b}

（请原谅我糟糕的 JS）

var ab = function(a, b) { this.a = a; this.b = b; }
sub.onNext(new ab('a'));        // process a alone
sub.onNext(new ab('a', 'b'));   // process a and b together
sub.onNext(new ab(null, 'c'));  // process c alone

observable 查询的形状不再重要。必须定义观察者以接受此数据类型。生成器有责任根据其内部状态的语义应用任何必要的缓冲或计时计算，以便为其观察者生成正确通知。

顺便说一句，感谢您在编辑中提供了一个简单的解释（无论如何我似乎很清楚）。我第一次听说this Rx forum discussion 中的“故障”。如您所见，它从未真正结束。现在我想知道那个OP的问题是否真的像这样简单，当然，假设我已经正确理解了你的问题。 :-)

更新：

这是另一个相关的讨论，包括我对为什么 Rx 不是 FRP 的一些想法：

https://social.msdn.microsoft.com/Forums/en-US/bc2c4b71-c97b-428e-ad71-324055a3cd03/another-discussion-on-glitches-and-rx?forum=rx

【讨论】：