如何避免RxJS Timer实现的HTTP轮询取消未完成请求？

你遇到的问题根源在于switchMap的特性：当上游的timer按时发射新值时，switchMap会立即取消之前尚未完成的API请求订阅，转而订阅新的API请求。这就是为什么当mock API耗时超过500ms的轮询间隔时，会出现随机取消的情况。

要实现「等待API调用完成后，再执行下一次轮询」的逻辑，我们需要用串行执行的操作符替代switchMap，下面提供几种靠谱的方案：

方案1：使用expand操作符（推荐） #

expand是处理轮询场景的绝佳工具，它会在上一个Observable完成后，自动触发下一次请求，完美契合你的需求：

let n = 0;
function mockApiCall() { 
  n++;
  return Rx.Observable.of([n])
    .delay(400 + Math.round(Math.random()) * 400);
}

// 先发起第一次请求，之后每次完成后等待500ms再发起下一次
mockApiCall()
  .expand(() => 
    Rx.Observable.timer(500) // 等待轮询间隔
      .concatMap(() => mockApiCall())
  )
  .take(5) // 只取前5次结果
  .subscribe(console.log);

原理 #

• 首先执行一次mockApiCall，得到第一个结果
• expand会把每个结果传入回调函数，我们在回调里先等待500ms，再发起新的API请求
• 这个过程会循环，直到take(5)终止订阅，完全不会出现中途取消请求的情况

方案2：使用concatMap+递归链式调用 #

如果觉得expand不够直观，也可以用concatMap结合递归函数实现：

let n = 0;
function mockApiCall() { 
  n++;
  return Rx.Observable.of([n])
    .delay(400 + Math.round(Math.random()) * 400);
}

function poll() {
  // 先执行API请求，完成后等待500ms，再递归调用poll
  return mockApiCall().concatMap(() => 
    Rx.Observable.timer(500).concatMap(poll)
  );
}

poll().take(5).subscribe(console.log);

原理 #

• concatMap会严格等待前一个Observable完成后，才会订阅下一个
• 每次API请求完成后，我们先等待500ms的轮询间隔，再发起下一次请求，确保串行执行

为什么原来的写法不行？ #

再回头看你原来的代码：

Rx.Observable
.timer(500, 500)
.switchMap( _=>mockApiCall()).take(5)

这里timer是每隔500ms固定发射值，不管mockApiCall是否完成。当mockApiCall耗时超过500ms时，下一个timer值发射，switchMap就会立刻取消之前还在pending的mockApiCall订阅，导致请求被终止。

而我们的方案都是把「等待间隔」的逻辑放在API请求完成之后，从根源上避免了这种冲突。

内容的提问来源于stack exchange，提问作者arunmj