基于RxJS实现可恢复遗漏事件的轮询方案求助

刚好我之前处理过类似的RxJS轮询补全场景，给你一个完整的实现方案，让事件流的消费者完全不用操心遗漏事件的问题：

核心思路 #

每次轮询拿到最新事件后，对比上一次处理的事件ID，找出中间遗漏的ID，然后按顺序请求这些遗漏的事件，最后把所有事件（遗漏的+最新的）按ID递增的顺序推送给消费者。

完整代码实现 #

import { timer, from, concat } from 'rxjs';
import { concatMap, scan, filter } from 'rxjs/operators';

// 假设你的getEvent是返回Promise的异步函数，这里是模拟实现
function getEvent(id: string | number): Promise<{ id: number; /* 其他事件字段 */ }> {
  // 实际场景替换成你的API调用逻辑
  return new Promise(resolve => {
    if (id === "latest") {
      // 模拟随机跳过某些ID的情况
      resolve({ id: Math.floor(Math.random() * 15) });
    } else {
      resolve({ id: Number(id) });
    }
  });
}

// 构建完整的事件轮询流
const continuousEventStream$ = timer(0, 500).pipe(
  // 每次轮询获取最新事件
  concatMap(() => from(getEvent("latest"))),
  // 用scan维护状态：记录最后处理的事件ID，计算需要补全的遗漏ID
  scan((state, latestEvent) => {
    const missingIds: number[] = [];
    // 仅当最新事件ID大于上次处理的ID时，才计算遗漏的ID
    if (latestEvent.id > state.lastProcessedId) {
      // 生成从lastProcessedId+1到latestEvent.id-1的所有遗漏ID
      for (let id = state.lastProcessedId + 1; id < latestEvent.id; id++) {
        missingIds.push(id);
      }
      // 更新状态：最后处理ID改为最新事件ID，同时带上需要请求的ID列表（遗漏的+最新的）
      return {
        lastProcessedId: latestEvent.id,
        idsToFetch: [...missingIds, latestEvent.id]
      };
    }
    // 如果没有新事件，返回空的请求列表
    return {
      ...state,
      idsToFetch: []
    };
  }, { lastProcessedId: -1, idsToFetch: [] }),
  // 过滤掉没有事件需要请求的情况，避免无效操作
  filter(state => state.idsToFetch.length > 0),
  // 按顺序请求每个需要获取的事件ID，确保事件按ID递增输出
  concatMap(state => {
    const eventRequests = state.idsToFetch.map(id => from(getEvent(id)));
    return concat(...eventRequests);
  })
);

// 订阅测试：消费者拿到的是连续的事件流
continuousEventStream$.subscribe(event => {
  console.log('收到事件:', event);
});

代码关键点解释 #

• timer(0, 500)：启动轮询，立即执行第一次请求，之后每500ms执行一次。
• scan操作符：核心状态管理工具，用来跟踪我们最后处理过的事件ID，同时计算出遗漏的事件ID列表。比如上次处理到ID1，最新事件是ID3，就会生成[2, 3]的ID列表。
• concatMap处理请求：把需要获取的ID列表转成一个个getEvent的Observable，用concat按顺序执行，保证事件严格按ID递增的顺序输出。
• filter过滤：如果轮询时最新事件ID和上次处理的ID一致，说明没有新事件，直接跳过，避免无效的API调用。

这样你的消费者订阅后，拿到的就是完全连续的事件流，哪怕轮询间隔大导致遗漏了事件，也会自动补全并按顺序推送，完全不需要消费者额外处理。

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