RxJS 如何用 catch 忽略错误并继续前进

您需要在其中使用 flatMap 运算符进行过滤。在此示例的 flatMap 中，我使用 Observable.if（） 进行过滤，因为它保证我一直在返回可观察对象。我相信你可以用其他方式做到这一点，但这对我来说是一个干净的实现。

const source = Rx.Observable.interval(100).take(10).flatMap((x)=>
    Rx.Observable.if(() => x !== 4, 
    Rx.Observable.of(x),
    Rx.Observable.throw("Bad value"))
    .catch((err) => {
        return Rx.Observable.empty()
    })
);

source.subscribe(v => console.log(v));

您将希望保持源可观察对象运行，但如果您让错误发生在主事件流上，它将折叠整个可观察对象，并且您将不再接收项目。

该解决方案涉及创建一个单独的流，您可以在其中进行过滤和捕获，而不会让上游管道塌陷。

const Rx = require('rxjs/Rx');
function checkValue(n) {
  if(n === 4) {
    throw new Error("Bad value");
  }
  return true;
}
const source = Rx.Observable.interval(100).take(10);

source
  // pass the item into the projection function of the switchMap operator
  .switchMap(x => {
     // we create a new stream of just one item
     // this stream is created for every item emitted by the source observable
     return Observable.of(x)
       // now we run the filter
       .filter(checkValue)
       // we catch the error here within the projection function
       // on error this upstream pipe will collapse, but that is ok because it starts within this function and will not effect the source
       // the downstream operators will never see the error so they will also not be effect
       .catch(err => Rx.Observable.empty());
     })
     .subscribe(v => console.log(v));

您还可以使用传递给 catch 选择器的第二个参数来重新启动可观察源，但这会启动它，就好像它以前没有运行过一样。

const Rx = require('rxjs/Rx');

function checkValue(n) {
  if(n === 4) {
    throw new Error("Bad value");
  }
  return true;
}
const source = Rx.Observable.interval(100).take(10);

source.filter(x => checkValue(x))
  .catch((err, source) => source)
  .subscribe(v => console.log(v));

但这并没有达到预期的效果。您将看到一个重复发出 1..3 的流，直到时间结束......或者关闭脚本。以先到者为准。（这是必不可少的）.retry()