RXJS如何实现指定时间范围内流数据累计结果的实时输出

实现思路 #

核心逻辑为：将上游流的每个元素拆成「新增计数」和「到期减计数」两个事件，合并所有事件后累计计数，每次计数变更就输出最新统计结果，天然满足新值进入、旧值移出都触发更新的要求。

完整实现代码 #

import { Observable, mergeMap, map, scan, startWith, timer } from 'rxjs';

/**
 * 统计滑动时间窗口内的流元素出现次数
 * @param source$ 输入RXJS流
 * @param windowDuration 窗口时长，单位毫秒
 * @returns 输出每个时刻窗口内的元素统计结果
 */
function countInSlidingWindow<T>(source$: Observable<T>, windowDuration: number): Observable<Record<string, number>> {
  return source$.pipe(
    // 每个元素生成两个事件：立刻加1，到期后减1
    mergeMap(value => [
      { type: 'add', value: String(value) },
      timer(windowDuration).pipe(map(() => ({ type: 'remove', value: String(value) })))
    ]),
    // 累计计数
    scan((stats, event) => {
      const current = stats[event.value] ?? 0;
      const next = event.type === 'add' ? current + 1 : current - 1;
      // 计数为0仍保留字段，和需求示例格式一致
      return {
        ...stats,
        [event.value]: next < 0 ? 0 : next
      };
    }, {} as Record<string, number>),
    // 可选添加初始空值，不需要可以删除
    startWith({})
  );
}

使用示例 #

以统计过去1分钟的元素计数为例：

import { fromEvent } from 'rxjs';

// 示例：统计页面点击事件过去1分钟内不同标签元素的点击次数
const click$ = fromEvent(document, 'click').pipe(map(e => (e.target as HTMLElement).tagName));
const oneMinuteStats$ = countInSlidingWindow(click$, 60 * 1000);

// 订阅即可拿到每次变更的最新统计结果
oneMinuteStats$.subscribe(stats => {
  console.log('当前窗口统计：', stats);
});

效果验证 #

完全匹配你给出的弹珠图场景（窗口为60秒）：

• 前4个元素依次为a、b、c、a进入窗口时，统计结果为{a:2, b:1, c:1}
• 后续再进入2个a，此时最早的a还未过期，统计结果为{a:4, b:1, c:1}
• 再过一段时间，最早的a、b、c全部过期，三个值的计数分别减1，得到{a:3, b:0, c:0}，后续随着更多c进入和老a过期，结果完全符合示例预期。

可选优化 #

如果不需要保留计数为0的key，可以修改scan逻辑：

scan((stats, event) => {
  const current = stats[event.value] ?? 0;
  const next = event.type === 'add' ? current + 1 : current - 1;
  const newStats = {...stats};
  if (next <= 0) {
    delete newStats[event.value];
  } else {
    newStats[event.value] = next;
  }
  return newStats;
}, {} as Record<string, number>)

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