当ObservableA提前1秒触发时过滤ObservableB，如何避免回调重复执行？

需求回顾 #

需要处理两个Observable的触发场景，确保回调不会执行两次：

• 场景1：仅observableA触发 → 回调正常执行1次
• 场景2：仅observableB触发 → 回调正常执行1次
• 场景3：observableA先触发，1秒后observableB触发 → 仅执行observableA的回调，过滤observableB的触发

实现方案（RxJS） #

通过记录observableA的触发时间戳，对observableB的触发进行时间差校验，过滤掉1秒内的重复触发：

import { merge, Observable } from 'rxjs';
import { filter, map, shareReplay, withLatestFrom } from 'rxjs/operators';

// 假设已定义 observableA 和 observableB
const lastATimestamp$ = observableA.pipe(
  map(() => Date.now()),
  shareReplay(1) // 缓存最近一次A触发的时间戳
);

// 直接保留observableA的触发
const processedA$ = observableA;

// 过滤1秒内跟随A触发的B
const processedB$ = observableB.pipe(
  withLatestFrom(lastATimestamp$, (bValue, lastA) => ({ bValue, lastA })),
  filter(({ lastA }) => {
    // 若A从未触发过，直接允许B通过；否则判断时间差是否超过1秒
    return lastA === undefined || (Date.now() - lastA) > 1000;
  }),
  map(({ bValue }) => bValue)
);

// 合并处理后的流，订阅此流即可避免重复回调
const finalObservable$ = merge(processedA$, processedB$);

// 订阅示例
finalObservable$.subscribe(result => {
  // 你的回调逻辑，不会出现重复执行的情况
});

逻辑说明 #

1. lastATimestamp$：实时记录observableA每次触发的时间戳，通过shareReplay(1)确保后续订阅能获取到最近一次的时间数据。
2. processedA$：直接透传observableA的所有触发，不做过滤。
3. processedB$：结合最近一次A的触发时间，判断当前B触发是否满足“距离A触发超过1秒”的条件，不满足则过滤该次B的触发。
4. 合并流：将处理后的A和B流合并，订阅后只会在符合预期的场景下触发回调，不会出现场景3中的重复执行问题。

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