RxJS：如何抑制特定时间段内重复触发的相同事件？

我需要实现忽略指定时间段内重复触发的事件，但如果相同事件在该时间段外触发，则必须处理。尝试了两种RxJS方案都不满足需求：

• 方案1（throttleTime）：会过滤时间段内的所有事件，哪怕是不同的事件，不符合要求

  // 无法满足需求：时间段内的不同事件也会被过滤，而我需要处理这些事件
  event$.pipe(
    throttleTime(300)
  ).subscribe((event) => console.log(event));
  

• 方案2（distinctUntilChanged）：会永久忽略重复事件，哪怕间隔很久触发，不符合要求

  // 无法满足需求：即使相同事件间隔两分钟触发，也会被永久忽略，而我需要处理这类事件
  event$.pipe(
    distinctUntilChanged((previous, current) => JSON.stringify(previous) === JSON.stringify(current))
  ).subscribe((event) => console.log(event));
  

请问有没有办法结合两者或用其他方式实现需求？

可以通过groupBy + throttleTime的组合实现需求：先按事件的唯一标识将事件流分组，再对每个分组单独应用节流逻辑。这样不同事件不会互相干扰，同一事件在指定时间段内只会触发一次，超时后再次触发会被正常处理。

以下是封装好的可复用RxJS操作符：

/**
 * RxJS 操作符：按分组节流事件，确保同一分组的事件在指定时间段内最多触发一次
 * 默认使用 JSON.stringify 对事件进行分组，即完全相同的事件会被归为一组
 *
 * @param duration - 节流的时间窗口（毫秒），默认值 100
 * @param keySelector - 分组键选择函数，返回用于分组的字符串，默认使用 JSON.stringify(event)
 * @returns 可用于 RxJS pipe 的操作符函数
 *
 * 示例：
 * import { throttleGroups } from "rxjs-utils"
 *
 * // 对拥有相同 url 属性的事件进行节流（500ms 内同一 url 只触发一次）
 * rapidEvents$.pipe(
 *      throttleGroups(500, (event) => event.url)
 * ).subscribe((event) => console.log('500ms 内不会重复输出相同 url', event.url));
 *
 * // 对完全相同的事件进行节流（默认 100ms 内同一事件只触发一次）
 * rapidEvents$.pipe(
 *      throttleGroups()
 * ).subscribe((event) => console.log('100ms 内不会重复输出相同事件', event));
 */
export const throttleGroups = (duration: number = 100, keySelector = (event): string => JSON.stringify(event)) => {
    return function<T> (source: Observable<T>): Observable<T> {
        return source.pipe(
            groupBy(keySelector),
            mergeMap((eventGroup$) => eventGroup$.pipe(throttleTime(duration)))
        );
    };
};

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