RxJS中debounceTime与merge配合失效问题排查及单流多值需求解决方案

最近碰到一个业务场景：在表单输入框（inputController）输入内容时，需要让数据流立刻发射null，等后端校验请求完成后，再返回后端的true/false结果（任何错误情况都返回false）。

最开始写的代码遇到了问题——debounceTime完全没起作用，而且没法实现“输入立即发null”的要求：

slugChecker<PipeInput, Out>(
  inputController: AbstractControl,
  pipeInput: Observable<PipeInput>,
  checker: (input: PipeInput & { name: string, slug: string }) => Observable<Out>,
  opt?: { debounceTime?: number; ignoreFirstValue?: boolean }
) {
  return inputController.valueChanges.pipe(
    shareReplay(1),
    debounceTime(opt?.debounceTime || 1000), // <= 无法正常工作
    // startWith(true),
    skip(opt?.ignoreFirstValue ? 1 : 0), // 编辑模式下跳过首次检查
    switchMap(e => merge(
      of(null)
        .pipe(tap(_ => console.log('set null'))),
      of(e)
        .pipe(
          // delay(2000), <= 不合理的方案
          tap(_ => console.log('run pipe', opt)),
          map(input => slugify(input, { replacement: '-', remove: /[*+~.()'"!:@]/g, locale: 'pl', lower: true, })),
          switchMap(slug => pipeInput.pipe(map(input => ({input, slug})))),
          switchMap(({input, slug}) => {
            console.log('mapped with ', {input, slug})
            return checker({...input, name: inputController.value, slug})
          }),
        )
    )),
  )
}

问题根源 #

后来想明白，问题出在流的结构上：我把debounceTime放在了整个流的最上游，这导致所有后续操作（包括发射null）都要等防抖时间结束才会触发，完全违背了“输入立即发null”的需求。而且用merge把null流和请求流嵌套在switchMap里，逻辑上就错了——只有等防抖后的事件触发，才会进入switchMap生成这两个流，根本做不到输入时立刻响应。

修正后的解决方案 #

把逻辑拆成两个独立的并行流，再用merge合并：

1. 第一个流：监听valueChanges，每次输入直接映射成null，保证输入时立刻发射
2. 第二个流：对valueChanges做防抖，再执行slug生成、后端请求等逻辑，返回最终校验结果

修改后的代码可以正常工作：

slugChecker<PipeInput>(
  inputController: AbstractControl,
  pipeInput: Observable<PipeInput>,
  checker: (input: PipeInput & { name: string, slug: string }) => Observable<boolean>,
  opt?: { debounceTime?: number; ignoreFirstValue?: boolean, slugController?: AbstractControl, }
) {
  return merge(
    // 输入时立即发射null
    inputController.valueChanges.pipe(
      map(_ => null)
    ),
    // 防抖后执行后端校验，返回结果
    inputController.valueChanges.pipe(
      shareReplay(1),
      debounceTime(opt?.debounceTime || 1000),
      map(input => slugify(input, { replacement: '-', remove: /[*+~.()'"!:@]/g, locale: 'pl', lower: true, })),
      switchMap(slug => pipeInput.pipe(map(input => ({input, slug})))),
      switchMap(({input, slug}) => {
        console.log('mapped with ', {input, slug})
        if (opt?.slugController) {
          opt.slugController.patchValue(slug)
        }
        return checker({...input, name: inputController.value, slug})
      }),
    )
  );
}

这样调整后，每次输入都会立刻发射null，同时防抖后的流会在指定时间后执行后端请求，完成后发射校验结果，完全符合业务需求。

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