javascript - 为什么这些 RxJS observables 会产生奇怪的输出？

Question

我有两个从同一来源创建的可观察量。它们的区别在于映射将随机值分配给发出的元素的属性。以下是逻辑示例:

var Rx = require('rx');
var _ = require('lodash');

// create a source that emits a single event, and map that to an empty object
var source = Rx.Observable
    .range(0, 1)
    .map(function makeObject() { return {}; });

// map the empty object and give each one a type property with the
// value randomly chosen between "a" or "b"
var typed = source.map(function type(obj) {
    obj.type = _.sample(['a', 'b']); // obj.type will randomly be 'a' or 'b'
    return obj;
});

// create an observable that only contains "a"
var a = typed.filter(function(obj) {
    return obj.type === 'a';
});

// create an observable that only contains "b"
var b = typed.filter(function(obj) {
    return obj.type === 'b';
});

// merge both observables and log the result in the subscription
Rx.Observable.merge(a, b).subscribe(function(obj) {
    console.log(obj);
});

我希望这个最终的合并流将始终生成一个具有 obj.type === 'a' 的单个对象。或obj.type === 'b' ，然后完成。

但是，每次运行此脚本时，我都会得到各种结果，有些是预期的，有些是意外的。

预期结果“a”:

{ type : 'a' }

预期结果“b”:

{ type : 'b' }

两者都出乎意料:

{ type : 'a' }
{ type : 'b' }

而且，有时我根本没有任何输出。我在这里缺少什么？

Answer 1

该问题与 RX 的惰性性质有关:

您有两个由合并调用创建的订阅，每个订阅都会导致对所有可观察运算符的评估。

这意味着:

订阅 a -> 可能会导致:

• 生成项目 a，然后发出。
• 生成item b，然后过滤掉

订阅 b -> 相同，或者:

• 生成并发出项目 b
• 生成item a，然后过滤掉

如果合并这些流，您将得到以下结果之一: 只有a，只有b，a&b两者，两者都不是。

更多详情

让我们看一个更简单的例子:

 var source = Rx.Observable
    .range(0, 1)
    .map(function () { return Math.random(); })

现在在常规的发布-订阅系统中，我们希望如果我添加 2 个订阅者，每个订阅者都会输出相同的值:

source.subscribe(function(x){console.log("sub 1:" + x)})
source.subscribe(function(x){console.log("sub 2:" + x)})

只是它们不是，每个都会打印不同的值，因为每个订阅都会再次调用 Math.Random()。

虽然这有点奇怪，但这实际上是 rx observables 的正确行为，每个新的订阅都会导致对 observable 流的新评估。

合并订阅这两个可观察量(这意味着创建了两个值而不是一个)并将这些值发送到一个新的可观察量。

为了避免这种行为，我们可以使用 RX 的发布运算符。 这里有更详细的解释:

http://www.introtorx.com/content/v1.0.10621.0/14_HotAndColdObservables.html

所以，在这种情况下:

var source = Rx.Observable
    .range(0, 1)
    .map(function makeObject() { return {}; });

var typed = source.map(function type(obj) {
    obj.type = _.sample(['a', 'b']); // obj.type will randomly be 'a' or 'b'
    return obj;
}).replay().refCount();