代码之家 › 专栏 › 技术社区 › jigfox

重试内部可观察,

rxjs

jigfox · 技术社区 · 8 年前

我有外在的和内在的 Observable 。内部Obersable有时会出错,可以通过重试来解决。

function sm$(val) {
  if (Math.random() > .4) {
    return Rx.Observable.throw(val)
  } else {
    return Rx.Observable.of(val)
  }
}

function sm(val) {
  return Rx.Observable.of(val)
    .switchMap(sm$)
    .catch(() => Rx.Observable.of(val).delay(1000).switchMap(sm))
}

Rx.Observable
  .interval(500)
  .switchMap(sm)
  .take(5)
  .subscribe(
    val => console.log("val:", val),
    err => console.log("err:", err),
    () => console.log("complete")
  )

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/5.4.0/Rx.min.js"></script>

此日志,即:

val: 1
val: 2
val: 6
val: 10
val: 11
complete

当我在catch块中返回一个延迟的observable时,它工作不正常,只是跳过了一些值。毫不迟疑,它按预期工作,但这不是我想要的。我想推迟多次处决。

function sm$(val) {
  if (Math.random() > .4) {
    return Rx.Observable.throw(val)
  } else {
    return Rx.Observable.of(val)
  }
}

function sm(val) {
  return Rx.Observable.of(val)
    .switchMap(sm$)
    .catch(() => Rx.Observable.of(val).switchMap(sm))
    // removed the delay(1000)
}

Rx.Observable
  .interval(500)
  .switchMap(sm)
  .take(5)
  .subscribe(
    val => console.log("val:", val),
    err => console.log("err:", err),
    () => console.log("complete")
  )

<脚本src=”https://cdnjs.cloudflare.com/ajax/libs/rxjs/5.4.0/Rx.min.js“></script>

这个日志总是这样,这就是我所期望的:

val: 0
val: 1
val: 2
val: 3
val: 4
complete

2 回复 | 直到 8 年前

martin 8 年前

问题是你正在使用 Observable.interval(500).switchMap(sm) 。

尤其是 switchMap 在你的情况下。来源 Observable.interval 即使在使用 delay(1000) 就因为 开关映射 当它想在1000毫秒后发射时,它已经取消订阅。

看来你可以用 concatMap 而不是外部 开关映射 。

function sm$(val) {
  if (Math.random() > .4) {
    return Rx.Observable.throw(val)
  } else {
    return Rx.Observable.of(val)
  }
}

function sm(val) {
  return Rx.Observable.of(val)
    .switchMap(sm$)
    .catch(() => Rx.Observable.of(val).delay(1000).switchMap(sm))
}

Rx.Observable
  .interval(500)
  .concatMap(sm)
  .take(5)
  .subscribe(
    val => console.log("val:", val),
    err => console.log("err:", err),
    () => console.log("complete")
  )

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/5.4.0/Rx.min.js"></script>

Picci 8 年前

我认为问题与这样一个事实有关,即使用1000 ms的延迟,同时在发出5个事件后完成流,每个事件每500 ms发生一次。

这意味着延迟排放,延迟是因为 catch 块,可以在流完成后到达。

如果你减少延迟,你可能会得到你想要的。我还添加了一些日志以获得更好的图片

import {Observable} from 'rxjs';

function sm$(val) {
    if (Math.random() > .4) {
      return Observable.throw(val)
    } else {
      return Observable.of(val)
    }
}

function sm(val) {
    return Observable.of(val)
      .switchMap(sm$)
      .catch(() => {
                        console.log('error', val);
                        return Observable.of(val).delay(10).switchMap(d => sm(d));
                    })
}

Observable
    .interval(500)
    .switchMap(sm)
    .take(5)
    .subscribe(
      val => console.log("val:", val),
      err => console.log("err:", err),
      () => console.log("complete")
    )