RxJS 重试未按预期运行？

Question

我在让 RxJS 的重试运算符与 map 运算符结合使用时遇到了一些困难。

基本上我要做的是获取一个值流（在本例中为 GitHub 用户用于测试目的），并为每个值对该值执行一些操作。

但是，我希望能够以允许我为该操作设置有限重试次数的方式处理错误。我还假设 RxJS 异步处理这样的事情。

Rx.Observable.fromPromise(jQuery.getJSON("https://api.github.com/users"))
.concatAll(x => Rx.Observable.of(x))
.map(x => {
    if(x.id % 5 == 0) {
        console.log("error");
        return Rx.Observable.throw("error");
    }
    return x;
})
.retry(3)
.subscribe(x => console.log(x), e => console.log(e));

这是我到目前为止所得到的，但是它应该重试相同的值 3 次，它只会打印一次“错误”，这不是我正在寻找的行为。

我是否遗漏了一些琐碎的事情，或者 RxJS 根本不可能做到这一点？简而言之，我想对值异步执行操作，将所述操作重新绑定有限次，但允许其他有效操作并行继续。我只使用 promises 实现了类似的东西，但代码相当混乱，我希望 RxJS 能让我让事情变得更好读。

Answer 1

编辑 - 为清晰起见删除早期失败的解决方案（某些 cmets 可能不再相关）

要重试单个失败，您需要中断单个请求并重试这些请求。

为了允许正确处理错误，请将流拆分为每个预期结果的子流。

我还添加了一些代码以允许一次重试成功。

const Observable = Rx.Observable

// Define tests here, allow requestAgain() to pass one
const testUser = (user) => user.id % 5 !== 0
const testUserAllowOneSuccess = (user) => user.id === 5 || user.id % 5 !== 0

const requestAgain = (login) => Observable.of(login)
  .switchMap(login => jQuery.getJSON("https://api.github.com/users/" + login ))
  .map(x => {
    if(!testUserAllowOneSuccess(x)) {
      console.log("logging error retry attempt", x.id);
      throw new Error('Invalid user: ' + x.id)
    }
    return x;
  })
  .retry(3)
  .catch(error => Observable.of(error))

const userStream = Observable.fromPromise(jQuery.getJSON("https://api.github.com/users"))
  .concatAll()

const passed = userStream.filter(x => testUser(x))
const failed = userStream.filter(x => !testUser(x))
  .flatMap(x => requestAgain(x.login))

const retryPassed = failed.filter(x => !(x instanceof Error))
const retryFailed = failed.filter(x => (x instanceof Error))
  .toArray()
  .map(errors => { throw errors })

const output = passed.concat(retryPassed, retryFailed)

output.subscribe(
  x=> console.log('subscribe next', x.id ? x.id : x), 
  e => console.log('subscribe error', e)
);

工作示例CodePen

Answer 2

我发现有几个问题可能会导致您遇到问题：

1) 在你的 map 语句中，你通过返回一个 observable 来进行抛出。如果成功案例还返回了一个您正在展平的可观察对象（使用 concat、switch 或合并），那么这将是合适的，但是您的成功案例只是发出一个值而从不展平。因此，当返回 Throw observable 时，它​​永远不会被展平，因此只会通过并记录 observable 对象。

相反，你可以只做一个普通的 javascript throw 'error'（尽管考虑抛出一个错误对象而不是一个字符串）

2) 当重试出错时，它会重新订阅。如果您使用 observables 进行获取，这将意味着它将再次进行获取。但是，您使用 Promise 进行了获取，然后仅从该 Promise 中创建了一个 observable。因此，当您重新订阅 observable 时，它​​将最终使用与第一次相同的 Promise 对象，并且该 Promise 已经处于已解决状态。不会进行额外的提取。

要解决此问题，您需要有一个可观察的版本来进行提取。如果你需要从头开始，它可能看起来像这样：

Rx.Observable.create(observer => {
  jQuery.getJSON("https://api.github.com/users")
    .then(result => {
      observer.next(result);
      observer.complete();
    }, err => {
      observer.error(err);
      observer.complete();
    });
})

如果您打算经常这样做，我建议您创建一个为您进行包装的函数。

编辑：

我的最终目标是基本上能够获取一组 api url，并使用每个 URL 启动 get 请求。如果 get 请求由于某种原因失败，它应该总共尝试 3 次才能再次到达该端点。但是，此类获取请求的失败不应阻止其他请求的发生，它们应该并行发生。此外，如果一个 get 请求失败 3 次，它应该被标记为失败并且它不应该停止整个过程。到目前为止，我还没有找到使用 RxJS 的方法。

然后我会做这样的事情：

function getJson (url) {
  return Rx.Observable.create(observer) => {
    jQuery.getJSON(url)
      .then(result => {
        observer.next(result);
        observer.complete();
      }, err => {
        observer.error(err);
        observer.complete();
      });
  }
}

const endpoints = ['someUrl', 'someOtherUrl'];
const observables = endpoints.map(endpoint => 
  return getJson(url)
    .retry(3)
    .catch(err => Rx.Observable.of('error'));
});

Rx.Observable.forkJoin(...observables)
  .subscribe(resultArray => {
    // do whatever you need to with the results. If any of them
    //   errored, they will be represented by just the string 'error'
  });

Answer 3

您的地图函数返回Rx.Observable.throw("error");。这是错误的，因为下面的retry() 将收到它的包装（即Observable<Observable<T>> 而不是Observable<T>。您可以使用flatMap 来纠正它，但是您还需要通过Rx.Observable.of(x) 包装该值.

.flatMap(x => {
    if(x % 5 == 0) {
        console.log("error");
        return Rx.Observable.throw("error");
    }
    return Rx.Observable.of(x);
})
.retry(3)
.subscribe(x => console.log(x), e => console.log(e));

另一种选择是throw new Error()，而不是尝试return Rx.Observable.throw("error");

---

确保你在正确的事情上重试

Rx.Observable.fromPromise(jQuery.getJSON("https://api.github.com/users"))
  .retry(3)
  .concatAll(x => Rx.Observable.of(x))
  ...

Answer 4

然后用from 转换promise 是不可重复的

示例：https://stackblitz.com/edit/rxjs-dp3md8

但是，Observable.create 效果很好

如果有人有解决方案可以重复承诺，请发表评论。