操作符 - 错误处理

优质

小牛编辑

128浏览

2023-12-01

有两种主要的方法来处理流中的错误。你可以重试流并保证流最终会正常运行，或者处理错误并进行转换。

重试 - 现在怎么样？

当你认为错误是由于某些原因是暂时导致的，那么这种方法是适用的。通常不稳定的网络是个很好的例子。当网络不稳定时端点可能会在你多次尝试后才能回应。要点是你的首次尝试可能失败，但重试x次并且在两次尝试之间有一定的时间间隔，最终端点会回应。

retry

retry() 操作符可以让我们重试整个流，只接收一个参数，参数的值是要重试的次数，函数签名如下：

retry([times])

重要的是要注意当错误回调被调用的话， retry() 操作符会有延迟。下面代码中的错误回调会立即被调用：

let stream$ = Rx.Observable.of(1,2,3)
.map(value => {
   if(value > 2) { throw 'error' }
});
stream$.subscribe(
   data => console.log(data),
   err => console.log(err)
)

这个流很快的就死了，错误回调被调用，这个时候 retry() 操作符登场。像下面这样把它附加上即可：

let stream$ = Rx.Observable.of(1,2,3)
.map(value => {
   if(value > 2) { throw 'error' }
})
.retry(5)

这将运行值序列5次，最后放弃并进入错误回调。然而在这个案例中，由于编写代码的方式，它只会生成5次1,2。所以我们的代码并没有真正利用操作符的最大潜力。你可能想要的是能够在每次尝试之间改变一些东西。想象下你的 observable 看起来像这样：

let urlsToHit$ = Rx.Observable.of(url, url2, url3);

在这一点上，它清楚地表明，在你的第一次尝试中，端点可能回应的不好，或者根本就没有回应，所以重试x次是很有用的。

然而在调用 ajax 的情况下，并想象一下我们的业务场景中网络不稳定，那么立即重试是没有意义的，所以我们需要再找到一个更好的操作符，那就是 retryWhen()

retryWhen

retryWhen() 操作符让我们有机会对流进行操作并恰当地处理。

retryWhen( stream => {
   // 希望能在更好的条件下返回
})

现在我们来写段简单的代码：

let values$ = Rx.Observable
.of( 1,2,3,4 )
.map(val => {
    if(val === 2) { throw 'err'; }
    else return val;
})
.retryWhen( stream => {
    return stream;
} );
values$.subscribe(
    data => console.log('Retry when - data',data),
    err => console.error('Retry when - Err',err)
)

这样写的话会一直返回 1，直到我们用完内存为止，由于缺少结束条件，算法总是会在值2上崩溃，并将永远重试流。我们需要做的就是以某种方式告知错误已经修复。如果流尝试点击网址而不是发出数字，响应端点将会被压垮，所以在这种情况下，我们必须写这样的东西：

let values$ = Rx.Observable.interval(1000).take(5);
let errorFixed = false;
values$
.map((val) => {
   if(errorFixed) { return val; }
   else if( val > 0 && val % 2 === 0) {
      errorFixed = true;
      throw { error : 'error' };
   } else {
      return val;
   }
})
.retryWhen((err) => {
    console.log('retrying the entire sequence');
    return err.delay(200);
})
.subscribe((val) => { console.log('value',val) })
// 0 1 '等待200毫秒' retrying the whole sequence 0 1 2 3 4

然而，这与我们用 retry() 运算符所做的很多类似，上面的代码只会重试一次。使用 retryWhen() 真正的好处是可以改变操作符中返回的流，也就是这里调用的 delay() 操作符，像这样：

.retryWhen((err) => {
    console.log('retrying the entire sequence');
    return err.delay(200)
})

这会确保在流重试前有200毫秒的延迟，如果是在 ajax 场景下，可以确保端点有足够的时间重整旗鼓，然后开始响应。

陷阱

在 retryWhen() 中使用 delay() 操作符来确保重试晚一点发生，在这个案例中可以给网络一个恢复的机会。

retryWhen 和 delay 一起使用没有次数限制

到目前为止，当我们想要重试整个流x次时使用的是 retry() 操作符，当我们想要在重试之间有一些延迟时间时使用的是 retryWhen() 操作符，但是如果我们两者都想要，可以做到吗？可以的。我们需要考虑一下要以某种方式记住到目前为止我们的尝试次数。引入一个外部变量用来保持这个数量是非常诱人的，但那不是函数式做事的方式，记住副作用是被禁止的。那么我们该如何解决呢？有一个叫做 scan() 的操作符，它允许我们累积每次迭代的值。所以如果在 retryWhen() 中使用 scan 的话，我们就可以追踪尝试的次数：

