RxJS：即使在源 observable 上使用了 `shareReplay()`，`throwError()` 也会为每个 observable 单独执行答案

【问题标题】：RxJS: Even if used `shareReplay()` on source observable, `throwError()` gets executed separately for each observableRxJS：即使在源 observable 上使用了 `shareReplay()`，`throwError()` 也会为每个 observable 单独执行
【发布时间】：2020-05-19 06:45:45
【问题描述】：

我在 observable (courses$) 上使用 RxJS shareReplay() 运算符在其他两个 observables (beginnerCourses$ 和 advancedCourses$) 之间共享 observable 流。它工作正常，成功时两个可观察对象之间共享单个 API 调用响应。

但是，当涉及到错误时，这些 observable 不会共享错误，并且会在浏览器控制台中看到错误被抛出两次。 shareReplay() 操作员也不会共享错误吗？这是一种预期的行为吗？

const http$ = createHttpObservable('/api/courses');

const courses$ = http$
  .pipe(
    map(res => res['payload'] ),
    shareReplay(),
    catchError(err => {
      return throwError(err);
    })
  );

this.beginnerCourses$ = courses$
  .pipe(
    map(courses => courses
      .filter(course => course.category === 'BEGINNER')));

this.advancedCourses$ = courses$
  .pipe(
    map(courses => courses
      .filter(course => course.category === 'ADVANCED')));

}

【问题讨论】：

标签： rxjs rxjs6 rxjs-pipeable-operators

【解决方案1】：

当 observable 抛出错误时，这是预期的行为，shareReplay() 将尝试重新订阅/重新执行源

你可以试试这个来验证

const a=defer(()=>{
  console.log('run')
  return throwError(new Error('Error'))
}).pipe(shareReplay())

a.subscribe(console.log,console.error,()=>console.log('complete'))
a.subscribe(console.log,console.error,()=>console.log('complete'))

https://stackblitz.com/edit/typescript-jutfxe

如果您希望 observable 共享错误而不再次执行改用publishReplay(1),refCount()

【讨论】：

【解决方案2】：

我认为这是预期的行为，有点意外的是您遇到了 2 个不同的错误。

shareReplay 在数据消费者和数据生产者之间放置一个ReplaySubject。当错误通知到达时，正在使用的ReplaySubject 将向所有注册订阅者发送相同错误通知：

error(err: any) {
  if (this.closed) {
    throw new ObjectUnsubscribedError();
  }
  this.hasError = true;
  this.thrownError = err;
  this.isStopped = true;
  const { observers } = this;
  const len = observers.length;
  const copy = observers.slice();
  for (let i = 0; i < len; i++) {
    copy[i].error(err);
  }
  this.observers.length = 0;
}

Source

但是当使用shareReplay 时，当一个新订阅者 即将订阅时，如果发生错误，正在使用的ReplaySubject 将被另一个替换。说它正在被替换，这也需要重新订阅源。

所以我认为所有订阅者都应该收到相同的错误通知，只要他们已经是ReplaySubject 订阅者列表的一部分。否则，当有新订阅者进来时，源将被重新订阅。

您可以做的是防止ReplaySubject接收错误通知并允许其订阅者按原样接收它，是使用materialize和dematerialize运营商：

const courses$ = http$
  .pipe(
    materialize(), // Everything as a `next` notification
    map(res => res['payload'] ),
    shareReplay(),
    dematerialize() // Back to the original event
  );

使用这种方法，如果已注册的订阅者收到错误通知，它将被取消订阅，这意味着它也将从ReplaySubject 的订阅者列表中删除。但ReplaySubject 仍会存在，并且不会在后续订阅者中被替换。

另外，我认为这是非常多余的：

catchError(err => throwError(err));

【讨论】：