如何将 Observable 数组与 RxJS 6.x 和 Node.js 合并？

Question

出于学习目的，我正在创建一个 Node 应用程序，该应用程序需要从一个数组中获取 x 个 RxJS observables 并组合成一个事件流。我想知道事件何时以任何可观察的方式发生，以任何顺序（不是以任何顺序或完全完成）。我觉得它应该在一个合并的事件流中。基本上，来自任何可观察对象的第一个事件将完成。

为此，我觉得 merge() 可以解决问题。由于merge 不直接将数组作为参数，因此到目前为止我使用reduce 来帮助合并。

然而，最终的结果不是一个可观察的，而是一个函数。我也订阅不了 代码的简化版本可以在下面看到。

我怎样才能改变这个 Node 10.14.2, RxJS 6.4.x 代码来返回一个 observable 而不是我可以添加 .subscribe() 的“[function]”？

const { Observable } = require('rxjs');
const { merge } = require('rxjs/operators');

const observables = [
    Observable.create(observer => observer.next('Hello')),
    Observable.create(observer => observer.next('Hello')),
    Observable.create(observer => observer.next('Hello'))
];

const mergedObservables = observables.reduce((merged, observable) => {
    console.log(observable);
    return merge(merged, observable);
});

// outputs:
// Observable { _isScalar: false, _subscribe: [Function] }
// Observable { _isScalar: false, _subscribe: [Function] }

console.log(mergedObservables);

// outputs:
// [Function]

mergedObservables.subscribe();
// error:
// TypeError: mergedObservables.subscribe is not a function

Answer 1

编辑：您正在导入merge 运算符而不是静态merge函数。前者对来自源 observable 的事件进行操作，而后者从一个或多个源 observable 创建一个新的 observable。虽然考虑下面的传播语法简化了您的代码，但这并不是您真正的问题。

看起来Node.js >= 5.0支持函数调用中数组的展开运算符（不过，您没有指定您使用的是哪个版本的 Node.js）。如果您使用的是现代版本的 Node.js，以下内容应该可以工作：

const { Observable, merge } = require('rxjs')

const observables = [
    Observable.create(observer => observer.next('Hello')),
    Observable.create(observer => observer.next('Hello')),
    Observable.create(observer => observer.next('Hello'))
]

const mergedObservables = merge(...observables)
mergedObservables.subscribe(event => { console.log(event) })