用于可观察链接的仓促 forkjoin 替代 rxjs？答案

【问题标题】：Hasty forkjoin alternative rxjs for observable chaining?用于可观察链接的仓促 forkjoin 替代 rxjs？
【发布时间】：2020-04-26 06:03:19
【问题描述】：

我有 5 个不同的 API 调用要进行，它们现在都链接在 forkJoin 中。我的新要求是订阅应该在任何新的 observable 解决时触发。

在 rxjs 中是否有任何操作符或任何其他技巧可以让我保持链接，但是每次任何可观察到的解决方案都应该触发它？

forkJoin(
        this.myService.api1(),
        this.myService.api2(),
        this.myService.api3(),
        this.myService.api4(),
        this.myService.api5()
    )
        .subscribe(
            ([r1,r2,r3,r4,r5]) => { ... do something })

【问题讨论】：

你的意思是你需要像CombineLatest这样的东西吗？
你所说的“链接”具体是什么意思。顺序重要吗？这些调用是否以任何方式相互依赖？
我尝试了 combineLatest，当我尝试登录订阅方法时它只记录一次
不，顺序无关紧要。他们不依赖。我可以为每种方法单独订阅，但这只是增加了额外的代码行。
你可以试试merge。

标签： angular rxjs rxjs5

【解决方案1】：

您可以使用merge 像 forkJoin 一样同时执行您的 observable，但会立即发出它们的值。要跟踪订单，请使用map 将可观察对象的索引添加到其输出中。使用scan 跟踪以前的值，在数组中的正确位置插入当前值并发出累积的数据。

export function forkJoinEarly(...sources: Observable<any>[]): Observable<any[]> {
  return merge(...sources.map((obs, index) => obs.pipe(
    // optional: only emit last value like forkJoin
    last(), 
    // add the index of the observable to the output
    map(value => ({ index, value })) 
  ))).pipe(
    // use scan to keep track of previous values and insert current values
    scan((acc, curr) => (acc[curr.index] = curr.value, acc), Array(sources.length).fill(undefined))
  );
}

https://stackblitz.com/edit/rxjs-gwch8m

【讨论】：

【解决方案2】：

您可以使用combineLatest 发出来自每个可观察源的最新值。在每个源 observable 至少发出一次之前它不会发出，因此您可以使用 startWith 提供起始值：

combineLatest(
        this.myService.api1().pipe(startWith(null)),
        this.myService.api2().pipe(startWith(null)),
        this.myService.api3().pipe(startWith(null)),
        this.myService.api4().pipe(startWith(null)),
        this.myService.api5().pipe(startWith(null))
    )
        .subscribe(
            ([r1,r2,r3,r4,r5]) => { ... do something })

初始输出为[null, null, null, null, null]。当每个 observable 发出时，它会替换数组中对应的null 值。

如果要忽略初始发射，可以使用skip(1)。

const sourceOne = of('Hello').pipe(delay(1000));
const sourceTwo = of('World!').pipe(delay(2000));
const sourceThree = of('Goodbye').pipe(delay(3000));
const sourceFour = of('World!').pipe(delay(4000));

//wait until all observables have emitted a value then emit all as an array
const example = combineLatest(
  sourceOne.pipe(startWith(null)),
  sourceTwo.pipe(startWith(null)),
  sourceThree.pipe(startWith(null)),
  sourceFour.pipe(startWith(null))
)
.pipe(skip(1));

//output:
//["Hello", null, null, null]
//["Hello", "World!", null null]
//["Hello", "World!", "Goodbye", null]
//["Hello", "World!", "Goodbye", "World!"]
//Complete
const subscribe = example.subscribe(val => console.log(val), null, () => console.log('Complete'));

这里有一个StackBlitz 来试试看。

【讨论】：