如何让RxJS throttleTime同时启用首尾触发且避免请求连续执行？

问题背景 #

原生RxJS的throttleTime操作符在启用{ leading: true, trailing: true }时，触发尾项通知后会立即重置时间窗口。这会导致后续新通知若在尾项触发后不久到达，会被当作新窗口的首项立即触发，打破了“任意两次触发间隔至少X时长”的要求。

以你的示例为例：

• 源通知序列（每字符间隔1000ms）：-AB---C----
• 原生throttleTime(3000, asyncScheduler, { leading: true, trailing: true })输出：-A---BC----
• 期望输出：-A---B---C-

核心需求 #

• 首次通知立即触发
• 任意两次触发（无论首项还是尾项）之间必须至少间隔X时长
• 能捕获时间窗口内的最后一个通知（等效trailing: true的效果）

解决方案：自定义严格节流操作符 #

以下是直接实现的自定义操作符，完全满足你的三点需求：

import { Observable, OperatorFunction, asyncScheduler } from 'rxjs';

function strictThrottleTime<T>(
  duration: number,
  scheduler = asyncScheduler
): OperatorFunction<T, T> {
  return (source) =>
    new Observable((subscriber) => {
      let lastEmitTime: number | null = null;
      let pendingValue: T | null = null;
      let timerId: any = null;

      // 触发待处理的值并更新状态
      const emitPending = () => {
        if (pendingValue !== null) {
          subscriber.next(pendingValue);
          lastEmitTime = scheduler.now();
          pendingValue = null;
        }
        timerId = null;
      };

      return source.subscribe({
        next(value) {
          const now = scheduler.now();
          
          // 首次通知直接触发
          if (lastEmitTime === null) {
            subscriber.next(value);
            lastEmitTime = now;
            return;
          }

          const timeSinceLastEmit = now - lastEmitTime;
          
          if (timerId === null) {
            // 距离上次触发已满足间隔，直接触发
            if (timeSinceLastEmit >= duration) {
              subscriber.next(value);
              lastEmitTime = now;
            } else {
              // 未满足间隔，缓存最新值并设置定时器
              pendingValue = value;
              timerId = scheduler.schedule(emitPending, duration - timeSinceLastEmit);
            }
          } else {
            // 已有定时器运行，仅更新缓存的最新值
            pendingValue = value;
          }
        },
        error(err) {
          // 清理定时器并传递错误
          if (timerId !== null) scheduler.clear(timerId);
          subscriber.error(err);
        },
        complete() {
          // 完成时触发剩余的待处理值
          if (pendingValue !== null) subscriber.next(pendingValue);
          subscriber.complete();
        },
      });
    });
}

逻辑说明 #

1. 状态维护：通过lastEmitTime记录上次触发时间，pendingValue缓存等待触发的最新值，timerId跟踪定时器实例。
2. 首次触发：收到第一个值时立即输出，并更新lastEmitTime。
3. 后续值处理：
   • 若距离上次触发已过X时长且无待处理定时器，直接输出当前值。
   • 若未满足间隔，缓存当前值并设置定时器，等到上次触发后X时长的节点输出。
   • 若已有定时器在运行，仅更新缓存值（保证始终输出窗口内的最后一个值）。
4. 完成处理：源Observable结束时，立即输出剩余的待处理值。

示例验证 #

用你的测试序列验证：

• 1s收到A：首次触发，输出A，lastEmitTime=1000
• 2s收到B：距离上次触发仅1000ms，缓存B并设置定时器在4000ms触发
• 4s定时器触发：输出B，lastEmitTime=4000
• 6s收到C：距离上次触发2000ms，缓存C并设置定时器在7000ms触发
• 7s定时器触发：输出C
• 最终输出序列：-A---B---C-，完全符合期望。

替代方案：现有操作符组合 #

如果你倾向于用RxJS内置操作符组合实现，也可以用scan跟踪状态+delayWhen控制时机：

import { scan, delayWhen, filter, map } from 'rxjs/operators';

function strictThrottleTime<T>(duration: number) {
  return source => source.pipe(
    scan((state, value) => {
      const now = Date.now();
      const timeSinceLastEmit = now - (state.lastEmitTime ?? 0);
      
      if (state.lastEmitTime === null) {
        return {
          lastEmitTime: now,
          emitNow: true,
          value,
          pending: null,
          delay: 0
        };
      }

      if (state.pending !== null) {
        return { ...state, pending: value };
      }

      if (timeSinceLastEmit >= duration) {
        return {
          lastEmitTime: now,
          emitNow: true,
          value,
          pending: null,
          delay: 0
        };
      } else {
        return {
          lastEmitTime: state.lastEmitTime,
          emitNow: false,
          value: null,
          pending: value,
          delay: duration - timeSinceLastEmit
        };
      }
    }, { lastEmitTime: null as number | null, emitNow: false, value: null as T | null, pending: null as T | null, delay: 0 }),
    delayWhen(state => state.emitNow ? Promise.resolve() : new Promise(resolve => setTimeout(resolve, state.delay))),
    map(state => state.emitNow ? state.value : state.pending),
    filter(value => value !== null)
  );
}

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