RxJS如何实现仅在源Observable收到数据后开启指定时长缓存，不输出空值

以下给出两种适配不同场景的实现方式：

方案1：最小改动适配现有代码 #

如果可以接受bufferTime内部依然按固定间隔运行定时器，仅需过滤掉空数组输出，直接追加filter操作符即可，代码改动量最小：

const source$ = eventSubject.asObservable();
source$.pipe(
  takeUntil(destroyed$),
  bufferTime(2000),
  filter(dataArr => dataArr.length > 0) // 过滤掉空数组输出
).subscribe(dataArr => {
  console.log(dataArr);
});

该方案优点是逻辑简单易维护，缺点是内部定时器仍会每2秒触发一次，对性能要求极高的场景可选择方案2。

方案2：完全匹配需求，有数据才启动计时 #

该方案完全符合「只有收到数据才开始2秒计时缓存，无数据时不运行任何定时器、不输出任何内容」的要求，使用buffer配合auditTime实现：

import { auditTime, buffer, takeUntil } from 'rxjs/operators';

const source$ = eventSubject.asObservable();
// 缓存关闭通知：源第一次收到数据后启动2秒计时，到点触发缓存输出
const bufferCloseNotifier$ = source$.pipe(auditTime(2000));

source$.pipe(
  takeUntil(destroyed$),
  buffer(bufferCloseNotifier$)
).subscribe(dataArr => {
  console.log(dataArr);
});

逻辑说明：

• auditTime(2000) 仅在源Observable首次发射数据时启动2秒定时器，期间的新数据不会重置计时，2秒后发射一个通知信号
• buffer 持续收集源发射的所有数据，直到收到bufferCloseNotifier$的通知信号时，将收集到的全部数据作为数组输出，之后自动进入下一轮收集周期
• 源没有数据发射时，不会启动任何定时器，也不会输出任何内容

如果你的场景需要「收到最后一条数据后再等2秒无新数据才输出缓存」（即防抖式缓存，用户持续触发事件时一直攒数据，停下2秒才输出），把auditTime替换为debounceTime(2000)即可。

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