如何让RxJS Observable遇错误仍保持活跃（忽略错误）

我需要在RxJS管道中处理一系列值，当遇到错误时返回null并继续处理剩余输入。示例输入流如下：

of('foo', 'bar', 'error', 'bazz', 'nar', 'error', 'car').subscribe(...)

我写了如下处理逻辑：

async function failIfError(item: string) {
  if (item === 'error') throw new Error('HTTP 500'); // 不可控的错误
  return item;
}

of('foo', 'bar', 'error', 'bazz', 'nar', 'error', 'car')
  .pipe(
    concatMap((item) => failIfError(item)),
    catchError((error) => of(null))
  )
  .subscribe((val) => console.log('value:', val));

当前实际输出：

value: foo
value: bar
value: null

预期输出：

value: foo
value: bar
value: null
value: bazz
value: nar
value: null
value: car

请问该如何修改代码？

问题出在顶层的catchError会终止整个流：当第一个error触发错误时，catchError会替换整个源Observable为of(null)，后续的bazz、nar等值都不会再被处理。

正确的做法是把错误处理逻辑放到concatMap内部，让每个单独的任务（对应每个输入值）自己处理错误，这样单个任务的错误不会影响整个流的继续执行：

async function failIfError(item: string) {
  if (item === 'error') throw new Error('HTTP 500');
  return item;
}

of('foo', 'bar', 'error', 'bazz', 'nar', 'error', 'car')
  .pipe(
    concatMap((item) => 
      from(failIfError(item)).pipe(
        catchError(() => of(null))
      )
    )
  )
  .subscribe((val) => console.log('value:', val));

关键说明 #

• 因为failIfError是async函数，调用它会返回Promise，需要用from()把Promise转换成Observable，才能使用RxJS的操作符处理错误。
• 把catchError放到concatMap内部的子Observable管道中，每个输入值对应的子流错误只会被当前子流捕获，处理成null后，整个源流会继续处理下一个值。

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