Angular每2个流式传输多个HTTP调用

发布于09月22日

我有一个有效负载元素数组:[1,2,3,4]. 我想通过多个HTTP调用流有效负载，同时2个HTTP调用.

启动前两个HTTP调用，第一个使用paylaodElement[0]，第二个使用paylaodElement[1].等待结果，然后进行2个新的HTTP调用...

下面的代码可以同时触发多个调用的所有有效负载:

const streams = [].concat.apply(
  [],
  payload.map((el) => this.http.get(API_ENDPOINT, { el }))
);

然后:

forkJoin(streams).subscribe(result => {
 // 
})

此FIRE N个HTTP调用，其中，N=payload.length. 我们怎么能同时只用2个电话来分块呢？

推荐答案

如果我没弄错问题，我会这样做

// first use the from rxjs function to turn an array into a stream of its values
from(payload).pipe(
   // then with bufferCount create a stream of tuples
   bufferCount(2),
   // now we have a stream of tuples - we turn the element of each tuple into 
   // a tuple of http calls using the map operator
   // consider that the inner map is the map function of the javascript array 
   // while the outer map is the rxjs operator
   map(tuple => tuple.map(t => this.http.get(API_ENDPOINT, { el })),
   // now we fire the calls in parallel using forkJoin
   // we use concatMap to say "wait until forkJoin notifies its result
   // before firing the next tuple of calls
   concatMap(tupleOfHttpCalls => forkJoin(tupleOfHttpCalls))
)
.subscribe(tupleOfResults => {
   // here you do whatever you want with the tuple of results
})

虽然这段代码应该能做您想做的事情，但我也会考虑一种解决方案，将并行调用的数量限制为2个，但不创建元组，而是拥有一个连续的流，即当两个调用正在进行并且第一个调用完成时，您立即开始一个新的调用.

这样的解决方案应该是这样的

// first use the from rxjs function to turn an array into a stream of its values
from(payload).pipe(
   // then we use mergeMap with the concurrent value (the second parameter) to limit
   // concurrency to 2
   mergeMap(value => this.http.get(API_ENDPOINT, { el }), 2),
)
.subscribe(result => {
   // here you do whatever you want with each single result
})

This blog可以为您提供有关用于异步调用的rxjs模式的更多信息.