rxjs - 如何组合多个 rxjs BehaviourSubjects

Question

我正在构建一个 Angular2 应用程序，并且有两个 BehaviourSubjects我想在逻辑上合并为一个订阅。我正在发出两个 http 请求，并希望在它们都返回时触发一个事件。我在看 forkJoin对比 combineLatest .似乎在更新任何一个 behvaviorSubjects 时 combineLatest 会触发，而 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);
                    }
                })
            });
    }
}

这是我的组件订阅的片段:

  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);
                }
            });