concat：如何使用特定的下一个函数独立处理每个订阅答案

【问题标题】：concat: how to handle each subscribe with a specific next function as independentconcat：如何使用特定的下一个函数独立处理每个订阅
【发布时间】：2016-11-27 02:10:01
【问题描述】：

我想在 concat observable 中独立处理每个订阅的响应。我在 angular2 中使用 Rxjs 5。

let source1 = //some Http service
let source2 = //some Http service
let source3 = //some Http service  

Observable.concat(source1,source2,source3)
      .subscribe(data=> console.log(data));

上面的 concat observable 工作正常，只要从每个源返回的数据是相同的类型，并且我必须以相同的方式处理所有 3 个源的返回数据。

但是，如果从不同服务生成承诺的来源以及订阅中返回的每个来源的数据不同并且需要唯一处理，该怎么办。这可能吗？以下是我所要求的伪代码。

let source1 = //some Http service1 returns a string
let source2 = //some Http service2 returns a number
let source3 = //some Http service3 returns a boolean  

//The following is pseudo code
Observable.concat(source1,source2,source3)
      .subscribe(data_source1=> console.log(data_source1))
      .subscribe(data_source2=> console.log(data_source2 +3))
      .subscribe(data_source3=> console.log(data_source3 === true));

附言显然，所有响应都需要按顺序处理，而不是并行处理。

【问题讨论】：

标签： javascript asynchronous rxjs reactive-programming rxjs5

【解决方案1】：

您可以使用forkJoin()，它会在所有 Observable 完成后发出。

但是，您说您想按顺序处理响应，所以我猜您想一个接一个地启动请求。

这仍然可以使用concat 解决，然后使用toArray() 将所有响应合并到一个数组中。

const Observable = Rx.Observable;

let source1 = Observable.of(42).do(() => console.log('source1')).delay(500);
let source2 = Observable.of(true).do(() => console.log('source2')).delay(500);
let source3 = Observable.of("Hello, World!").do(() => console.log('source3')).delay(500);

Observable.concat(source1, source2, source3)
    .toArray()
    .subscribe(data => console.log(data));

这会打印到控制台：

"source1"
"source2"
"source3"
[42, true, "Hello, World!"]

有多个 do() 运算符表明 Observables 是延迟启动的。

观看现场演示：https://jsbin.com/kikuceh/1/edit

【讨论】：