RxJS Angular 6/7：如何使用mergeMap和delay处理HTTP请求？

Question

我想使用这段代码发送请求（我也尝试过forkJoin），并且在调用之间设置延迟：

duplicateElement(id: string): Observable<any> {
    return this.http.get({ routeName: 'route_name', params: { id } });
}

duplicateElements(ids: string[]): Observable<any> {
    return from(ids)
    .pipe(
        mergeMap(id => this.duplicateElement(id).pipe(delay(1000))
    ));
}

但是 .pipe(delay(1000) 没有按照我期望的方式工作：每个 HTTP 请求之间间隔 1000 毫秒。

- JPFs

---

你希望它如何工作？ - Arne

我想要每个HTTP请求之间延迟1000毫秒发送。 - JPFs

Answer 1

现在有两种选择！

基本设置

import * as rx from "rxjs";
import * as rxop from "rxjs/operators";

const startTime = new Date();

function getTimestamp() {
  return (new Date().getTime() - startTime.getTime()) / 1000;
}


const desiredDelay = 750;
const serviceDelay = 500;

// simulating service, you can ignore what's inside
var myService = (b: any) => {
  return rx.of(Math.random()).pipe(
    // To simulate long running service
    rxop.delay(serviceDelay),
    // Log the timestap after execution, should be ~ desiredDelay + serviceDelay, so by default 1250ms each emitted value
    rxop.tap(() => console.log(`${b} after service result, ${getTimestamp()}`)),
    // simulating the result
    rxop.map(a => "result" + b)
  );
};

延迟一段时间后逐个发出值，尽快执行服务并收集结果

of([1, 2, 3, 4, 5])
  .pipe(
    // See the initial values
    tap(console.log),
    // Split array into single values during emit
    // Collect observables and subscribe to next when previous completes
    concatAll(),
    // Emit each value as a sequence of observables with a desired delay
    concatMap(a => of(a).pipe(delay(desiredDelay))),
    // Call service on each value as soon as possible, do not care about the order
    mergeMap(a => myService(a)),
    // Reduce single values back into array
    // Reduces the values from source observable to a single value that's emitted when the source completes
    reduce<any>((acc, val) => [...acc, val], []),
    // See the result, not necessary
    tap(console.log)
  )
  .subscribe();

在一个值上调用服务，等待一段时间后，再使用另一个值调用服务。

of([1, 2, 3, 4, 5])
  .pipe(
    // See the initial values
    tap(console.log),
    // Split array into single values during emit
    // Collect observables and subscribe to next when previous completes
    concatAll(),
    // Call the service
    // Map values to inner observable, subscribe and emit in order
    concatMap(a => myService(a).pipe(delay(desiredDelay))),
    // Reduce single values back into array
    // Reduces the values from source observable to a single value that's emitted when the source completes
    reduce<any>((acc, val) => [...acc, val], []),
    // See the result, not necessary
    tap(console.log)
  )
  .subscribe();

- Jacek Lipiec

---

谢谢Jacek，但这种方式会在1000毫秒后发送所有请求，但我想每1000毫秒按ID发送一个请求。 - JPFs

试试这个。如果不行，等我真正坐在电脑前时，我就能给你更好的结果 :) return from(unitIds).pipe( switchMap(id => this.duplicateElement(id).pipe( delay(1000))), concatAll() ); - Jacek Lipiec

谢谢Jacek！解释和示例非常好。但我们需要使用mergeMap，因为我们希望请求是“伪”(带有延迟)并行的，如果使用concatMap，则它们是顺序的。 - JPFs

更新。请注意，同时触发时要保留呼叫顺序实际上要困难得多。如果您需要特定的调用顺序，我建议采用稍微不同的想法。 - Jacek Lipiec

太棒了的例子！在我们的情况下，顺序并不重要，所以它非常有效...非常感谢Jacek！ - JPFs

Answer 2

这个怎么样：

duplicateElements(ids: string[]): Observable<any> {
    return interval(1000).pipe( // start emitting every 1000ms
        take(ids.length),  // limit emissions to length of array
        map(i => ids[i]),  // map (change) emission to the array item @ index i 
        mergeMap(id => this.duplicateElement(id)) // add the http request
    )
}

- dmcgrandle

---

嗯...它哪里出了问题？我想到了一个需要添加的东西...我会用 take(ids.length) 更新答案，这样间隔就会在正确的尝试次数后停止发出。另外，你希望每个后续的 http 调用替换前一个 http 调用，还是希望它添加另一个 http 调用？mergeMap 会不断地构建它们，但如果你想要替换它们，那么 switchMap 可能是更好的选择。 - dmcgrandle

谢谢，太棒了！是的，你说得对...在正确的尝试次数之后发送了未定义。 - JPFs