如何组合多个 rxjs BehaviourSubjects

Question

我正在构建一个 Angular2 应用程序，并且有两个 BehaviourSubjects 我想在逻辑上合并为一个订阅。我正在发出两个 http 请求，并希望在它们都回来时触发一个事件。我正在查看 forkJoin 与 combineLatest。似乎 combineLatest 会在任何一个 behvaviorSubjects 更新时触发，而 forkJoin 只会在所有 behavoirSubjects 都更新后触发。它是否正确？必须有一个普遍接受的模式，不是吗？

编辑
这是我的 angular2 组件订阅的其中一个行为主题的示例：

export class CpmService {

    public cpmSubject: BehaviorSubject<Cpm[]>;

    constructor(private _http: Http) {
        this.cpmSubject = new BehaviorSubject<Cpm[]>(new Array<Cpm>());
    }

    getCpm(id: number): void {
        let params: URLSearchParams = new URLSearchParams();
        params.set('Id', id.toString());

        this._http.get('a/Url/Here', { search: params })
            .map(response => <Cpm>response.json())
            .subscribe(_cpm => {
                this.cpmSubject.subscribe(cpmList => {
                    //double check we dont already have the cpm in the observable, if we dont have it, push it and call next to propigate new cpmlist everywheres
                    if (! (cpmList.filter((cpm: Cpm) => cpm.id === _cpm.id).length > 0) ) {
                        cpmList.push(_cpm);
                        this.cpmSubject.next(cpmList);
                    }
                })
            });
    }
}

这是我的组件订阅的 sn-p：

  this._cpmService.cpmSubject.subscribe(cpmList => {
      doSomeWork();
  });

但我不想在单个订阅上触发 doSomeWork()，我只想在 cpmSubject 和 fooSubject 触发时触发 doSomeWork()。

Answer 1

您可以使用zip-operator，其工作方式类似于 combineLatest 或 forkJoin，但仅在两个流都发出时触发：http://reactivex.io/documentation/operators/zip.html

zip 和 combineLatest 的区别是： Zip 只会触发“并行”，而combineLatest 将触发任何更新并发出每个流的最新值。 因此，假设以下 2 个流：

streamA => 1--2--3
streamB => 10-20-30

zip:

• “1, 10”
• “2, 20”
• “3, 30”

combineLatest:

• “1, 10”
• “2, 10”
• “2, 20”
• “3, 20”
• “3, 30”

这也是一个活生生的例子：

const a = new Rx.Subject();
const b = new Rx.Subject();

Rx.Observable.zip(a,b)
  .subscribe(x => console.log("zip: " + x.join(", ")));
Rx.Observable.combineLatest(a,b)
  .subscribe(x => console.log("combineLatest: " + x.join(", ")));

a.next(1);
b.next(10);
a.next(2);
b.next(20);
a.next(3);
b.next(30);

<script src="https://unpkg.com/rxjs/bundles/Rx.min.js"></script>

---

还有一个旁注：永远不要在订阅内订阅。 改为这样做：

this._http.get('a/Url/Here', { search: params })
            .map(response => <Cpm>response.json())
            .withLatestFrom(this.cpmSubject)
            .subscribe([_cpm, cpmList] => {
                if (! (cpmList.filter((cpm: Cpm) => cpm.id === _cpm.id).length > 0) ) {
                    cpmList.push(_cpm);
                    this.cpmSubject.next(cpmList);
                }
            });