RxJS如何用Subject实现单并发异步队列，批量处理未处理任务无额外延迟

你可以通过buffer操作符配合自定义的释放信号实现零额外延迟的批量拉取未处理任务逻辑，完全满足你不能用bufferTime/bufferCount、并发数为1、批量处理当前所有积压任务的要求。

完整代码实现 #

// 缓冲区释放信号：用于通知上游吐出当前累计的所有未处理任务
const releaseBuffer$ = new Subject<void>();
// 任务处理状态标记，避免空闲状态下新入队任务无法触发处理逻辑
let isProcessing = false;

this.tasks
  .pipe(
    // 新任务入队时检测处理状态，空闲则直接触发处理
    tap(() => {
      if (!isProcessing) {
        releaseBuffer$.next();
      }
    }),
    // 每当释放信号触发时，输出从上次触发到现在累计的所有未处理任务数组
    buffer(releaseBuffer$),
    // 过滤空数组，避免无积压任务时执行空处理
    filter(tasks => tasks.length > 0),
    // 并发数固定为1，严格保证同一时间仅运行一批任务
    mergeMap(async tasks => {
      isProcessing = true;
      try {
        await this.processTasks(tasks);
      } finally {
        isProcessing = false;
        // 处理完成后立刻触发下一次积压任务拉取，无任何额外延迟
        releaseBuffer$.next();
      }
    }, 1)
  )
  .subscribe();

逻辑说明 #

• 零延迟机制：仅在新任务入队且系统空闲、或上一批任务处理完成的瞬间才会触发缓冲区释放，不会给任何任务引入固定等待延迟
• 批量处理逻辑：任务运行期间所有新入队的任务都会被暂存在缓冲区，处理完成后一次性取出所有积压任务批量执行，完全匹配你期望的运行效果
• 并发控制：mergeMap并发参数固定为1，严格保证同一时间只有一批任务在执行，符合异步任务队列的并发要求

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