如何断言RxJS Observable各次发射之间的最小时间间隔

核心要求是不丢值、保序，所以不能用过滤、跳过类操作符改动原流行为，只在管道中插入无侵入的校验逻辑，校验通过后原样透传所有原始值。

实现逻辑 #

• 用RxJS内置的timestamp操作符给每个发射值附加高精度发射时间戳，该操作符不会改变原流的发射节奏、顺序，也不会丢弃任何值，仅给每个值挂载时间元信息。
• 用scan操作符缓存前一个带时间戳的发射项，逐次计算相邻两个值的发射时间差：
  • 第一个值没有前序项，不需要做间隔校验，直接缓存后透传
  • 从第二个值开始，计算当前值和前一个值的时间差，若小于200ms直接抛出断言错误，校验通过后缓存当前值、透传原始值
• 整个校验过程不会修改原始值、不会打乱发射顺序、不会丢弃任何发射项，下游接收到的流和原Observable行为完全一致。

可直接复用的代码 #

import { Observable, timestamp, scan, map } from 'rxjs';

/**
 * 断言Observable相邻发射值的最小时间间隔，不丢值、不改变原流发射顺序和时机
 * @param minMs 要求的最小相邻间隔，默认200ms
 */
function assertMinInterval<T>(minMs: number = 200) {
  return (source$: Observable<T>) => source$.pipe(
    // 给每个发射值绑定当前时间戳
    timestamp(),
    // 缓存前项、逐次校验间隔
    scan((prevTimestamped, currTimestamped) => {
      // 第一个值没有前项，跳过校验直接缓存
      if (prevTimestamped !== null) {
        const gap = currTimestamped.timestamp - prevTimestamped.timestamp;
        if (gap < minMs) {
          throw new Error(`断言失败：相邻值发射间隔仅${gap}ms，未达到最小${minMs}ms的要求`);
        }
      }
      return currTimestamped;
    }, null),
    // 剥离时间戳元信息，透传原始值
    map(item => item!.value)
  );
}

使用方式 #

直接将操作符拼接在目标Observable的管道中即可，不需要修改原有订阅逻辑：

// 你自己的0~1000ms随机间隔发射的源Observable
const source$ = yourRandomIntervalObservable;

// 加了断言的流，发射行为、值内容、顺序和原流完全一致
const checked$ = source$.pipe(assertMinInterval(200));

// 正常订阅即可
checked$.subscribe({
  next: (val) => {
    // 这里收到的所有值和原流完全一致，没有丢失、没有乱序
    console.log('收到发射值:', val);
  },
  error: (err) => {
    // 校验失败会走到这里
    console.error('间隔不符合要求:', err.message);
  },
  complete: () => {
    console.log('流正常结束，所有相邻发射间隔均满足≥200ms的要求');
  }
});

避坑提示：不要直接单独使用pairwise实现校验，原生pairwise会默认忽略流的第一个发射值，会导致值丢失，不符合要求。

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