RxJS中使用expand运算符实现轮询出现非预期结果

• 固定布尔值输出：你大概率在流构造阶段直接调用了checkExistence()，比如写成of(checkExistence())——这类写法会在整个Observable被定义时就执行一次随机逻辑，后续整个链路复用这一次的结果，永远不会重新生成新的布尔值。
• 一次性批量输出：expand是同步递归触发的操作符，如果expand回调返回的是无延迟的同步Observable，它会在当前事件循环中一口气跑完所有递归层级，不会等待你预期的5秒间隔，自然会一次性吐出大量结果。
• 轮询停止/恢复错位：你设置的30秒停止、停5秒恢复的逻辑，因为没有和递归流的时序做绑定，会被同步批量触发，完全达不到间隔执行的效果。

完全匹配你的时序要求：订阅后立刻执行一次checkExistence输出结果，之后每5秒执行一次随机判断，满30秒自动停止轮询，停止5秒后重新开启轮询。

import { of, expand, timer, switchMap, takeUntil, repeatWhen, delayWhen } from 'rxjs';

// 随机布尔值生成方法
const checkExistence = () => Math.random() > 0.5;

// 轮询核心流
const poll$ = of(null).pipe(
  // 首次订阅时立刻执行一次随机判断
  switchMap(() => of(checkExistence())),
  expand(() => {
    // 关键：延迟5秒后重新执行随机逻辑，不复用上一次的返回值
    return timer(5000).pipe(
      switchMap(() => of(checkExistence()))
    );
  }),
  // 流启动满30秒时终止当前轮询
  takeUntil(timer(30000)),
  // 轮询终止后，延迟5秒重新启动整个轮询流程
  repeatWhen((completed$) => completed$.pipe(delayWhen(() => timer(5000))))
);

// 订阅触发执行
poll$.subscribe(res => console.log('轮询结果：', res, new Date().toLocaleTimeString()));

• 绝对不要把带副作用的方法（比如随机值生成、接口请求）作为参数直接传给of()/from()这类工厂方法，这类方法会在创建流时就执行一次副作用，后续复用固定结果；有副作用的逻辑必须放在switchMap这类能在执行时机触发的回调里，保证每次运行时都重新执行，从根源上避免固定值输出的问题。
• 不要在expand的返回逻辑里直接复用上游传入的上一次结果参数，否则很容易意外把上一次的结果当成新值输出，达不到递归获取新值的目的。
• 依赖timer操作符做expand的递归触发延迟，不要直接返回同步值，否则会触发无间隔的同步递归，出现一次性批量输出的问题。
• 全局的时间窗口控制（30秒停止、5秒恢复）不要手动拼接延迟值，优先用RxJS内置的takeUntil、repeatWhen这类操作符管理订阅生命周期，避免内存泄漏和时序错位。

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