如何确保RxJS管道中函数同一时间仅执行一次？

需求明确 #

确保title函数同一时间仅执行一次，且必须等当前任务末尾的race操作完成后，才能启动下一个任务的执行。

问题分析 #

原代码通过groupBy按title分组后使用mergeMap处理分组，会导致不同title的任务并行执行，从而打破“同一时间仅执行一次title”的限制。即便同一title的任务用concatMap串行，不同分组的并行依然会让title函数同时运行多次。

解决方案 #

方案1：全局所有任务串行（无论title是否相同） #

直接移除groupBy和mergeMap，用concatMap包裹完整任务流程，确保所有任务严格按顺序执行，前一个任务的race完成后才会启动下一个：

const title$ = titleTasks$.pipe(
  concatMap((task) =>
    defer(() => from(title(task))).pipe(
      map(() => ({
        success: true,
        error: undefined,
        ...task,
      })),
      catchError((error) => of({ success: false, error, ...task })),
      tap((taskResult) => {
        taskResult$.next(taskResult);
      }),
      switchMap((taskResult) =>
        race(
          timer(taskResult.success ? task.ttl * 1000 : 0),
          cancelTitleTimer$
        ).pipe(switchMap(() => of(taskResult)))
      )
    )
  )
);

方案2：仅同一title的任务串行（不同title可并行） #

如果需要保留按title分组的逻辑，确保同一title的任务必须等前一个的race完成后再执行，只需确保分组内的concatMap包裹从title执行到race完成的全流程即可：

const title$ = titleTasks$.pipe(
  groupBy(({ title }) => title),
  mergeMap((group$) =>
    group$.pipe(
      concatMap((task) =>
        defer(() => from(title(task))).pipe(
          map(() => ({ success: true, error: undefined, ...task })),
          catchError((err) => of({ success: false, error: err, ...task })),
          tap(res => taskResult$.next(res)),
          concatMap((taskResult) =>
            race(
              timer(taskResult.success ? task.ttl * 1000 : 0),
              cancelTitleTimer$
            ).pipe(map(() => taskResult))
          )
        )
      )
    )
  )
);

关键说明 #

• concatMap的核心特性是等待内部Observable完全完成后，才会订阅下一个Observable，这正好匹配“等待race完成后再执行下一个任务”的需求。
• 全局串行方案适用于所有任务必须严格排队的场景；分组串行方案适用于同一title任务排队、不同title任务可并行的场景。

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