let ATTEMPT_COUNT = 3;
let DELAY = 1000;
let delayWithTimes$ = Rx.Observable.of(1,2,3)
.map( val => {
  if(val === 2) throw 'err'
  else return val;
})
.retryWhen(e => e.scan((errorCount, err) => {
    if (errorCount >= ATTEMPT_COUNT) {
        throw err;
    }
    return errorCount + 1;
}, 0).delay(DELAY));
delayWithTimes$.subscribe(
    val => console.log('delay and times - val',val),
    err => console.error('delay and times - err',err)
)

转换 - 这个没什么好看的

这个方法是当出现错误时你选择将错误重制成一个有效的 Observable 。

所以我们可以通过创建一个 Observable 来体现这一点，这个 Observable 的使命就是报错

let error$ = Rx.Observable.throw('crash');
error$.subscribe(
  data => console.log( data ),
  err => console.log( err ),
  () => console.log('complete')
)

这段代码只会执行错误回调而不会执行完成回调。

修补它

我们可以通过引入 catch() 操作符来进行修补。它是这样使用的：

let errorPatched$ = error$.catch(err => { return Rx.Observable.of('Patched' + err) });
errorPatched$.subscribe((data) => console.log(data) );

如你所见，使用 .catch() 进行修补并返回一个新的 Observable 修复流。问题是这是否是你想要的。流确实存活下来最终完成了，它可以发出崩溃之后发生的任何值。

如果这不是你想要的，那么上面的重试方法可能会更适合你，决定权在你手中。

多个流呢？

你没想到会这么容易吧？当你编写 RxJS 代码时，通常会处理多个流，如果你知道在哪放置 catch() 操作符的话，那么使用 catch() 的方法是很棒的。

let badStream$ = Rx.Observable.throw('crash');
let goodStream$ = Rx.Observable.of(1,2,3,);
let merged$ = Rx.Observable.merge(
  badStream$,
  goodStream$
);
merged$.subscribe(
   data => console.log(data),
   err => console.error(err),
   () => console.log('merge completed')
)

猜猜发生了什么？1）错误和值都发出了，流也完成了 2）错误和值都发出了 3）只发出了错误

遗憾的是发生的是 3）。这意味着我们几乎没有处理错误。

修复 - 所以我们需要修复错误。我们使用 catch() 操作符进行修复。问题在哪呢？

来试试这个?

let mergedPatched$ = Rx.Observable.merge(
    badStream$,
    goodStream$
).catch(err => Rx.Observable.of(err));
mergedPatched$.subscribe(
    data => console.log(data),
    err => console.error(err),
    () => console.log('patchedMerged completed')
)

在这个案例中，得到结果的是 crash 和 patchedMerged completed 。所以流是完成了的，但我们还是没有得到 goodStream$ 的值。所以这是一个更好的解决方法，但还不够好。

更好的修复 - 在 merge() 后面添加 catch() 操作符可以确保流完成，但是还不够好。我们来尝试下在 merge 之前进行 catch 操作。

let preMergedPatched$ = Rx.Observable.merge(
    badStream$.catch(err => Rx.Observable.of(err)),
    goodStream$
).catch(err => Rx.Observable.of(err));
preMergedPatched$.subscribe(
    data => console.log(data),
    err => console.error(err),
    () => console.log('pre patched merge completed')
)

瞧，我们得到了值，我们捕获了错误并将错误作为一个新的 Observable 发出，并且流也完成了。

陷阱 catch() 所放的位置很重要。

适者生存

还有另外一种情况可能会很有意思。上面的场景假设你希望发出所有的，错误信息、值、所有的一切。

如果你关心的不是这些呢，你只关心流的值怎么办？我们来假设一下，有一个叫做 onErrorResumeNext() 的操作符

let secondBadStream$ = Rx.Observable.throw('bam');
let gloriaGaynorStream$ = Rx.Observable.of('I will survive');
let emitSurviving = Rx.Observable.onErrorResumeNext(
    badStream$,
    secondBadStream$,
    gloriaGaynorStream$
);
emitSurviving.subscribe(
    data => console.log(data),
    err => console.error(err),
    () => console.log('Survival of the fittest, completed')
)

输出的结果是 ‘I will survive’ 和 ‘Survival of the fittest, completed’ 。