基于NGRX Effect实现可取消且带超时的服务轮询方案

嘿，我来帮你解决这个NGRX轮询的问题！你的现有代码确实没做终止和超时处理，咱们用RxJS的几个实用操作符就能轻松搞定——让轮询要么在API返回true时停止，要么到10秒超时就自动终止。

核心实现思路 #

我们可以借助RxJS的race、takeWhile和timer这几个操作符来实现需求：

• 用timer(10000)创建一个超时信号，10秒后触发轮询终止
• 用takeWhile判断轮询结果，只要API返回false就继续轮询，返回true就停止
• 用race让轮询流和超时流“竞争”，任意一个先触发就终止整个流程，同时发出对应的Action

改进后的完整代码 #

下面是适配RxJS 6+的写法（如果是旧版本可以调整导入方式）：

import { Observable, race } from 'rxjs';
import { timer } from 'rxjs/observable/timer';
import { map, switchMap, takeWhile, catchError } from 'rxjs/operators';

@Effect() pollEffect$: Observable<Action> = this.actions$
  .ofType(tasksActions.ActionTypes.START_POLLING)
  .switchMap(() => {
    // 1. 创建超时流：10秒后发出超时Action
    const timeout$ = timer(10000).pipe(
      map(() => ({ type: tasksActions.ActionTypes.POLLING_TIMEOUT }))
    );

    // 2. 创建轮询流：每200ms调用一次API，直到返回true
    const polling$ = timer(0, 200)
      .pipe(
        // 每次定时器触发就调用后端API
        switchMap(() => this.myBackendService.getAvailability()),
        // 将API返回值映射成更新Action
        map(response => ({ 
          type: tasksActions.ActionTypes.UPDATE_VALUE, 
          payload: response 
        })),
        // 捕获API调用错误，避免单次错误中断整个轮询（可根据需求调整）
        catchError(error => {
          console.error('轮询API出错:', error);
          return Observable.of({ 
            type: tasksActions.ActionTypes.POLLING_ERROR, 
            payload: error 
          });
        }),
        // 只要返回值不是true就继续轮询，第二个参数true表示保留最后一次满足条件的Action
        takeWhile(action => !action.payload, true)
      );

    // 3. 让轮询流和超时流竞争，谁先触发就用谁的结果，同时终止另一个流
    return race(polling$, timeout$);
  });

代码细节解释 #

1. 超时流timeout$：
   用timer(10000)生成一个10秒后触发的流，触发时发出POLLING_TIMEOUT Action，告知应用轮询超时。

2. 轮询流polling$：

   • timer(0, 200)：表示立即开始第一次轮询，之后每200ms触发一次
   • switchMap：每次定时器触发时，取消之前未完成的API请求（如果有的话），发起新的请求
   • takeWhile(action => !action.payload, true)：核心逻辑——只要Action的payload不是true，就继续轮询；第二个参数true确保返回true的那次Action会被正常发出，然后停止轮询

3. race操作符：
   把轮询流和超时流传入race后，两个流会同时运行，任意一个流先发出值，另一个流就会被自动取消，完美实现“二选一”的终止逻辑。

额外优化建议 #

• 如果需要支持手动停止轮询，可以再加一个监听停止Action的流，把它也加入到race里即可
• 错误处理可以根据业务调整：比如如果API连续失败N次就停止轮询，或者直接终止轮询并发出错误提示
• 若使用RxJS 5版本，只需把pipeable操作符换成链式调用（比如.takeWhile(...)而不是.pipe(takeWhile(...))）

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