angular - 如何最好地处理模板上同一可观察对象的多个订阅？

Question

假设我有一个名为“todoList$”的可观察对象。使用“异步”运算符，我可以自动订阅/取消订阅它。下面代码中的问题是对同一个可观察对象有两个相同的订阅:

<ng-container *ngIf="(todoList$ | async).length > 0>
  <div *ngFor="let todo of todoList$ | async">
    ...

这不是很 DRY，因此，我们为可以更有效处理的订阅分配内存。

由于 ngIf 条件中的语法，我认为我无法使用“as”关键字为可观察输出创建模板变量。 相反，当我使用组件文件中的 RxJs ‘share’ 运算符时有效:

todoList$ = this.store.select(todoList).pipe(tap(x => {console.log('testing')}), share());
//testing  

如果没有共享运算符，“testing”会被打印两次。这让我相信 share() 运算符可以解决这个问题。如果是，不确定为什么/如何？由于这可能是一个普遍存在的问题/代码异味，处理同一模板中相同的多个订阅的最佳方式是什么？

我承认 StackOverflow 上有一些类似的问题。但没有人能准确地给出我正在寻找的东西。

Answer 1

If it does, not exactly sure why/how?

让我们看看如何share() is defined :

function shareSubjectFactory() {
  return new Subject<any>();
}

return (source: Observable<T>) => refCount()(multicast(shareSubjectFactory)(source)) as Observable<T>;

首先，

(source: Observable<T>) => refCount()(multicast(shareSubjectFactory)(source))

与

相同

(source: Observable<T>) => source.pipe(
  multicast(shareSubjectFactory),
  refCount()
)

multicast 将返回 ConnectableObservable ，它仍然是一个 Observable，但除其他外，它公开了一个 connect 方法。

// Inside `multicast` operator

const connectable: any = Object.create(source, connectableObservableDescriptor);
connectable.source = source;
connectable.subjectFactory = subjectFactory;

return <ConnectableObservable<R>> connectable;

Source

另一个有趣的事情是，当订阅时，订阅者将被添加到Subject的订阅者列表中，而主源不会 被订阅，直到 connect 被调用:

_subscribe(subscriber: Subscriber<T>) {
  return this.getSubject().subscribe(subscriber);
}

protected getSubject(): Subject<T> {
  const subject = this._subject;
  if (!subject || subject.isStopped) {
    this._subject = this.subjectFactory();
  }
  return this._subject!;
}

例如:

const src$ = privateSrc.pipe(
  tap(() => console.log('from src')),
  share(),
  tap(() => console.log('from share()')),
)

当 src$ 被订阅时:

// Subscriber #1
src$.subscribe(/* ... */)

订阅者将被添加到Subject 的订阅者列表中，源src$ 将被订阅。为什么？因为 share 还使用 refCount，如果在没有以前的活跃订阅者时注册了新订阅者，它会订阅到源并且将如果没有更多的活跃订阅者，请从来源退订。

让我们来看另一个例子:

const src$ = (new Observable(s => {
  console.warn('[SOURCE] SUBSCRIBED')

  setTimeout(() => {
    s.next(1);
  }, 1000);
})).pipe(share());

// First subscriber,
// because it's the first one, `refCount` will to its job and the source will be subscribed
// and this subscriber will be added to the `Subject`'s subscribers list
// note that the source sends the data asynchronously
src$.subscribe(/* ... */)

// The second subscriber
// since the source is already subscribed, `refCount` won't subscribe to it again
// instead, this new subscriber will be added to `Subject`'s list
src$.subscribe(/* ... */)

在 1s 之后，源将发送值 1，主题将收到该值并将其发送给其注册订阅者。

这是 how refCount发挥它的魔力:

// When a new subscriber is registered

(<any> connectable)._refCount++;

// `RefCountSubscriber` will make sure that if no more subscribers are left
// the source will be unsubscribed
const refCounter = new RefCountSubscriber(subscriber, connectable);

// Add the subscriber to the `Subject`'s list
const subscription = connectable.subscribe(refCounter);

if (!refCounter.closed) {
  (<any> refCounter).connection = connectable.connect();
}

return subscription;

并且 ConnectableObservable.connect 被定义为 as follows :

connect(): Subscription {
  let connection = this._connection;
  
  if (!connection) {
    // If the source wasn't subscribed before

    this._isComplete = false;
    connection = this._connection = new Subscription();
    
    // Subscribing to the source
    // Every notification send by the source will be first received by `Subject`
    connection.add(this.source
      .subscribe(new ConnectableSubscriber(this.getSubject(), this)));
    
    /* ... */
  }
  return connection;
}

---

因此，如果我们有一个需要在模板中多次订阅的 src$ observable，我们可以应用上述概念。

但是，我们应该注意一个重要方面。

如果我们的模板是这样的:

<!-- #1 -->
<div *ngIf="src$ | async"></div>

<!-- ... -->

<!-- #2 -->
<div *ngIf="src$ | async"></div>

和src$:

src$ = store.pipe(select(/* ... */), share())

然后，如果 store 已经有值，它将被同步检索，这意味着当 #1 将被注册时，store 将被订阅并发送该值，但请注意此时 #2 尚未订阅，因此它不会收到任何东西.

如果 source 是异步的，那么我们应该没有问题，因为模板中的订阅很可能是同步。

但是，当源是同步时，您可以这样解决这个问题:

src$ = store.pipe(
  select(/* ... */),
  subscribeOn(asyncScheduler),
  share()
)

subscribeOn(asyncScheduler) 大致与使用 setTimeout(() => {}, 0) 延迟订阅源相同。但是，这允许订阅 #2，这样当源最终被订阅时，两个订阅者都将收到该值。