代码之家 › 专栏 › 技术社区 › JSmith

可以避免嵌套订阅吗?

rxjs6 rxjs angular

JSmith · 技术社区 · 4 年前

我用的是棱角和 rxJS 我是新来的 rxJS 操作员无法找到将所有内容都保存在流中的方法。

我的流很长,所以我会直截了当地解释我正在处理的问题

我需要跑帕拉莱尔 Observables 它可以被多次触发,并以相同的方式处理它们的数据,但它们的订阅需要按照一定的顺序进行。例如

    firstObservable$.pipe(
      map(() => secondObservable$),//this will only be triggered once
      tap(value => doSomething(value)),//data is processed once following the stream
      mergeMap(() => ThirdObservable$),//this one will be triggered multiple times but is subscribed in a particular order
      tap(value => doAnotherThing(value)),//how can I attach the processing og data to the observable?
      mergeMap(() => FourthObservable$),//this one will be triggered multiple times but is subscribed in a particular order
      tap(value => andAnother(value)),//how can I attach the processing og data to the observable?
      map(() => FifthObservable$),//this will only be triggered once
      tap(value => again(value))//data is processed once following the stream
    ).subscribe()

现在我的问题是,如果 ThirdObservable$ 第二次触发时,它将继续流的其余部分并调用 FourthObservable$ 和 FifthObservable$ 我想要相当于

    firtsObservable$.subscribe( // triggered Once
      (value) => secondObservable$.subscribe( // triggered Once
        (secondValue) => {
          processSecondValueOnce(secondValue)
          ThirdObservable$.subscribe(thirdValue => process(thirdValue)) // this can be triggered multiple times
          fourthObservable$.subscribe(fourthValue => process(fourthValue)) // this can be triggered multiple times
          fifthObservable$.subscribe( // triggered Once
            (fifthValue) => {
              process(fifthValue)
            }
          )
        }
      )
    )

0 回复 | 直到 4 年前

Picci 4 年前

如果我没理解错你的问题,你可能想试试这样的

firstObservable$.pipe(
  map(() => secondObservable$),
  tap(value_1 => doSomething(value_1)),
  mergeMap(() => {
     // create 3 separate observables, each doing its specific processing
     third = ThirdObservable$.pipe(tap(value => doAnotherThing(value));
     fourth = FourthObservable$.pipe(tap(value => andAnother(value));
     fifth = FifthObservable$).pipe(tap(value => again(value));
     // return the merge of the 3 observables
     return merge(third, fourth, fifth);
  })
).subscribe()

虽然我不明白,但我有一个观点。在第一 map 运算符返回Observable( secondObservable$ ). 这意味着我调用的变量 value_1 将包含一个Observable,因此 doSomething 必须是一个需要Observable作为输入的函数。这是你真正想要的吗?

通常,在这种情况下,您希望使用“高阶”运算符,如 switchMap , mergeMap , concatMap 或 exaustMap 这些运算符期望一个返回Observable的函数作为输入,并将返回的Observable“展开”。也许这也是你想要的。