RxJS疑难问题：实现防抖同时为异步函数添加互斥机制

问题背景 #

我们有一个表单值的RxJS流formValue$，每次表单按键操作都会触发新值发射，这些值需要通过异步函数save()保存到后端。save()会计算与上次保存值的差异，有变化时发起一次或多次HTTP请求。

需要实现的核心逻辑：

• 避免每次formValue$发射都触发save()，防止过多HTTP请求和读写竞态条件
• 先处理第一个值并触发save()，等待save()完成（可额外等待数秒），再获取formValue$的最新值并再次触发save()

现有实现的问题 #

参考互斥实现写出的代码能等待异步handler完成，并跳过formValue$中堆积的发射，但存在缺陷：当handlingFinished$发射时stream$没有值，返回的流会停滞/终止，后续formValue$的新发射无法再触发handler。

现有代码：

// formValue$会作为stream$传入，save函数作为handler
function getEventSkippingMutExQueue<Event, Result>(
  stream$: Observable<Event>,
  handler: (e: Event) => Promise<Result>
): Observable<Result> {
  let handlingFinished$ = new Subject<Result>() // 也可以用Subject<void>

  let eventSkippingStream = merge(
    stream$.pipe(take(1)), // 让stream$的第一次发射通过
    stream$.pipe(audit((_) => handlingFinished$))
  )
  // 之后仅在handlingFinished$发射时处理最新值

  return eventSkippingStream.pipe(
    concatMap((input) => from(handler(input))),
    tap((value) => handlingFinished$.next(value))
  )
}

问题原因 #

原代码的核心问题在于merge的两个流逻辑冲突：

1. stream$.pipe(take(1))在第一次值发射后就会完成终止
2. stream$.pipe(audit((_) => handlingFinished$))只有当stream$有值发射时，才会进入等待handlingFinished$的状态；如果handlingFinished$触发时stream$没有待处理的值，这个流不会主动监听后续的stream$新值，导致整个流停滞。

解决方案 #

方案一：使用expand实现递归循环处理 #

expand操作符可以在每次处理完成后递归调用逻辑，天然形成持续监听的循环，不会出现流停滞的问题，还能轻松添加处理后的延迟。

import { Observable, from, EMPTY, timer } from 'rxjs';
import { expand, take(1), concatMap, tap } from 'rxjs/operators';

function getEventSkippingMutExQueue<Event, Result>(
  stream$: Observable<Event>,
  handler: (e: Event) => Promise<Result>,
  delayAfterFinish = 0 // 可选：处理完成后额外等待的毫秒数
): Observable<Result> {
  // 定义单次处理逻辑：获取最新值 -> 执行handler -> 可选延迟
  const processLatestValue = () => 
    stream$.pipe(
      take(1), // 获取当前流的最新值
      concatMap(input => from(handler(input))),
      // 如果需要额外等待，这里添加延迟
      tap(() => delayAfterFinish ? timer(delayAfterFinish).pipe(take(1)) : EMPTY),
      // 处理完成后递归调用，继续监听下一个最新值
      expand(() => processLatestValue())
    );

  return processLatestValue();
}

方案二：修复原audit逻辑，用repeat保持流活跃 #

通过repeat()让audit流在每次处理完成后重新激活，确保后续stream$的新值能被捕获，同时添加终止逻辑避免内存泄漏。

import { Observable, from, Subject } from 'rxjs';
import { audit, concatMap, tap, repeat, takeUntil, last } from 'rxjs/operators';

function getEventSkippingMutExQueue<Event, Result>(
  stream$: Observable<Event>,
  handler: (e: Event) => Promise<Result>
): Observable<Result> {
  const handlingFinished$ = new Subject<void>();

  const eventStream$ = stream$.pipe(
    audit(() => handlingFinished$),
    repeat(), // 每次处理完成后重新激活audit监听
    // 当原stream$完成时终止整个流
    takeUntil(stream$.pipe(last(null, 'stream-completed')))
  );

  return eventStream$.pipe(
    concatMap(input => from(handler(input))),
    tap(() => handlingFinished$.next())
  );
}

说明 #

• 方案一更简洁直观，适合大多数场景，且支持自定义处理后的等待时间
• 方案二更贴近原实现思路，适合需要保留audit逻辑的场景

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