我应该使用 shareReplay 作为最后一个运算符吗？

Question

我开发了一个 Angular 9 应用程序，但我不明白如何将shareReplay运算符与其他运算符一起使用。我做了类似以下的事情：

if (!this.cache[key]) {
  this.cache[key] = this.http.get(...).pipe(
    shareReplay(1),
    flatMap(...),
    map(...),
    reduce(...)
  );
}
return this.cache[key];

此后我的应用程序的 CPU 使用率一直为 100%。当我将其更改为：

if (!this.cache[key]) {
  this.cache[key] = this.http.get(...).pipe(
    flatMap(...),
    map(...),
    reduce(...),
    shareReplay(1)
  );
}
return this.cache[key];

看起来效果很好。是否有必要将shareReplay运算符作为最后一个使用？如此高的 CPU 使用率从何而来？

编辑：更详细的代码片段：

this.http.get(...).pipe(
  // I would like to avoid several same http calls
  shareReplay(1),
  // I would like to flatten an array of objects that comes from backend
  flatMap(option => option),
  // I need to map all objects to a format that is acceptable by some library
  map(option => ({
    value: option.key,
    label: option.value
  })),
  // I must reduce it back to an array of the new objects
  reduce((acc: { value: string; label: string }[], option) => {
    acc.push(option);
    return acc;
  }, [])
);

Answer 1

运营商在管道中的位置share很重要，因为它决定了有多少订阅者会订阅前面的运营商share。

举个例子：

const interval$ = interval(1000).pipe(
  tap(() => console.log("Interval Triggered")
);

interval$.subscribe();
interval$.subscribe();

您将看到我们每秒收到两条控制台消息。这是有道理的，因为在 RxJS 中，每个订阅者（“消费者”）都会导致工厂创建一个新的“生产者”（这里是 a setInterval）。在这种情况下 - 没有share运营商 - 每个订阅者都会沿着运营商链向上订阅前一个运营商 - 直到我们订阅interval两次。

现在，让我们看一下share：

const interval$ = interval(1000).pipe(
  tap(() => console.log("Interval Triggered"),
  share()
);

interval$.subscribe();
interval$.subscribe();

在这个版本中，我们每秒只能收到一条消息。为什么？

因为share只要有一个活跃订阅者就不会订阅它的源。所以，我们订阅了share()两次，但share()只订阅了interval一次。

如果我们交换位置，我们会得到不同的结果：

const interval$ = interval(1000).pipe(
  share(),
  tap(() => console.log("Interval Triggered"),
);

interval$.subscribe();
interval$.subscribe();

在本例中，我们订阅了tap两次 - 这两个tap操作员订阅了share两次。这就是为什么我们每秒收到两条控制台消息。

根据经验，您很可能希望您share位于管道的末端，因为这将导致之前的运算符share仅被订阅一次。

解释一下你的CPU消耗：我怀疑你经常订阅这个Observable - 而share不是最后一个操作员，这意味着你会flatMap一遍又一遍地重新订阅等。