Angular 10，rxjs - 订阅循环？

Question

我正在向 API 发出多个并发 DELETE 请求，导致服务器端出现死锁。

如果我有固定数量的删除，我可以这样做：

myAPI.delete(id).subscribe(res => {
   myAPI.delete(id2).subscribe(res => {
      myAPI.delete(id3).subscribe(res => {
      })
   })
})

等待上一个删除成功完成，然后再创建一个新的（连续的）。

问题是我可以有从1 到N 的请求，所以我正在寻找类似于for 订阅指令的东西。

有什么优雅的方法可以解决这个问题吗？

Answer 1

如果您有需要删除的 id 数组，您可以将from 运算符与concatMap 和toArray 结合使用。最后一个是在全部完成后才调用订阅监听器：

const ids = [ 1, 2, 3 ];

from(ids).pipe(
  concatMap((id) => deleteThis(id)),
  toArray()
).subscribe(() => {
   console.log('all deleted');
})

working stack 可观察到模拟请求

Answer 2

有多种方法。

顺序请求

您可以使用 RxJS range 运算符和映射运算符 concatMap 来顺序调用每个请求。

import { range } from 'rxjs';
import { concatMap } from 'rxjs';

range(1, N).pipe(                      // <-- will emit 1-N numbers
  concatMap(id => myAPI.delete(id))    // <-- set the correct `id` for each request here
).subscribe(
  res => console.log(res),
  err => console.log(err),
);

由于我们使用concatMap 运算符，每个id 的请求将在前一个请求完成之前等待。

您还可以针对不同的行为使用其他映射运算符，例如 switchMap、flatMap 和 exhaustMap。差异 b/n 他们here。

---

其他方法 - 并行请求

完成并行请求

您可以使用 RxJS 的 toArray 和 forkJoin() 运算符进行多个并发调用。它同时订阅多个可观察对象。

import { range, forkJoin } from 'rxjs';
import { concatMap, toArray } from 'rxjs/operators';

range(1, N).pipe(                      // <-- will emit 1-N numbers
  concatMap(id => myAPI.delete(id)),   // <-- set the correct `id` for each request here
  toArray(),                           // <-- buffer all notifications and emit once as an array
  concatMap(reqs => forkJoin(reqs))
).subscribe(
  res => console.log(res),
  err => console.log(err),
);

缓冲的并行请求

大多数浏览器对单个域的最大并行请求数有硬性限制（例如 Chrome - 6）。如果你遇到这个问题，你可以使用 RxJS 的 bufferCount 操作符而不是 toArray 操作符来控制最大并行请求数。

import { from, forkJoin } from 'rxjs';
import { concatMap, bufferCount } from 'rxjs';

range(1, N).pipe(                      // <-- will emit 1-N numbers
  concatMap(id => myAPI.delete(id)),   // <-- set the correct `id` for each request here
  bufferCount(6),                      // <-- adjust number of parallel requests here
  concatMap(reqs => forkJoin(reqs))
).subscribe(
  res => console.log(res),
  err => console.log(err),
);

Answer 3

另一种解决方案是使用concat 运算符：

import { concat } from 'rxjs';
import { toArray } from 'rxjs/operators';

concat(
   ...[1, 2, 3].map(id => this.delete(id)),
)
.pipe(toArray())
.subscribe(() => console.log('all deleted'))

或不带toArray() 运算符：

import { concat } from 'rxjs';

concat(...[1, 2, 3].map(id => this.delete(id)))
  .subscribe(
    res => {
      console.log('one deleted')
    },
    err => {},
    () => {
      console.log('all deleted')
    })

Answer 4

const N = 10;
range(1, N)
  .pipe(concatMap(deleteById), ignoreElements())
  .subscribe({
      complete: () => console.log('Complete')
  });

function deleteById(id: number): Observable<String> {
  return of(`deleted-${id}`).pipe(delay(1000));
}

如果你想同时发出多个请求：

const N = 10;
const CHUNKED = 3; // (3 requests same time) concat (3 requests same time) ...

range(1, N)
  .pipe(
    bufferCount(CHUNKED),
    concatMap((ids) => forkJoin(ids.map(deleteById))),
    ignoreElements()
  )
  .subscribe({
    complete: () => console.log(`Complete`)
  });