RxJS 使用两个参数订阅

Question

我有两个要组合的可观察对象，并且在订阅中使用两个参数或只使用一个参数。我尝试了 .ForkJoin、.merge、.concat 但无法实现我正在寻找的行为。

例子：

obs1: Observable<int>;
obs2: Observable<Boolean>;

save(): Observable<any> {
   return obs1.concat(obs2);
}

那么当使用这个功能时：

service.save().subscribe((first, second) => {
    console.log(first); // int e.g. 1000
    console.log(second); // Boolean, e.g. true
});

或

service.save().subscribe((first) => {
    console.log(first); // int e.g. 1000
});

有没有可能得到这种行为？

希望有人能提供帮助！

编辑：

在我的具体用例中，obs1<int> 和 obs2<bool> 是两个不同的发布请求：obs1<int> 是实际的保存功能，obs2<bool> 检查是否有其他服务正在运行。

请求完成后需要obs1<int> 的值来重新加载页面，如果服务正在运行，则需要obs2<bool> 的值来显示消息 - 独立于obs1<int>。

所以如果obs2<bool> 在obs1<int> 之前发出，这不是问题，消息会在重新加载之前显示。但是如果obs1<int> 在obs2<bool> 之前发出，则页面会重新加载并且消息可能不再显示。

我之所以这么说是因为对于给定的答案，无论值是在其他可观察对象的 onComplete 之前还是之后发出，都会有不同的行为，这可能会影响用例。

Answer 1

有几个运算符可以做到这一点：

合并最新

这个操作符将结合两个 observables 发出的最新值，如弹珠图所示：

obs1: Observable<int>;
obs2: Observable<Boolean>;

save(): Observable<any> {
   return combineLatest(obs1, obs2);
}

save().subscribe((val1, val2) => { 
   // logic
});

压缩

Zip 运算符将等待两个可观察对象都发出值，然后再发出一个。

obs1: Observable<int>;
obs2: Observable<Boolean>;

save(): Observable<any> {
   return zip(obs1, obs2);
}

save().subscribe((vals) => { 
   // Note Vals = [val1, val2]
   // Logic
});

或者如果你想对数组使用解构

save().subscribe(([val1, val2]) => {
   // Logic
});

WithLatestFrom

WithLatestFrom 发出由可观察对象发出的最后一个值的组合，注意此运算符会跳过其他可观察对象中没有对应值的任何值。

save: obs1.pipe(withLatestFrom(secondSource))

save().subscribe(([val1, val2]) => {
    // Logic
});

Answer 2

您可以为此目的使用forkJoin。并行调用它们，然后如果它们中的任何一个存在，则执行某些操作。

let numberSource = Rx.Observable.of(100);
let booleanSource = Rx.Observable.of(true);

Rx.Observable.forkJoin(
  numberSource,
  booleanSource
).subscribe( ([numberResp, booleanResp]) => {
  if (numberResp) {
    console.log(numberResp);
    // do something
  } else if (booleanResp) {
    console.log(booleanResp);
    // do something
  }
});

Answer 3

您可以使用zip 静态方法代替concat 运算符。

save(): Observable<any> {
   return zip(obs1, obs2);
}

那么您应该能够执行以下操作：

service.save().subscribe((x) => {
    console.log(x[0]); // int e.g. 1000
    console.log(x[1]); // Boolean, e.g. true
});

Answer 4

要使用的确切运算符取决于您要解决的问题的具体细节。

一个有效的选项是使用combineLatest - Docs：

obs1$: Observable<int>;
obs2$: Observable<Boolean>;
combined$ = combineLatest(obs1$, obs2$);

combined$.subscribe(([obs1, obs2]) => {
  console.log(obs1);
  console.log(obs2);
})

Answer 5

Concat 通过流发出两个事件，一个接一个完成，这不是您想要的。

Merge 将以相同的方式发出这两个事件，但按照它们最终完成的顺序，也不是你所追求的。

您想要的是同一流事件中两个项目的值。 forkJoin 和 zip 和 combineLatest 将执行此操作，您遇到的问题是它们都会发出一个您在订阅中未正确访问的值的数组。

每次压缩在一起的所有项目按顺序发射时，zip都会发射，所以如果可观察的1发射1,2,3，而可观察的2发射4,5； zip 的排放量将是 [1,4]、[2,5]。

combineLatest 每次发射时都会发射，所以你会得到类似 [1,4],[2,4],[2,5],[3,5] 的东西（取决于确切的发射顺序）。

最后 forkJoin 只发出一次，一旦它里面的每个项目都实际完成，a 然后自己完成。这可能是您最想要的，因为您似乎在“节省”。如果这些示例流中的任何一个没有完成，forkJoin 将永远不会发出，但如果它们都在最终值之后完成，forkjoin 只会发出一个：[2,5]。我更喜欢这个，因为它是“最安全”的操作，因为它保证所有流都正确完成并且不会造成内存泄漏。通常在“保存”时，你只期望一个发射，所以它也更明确。当您看到 forkJoin 时，您就知道您正在处理单个发射流。

我个人会这样做：

obs1: Observable<int>;
obs2: Observable<Boolean>;

save(): Observable<any> {
   return forkJoin(obs1, obs2);
}

service.save().subscribe(([first, second]) => {
    console.log(first); // int e.g. 1000
    console.log(second); // Boolean, e.g. true
});

Typescript 提供类似这样的语法来访问已知长度数组中的项目，但无法在订阅成功函数中真正创建多个参数，因为它的接口只接受单个参数。