什么 rxjs 运算符类似于 concatmap 但在触发下一个请求之前等待每个请求？

Question

例如，假设我想进行一些 api 调用。我正在处理的 api 容易出现竞争条件，因此如果我同时进行 3 个 api 调用来更新服务器上的同一条数据，它可能会丢失一些数据。

因此，我想对我的请求进行排队，然后在触发下一个请求之前等待响应返回。

基本上我需要像 concatMap 这样的东西，但是 concatMap 的问题是它同时触发所有请求。我需要 concatMap 在触发下一个请求之前等待。我正在使用 rxjs 5。

这是一个使用 angular2 的 plunker，您可以在其中单击按钮。当您单击 1 sec 按钮时，将创建一个 observable 并在 1 秒后返回。有 2 秒和 3 秒按钮。

https://plnkr.co/edit/6F4JrVueQX8PjPinZqIk?p=preview

@Component({
  selector: 'my-app',
  template: `
    <div>
      <h2>Wait:</h2>

      <button (click)="start(1)">1 sec</button>
      <button (click)="start(2)">2 sec</button>
      <button (click)="start(3)">3 sec</button>
    </div>
  `,
})
export class App {


  constructor() {
  }

  start(wait) {

    const waitSecs = parseInt(wait) * 1000;

    of('clicked').delay(waitSecs).subscribe(
      val => console.log(wait)  
    )

    // Expected behavior:
    // I click 3 sec, 2 sec, and 1 sec right after another.  Then 
    // the console log should output 3, 2, 1.  Right now it's 
    // outputting 1, 2, 3.

  }
}

我对这个应用程序的理想行为是在我依次单击 3 秒、2 秒和 1 秒之后。然后控制台应该输出 3, 2, 1. 现在它正在输出 1, 2, 3。

Answer 1

由于您只想执行请求，您可以使用设置为 1的mergeMap()运算符concurrency。请注意，这意味着要发送的项目将在您的 mergeMap 运算符中排队，这可能会导致内存问题。

Rx.Observable.from([3,5,2,1])
  .mergeMap(
    i => {
      return Rx.Observable.of(i)
        .do(i => console.log(`starting request for ${i}`))
        .delay(1000)
        .do(null,null, () => console.log(`finished request for ${i}`));
    },
    null,
    1 /* concurrency limited to 1 */
  )
  .subscribe(val => console.log(`received value: ${val}`));

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/5.4.2/Rx.js"></script>