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

Question

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

因此，我想将我的请求排队，然后触发一个，等待响应返回，然后再触发下一个请求。

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

这是一个使用 angular2 的 plunker，您可以在其中单击按钮。当您单击 1 秒按钮时，将创建一个可在 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

由于您只想让请求执行，您可以使用 mergeMap() 运算符并将 concurrency 设置为 1。请注意，这意味着要发送的项目将在您的 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>

Answer 2

您应该发布一些代码。 concatMap 应该适用于您的用例。如果您在订阅前观察到触发，则意味着您需要使用defer 运算符在订阅时触发但没有代码示例很难说更多。

过去问题的有用链接：

• How to start second observable *only* after first is *completely* done in rxjs
• Rx.js wait for callback to complete
• Rxjs, understanding defer