Rxjs 可观察，内部订阅需要很长时间，外部可观察在后续请求中首先执行

Question

我有一个启动 Observable 的按钮，然后我订阅结果并嵌套我有一个依赖于之前的 Observable 结果的订阅。

this.getData(params).subscribe(result => {
    // Make some first level process
    console.log('[Outter Execution]: ',result);

    this.getInnerData(result).subscribe(res => {
        // Make some inner level process
        console.log('[Inner Execution]: ',res);
    });
});

取决于用户的点击速度，顺序不一样：

//User click slowly

[Outer Execution]
[Inner Execution]

[Outer Execution]
[Inner Execution]

//User start clicking quickly

[Outer Execution]
[Inner Execution]

[Outer Execution]
[Inner Execution]

[Outer Execution]
[Outer Execution]
[Outer Execution]

[Inner Execution]
[Inner Execution]
[Inner Execution]

[Outer Execution]
[Inner Execution]
[Outer Execution]
[Inner Execution]

如您所见，如果嵌套订阅耗时较长，并且用户再次点击内订阅未解决，则在解决内执行之前，第一个[Outer Execution]消息被注销。一段时间后，解决了之前长时间的内部订阅，并返回了记录的消息。

我尝试使用 switchMap 和 mergeMap 运算符但没有成功。

[已编辑]：我需要的功能，是作为一个块执行，并且需要在第一次执行完成后执行下一个外部执行（外部和内部订阅）。

Answer 1

使用 RxJS 的方法有很多，如果没有具体的示例，很难为您提供具体的答案。我能做的最好的就是从我的代码中为您提供一个示例，并希望它接近您想要完成的目标。

我与他们的供应商有产品。当我得到一个产品时，它有一系列供应商 ID。所以我需要获取产品，然后在该操作中获取供应商。 （听起来和你的场景相似？）

这是我的代码：

  selectedProductSuppliers$ = this.selectedProduct$
    .pipe(
      filter(selectedProduct => Boolean(selectedProduct)),
      switchMap(selectedProduct =>
        forkJoin(selectedProduct.supplierIds.map(supplierId => this.http.get<Supplier>(`${this.suppliersUrl}/${supplierId}`)))
      ),
      tap(suppliers => console.log('product suppliers', JSON.stringify(suppliers)))
    );

此代码使用switchMap 来确保如果用户单击多次，它将切换到最新的选择。

然后它使用forkJoin 处理每个供应商 ID，发出 http get 请求以检索每个供应商数据并将它们加入一个发出的数组中。

返回值为Observable<Supplier[]>

你可以在这里找到一个完整的例子：https://github.com/DeborahK/Angular-RxJS

更新：

如果您将上述代码更改为使用concatMap 而不是switchMap，它将处理每个请求，等待一个完成后再执行下一个。

Answer 2

为确保处理等待，您需要在主 Observable 上使用concatMap，而不是在依赖的 Observable 上。否则不等待集合。

我在您的最新更改之前进行了分叉，但这是我想出的似乎可行的方法：

  private clickSubject = new Subject<number>();
  clickAction$ = this.clickSubject.asObservable();

  ngOnInit() {
    this.clickAction$
      .pipe(
        concatMap(value => this.mainProcess(value)
          .pipe(
            mergeMap(x => this.dependantProcess(x))
          )
        )
      )
      .subscribe();
  }

  onClick(value) {
    // Emits a value into the action stream
    this.clickSubject.next(value);
  }

  mainProcess(value) {
    console.log("[Emitted] Main", value);
    return of(value).pipe(delay(10));
  }

  dependantProcess(value) {
    console.log("[Emitted] Dependent", value);
    return of(value).pipe(delay(2000));
  }

注意concatMap是用来等待mainProcess的。

• 此代码通过使用 RxJS Subject 定义操作流来响应用户点击。

• 每次用户单击按钮时，都会发出操作流。

• 动作流管道使用concatMap缓存请求并等待处理，直到处理完之前的请求。

• 当主进程发出时，从属进程执行。

• 当管道完成（主进程及其依赖进程）后，下一个缓存请求将由concatMap 处理。

有意义吗？

你可以在这里找到更新的代码：

https://stackblitz.com/edit/angular-dependent-order-deborahk