如何结合Rxjs的bufferCount与bufferTime实现缓冲逻辑？

要实现缓冲N个事件，或到达指定时间间隔时立即发出累积内容的需求，你可以结合bufferWhen和race操作符，同时监听「数量达标」和「时间到期」两个触发条件，任一条件满足就输出当前缓冲并重置周期。

核心实现代码 #

import { Observable, timer, race } from 'rxjs';
import { bufferWhen, ignoreElements, take } from 'rxjs/operators';

/**
 * 创建一个双条件缓冲的Observable
 * @param source$ 源Observable
 * @param bufferSize 触发缓冲的元素数量阈值
 * @param bufferTimeMs 触发缓冲的时间间隔（毫秒）
 * @returns 缓冲后的Observable
 */
function bufferCountOrTime<T>(
  source$: Observable<T>,
  bufferSize: number,
  bufferTimeMs: number
): Observable<T[]> {
  return source$.pipe(
    bufferWhen(() => 
      race(
        // 条件1：到达指定时间间隔时触发缓冲
        timer(bufferTimeMs),
        // 条件2：收集到指定数量的元素时触发缓冲
        source$.pipe(
          take(bufferSize),
          ignoreElements() // 忽略元素值，仅监听完成信号
        )
      )
    )
  );
}

代码解释 #

• bufferWhen(() => ...)：为每个新的缓冲周期创建一个「通知Observable」，当这个通知Observable发出值时，就会输出当前累积的缓冲元素，并开始下一个周期。
• race(...)：同时监听两个子Observable，哪个先发出信号就触发缓冲结束：
  • timer(bufferTimeMs)：在指定时间间隔后发出信号，触发时间到期的缓冲。
  • source$.pipe(take(bufferSize), ignoreElements())：当源Observable产生bufferSize个元素后，通过ignoreElements()忽略元素值，仅用完成信号触发数量达标的缓冲。

示例用法 #

假设源Observable每500ms发出一个元素，我们设置缓冲数量为3、时间间隔为2000ms：

import { interval } from 'rxjs';

// 源Observable：每500ms发出一个递增数字
const source$ = interval(500);

// 创建双条件缓冲Observable
const buffered$ = bufferCountOrTime(source$, 3, 2000);

// 订阅输出
buffered$.subscribe(console.log);

输出结果：

• 第1500ms时，累积了3个元素[0,1,2]，触发数量条件输出。
• 第3500ms时（距离上一次输出过了2000ms），累积了2个元素[3,4]，触发时间条件输出。
• 第5000ms时，累积了3个元素[5,6,7]，触发数量条件输出，以此类推。

关键优势 #

相比单独使用bufferCount或bufferTime：

• 不会出现「元素一直不够数量，导致无限缓冲不输出」的问题。
• 不会出现「时间到了但元素数量少，仍需等待下一个时间周期」的问题。
• 所有事件都会被处理，不会丢失，任一条件触发后立即重置缓冲周期。

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