RxJS如何实现类似bufferCount但剩余元素不足缓冲数也发射的效果

核心需求匹配 #

你需要的是串行+动态批次大小的流处理逻辑：

• 批次最大大小固定（示例为5），流速快时每批最多处理该数量避免过载
• 流速慢时无需等缓冲区填满，当前积压多少就处理多少
• 所有处理逻辑串行执行，上一批处理完成后才会取下一批数据

实现代码 #

import { Subject, of } from 'rxjs';
import { concatMap, take, toArray, delay, tap } from 'rxjs/operators';

// 配置项：单批次最大处理数量
const MAX_BATCH_SIZE = 5;
// 你的业务事件源，逐个next元素即可
const eventSource$ = new Subject<string>();
// 处理完成触发器，用于触发下一次拉取批次
const processTrigger$ = new Subject<void>();

// 模拟你的耗时批次处理逻辑，替换为实际业务即可
const handleBatchProcess = (batch: string[]) => {
  console.log('开始处理批次，元素列表：', batch);
  // 模拟2秒耗时的服务端请求/计算逻辑
  return of(batch).pipe(delay(2000));
};

// 主逻辑流
processTrigger$.pipe(
  // 串行执行批次，上一批未完成不会进入下一批
  concatMap(() => 
    eventSource$.pipe(
      // 最多取MAX_BATCH_SIZE个元素，没凑够也会在队列空时直接返回当前所有
      take(MAX_BATCH_SIZE),
      // 把取出的元素转为批次数组
      toArray(),
      // 执行批次处理
      concatMap(batch => {
        // 无元素时直接跳过处理
        if (batch.length === 0) return of(null);
        return handleBatchProcess(batch);
      }),
      // 处理完成后触发下一次拉取
      tap(() => processTrigger$.next())
    )
  )
).subscribe();

// 初始化触发第一次拉取
processTrigger$.next();

逻辑说明 #

• 流启动后会自动从事件源拉取最多5个元素处理
• 若事件源当前积压少于5个，会直接取所有积压元素处理，不会等待缓冲区填满
• 上一批次处理完成后会自动触发下一次拉取，始终保持串行执行避免服务器过载
• 流速快时每批固定处理5个，流速慢时自适应处理1~4个，完全匹配你给出的时间线示例

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