带有 BehaviorSubject 和异步管道的 RxJS share() 运算符 - Angular

Question

我有一个 BehaviorSubject 正在作为可观察对象使用：

testForStack$: Observable<boolean>;

ngOnInit(){
    const bs = new BehaviorSubject(true);
    this.testForStack$ = bs
      .asObservable()
      .do(t => console.log('subscribed'))
      .share();
}

这个 observable 正在通过模板中的三个 async 管道传输：

Sub1: {{testForStack$ | async}}<br>
Sub2: {{testForStack$ | async}}<br>
Sub3: {{testForStack$ | async}}

问题只是第一个 (Sub1) 正在获取 true 的值

Sub1: true
Sub2: 
Sub3:

如果我删除 .share()，所有三个值都将获得 true 值，但这会导致多个订阅的问题。

有没有想过为什么使用 BehaviorSubject 会导致这种行为？它被用作可观察对象，所以我假设上面的代码可以正常工作。

这也与此类似。

Answer 1

这是正确的行为。 share() 运算符只保留对其父项的一个订阅，而 BehaviorSubject 仅在订阅时发出其值。

这意味着当您使用第一个 {{testForStack$ | async}} 时，它会在链的末尾订阅 share()，后者会订阅其父级，从而导致订阅源 BehaviorSubject立即发出它的值。

但是，第二个和所有连续的 {{testForStack$ | async}} 订阅了 share()，它已经订阅了它的父级并且不会再进行订阅，所以没有什么可以将源值推送给这些观察者.

一个简单的解决方案可能是使用 shareReplay(1)（取决于你的 RxJS 版本）你应该使用 publishReplay(1).refCount() 而不是因为这些问题（或它的 pipable 等价物）：

Answer 2

不要使用共享运算符。而是做这样的事情：

<ng-container *ngIf="testForStack$ | async as testForStack">

  Sub1: {{ testForStack }}
  Sub2: {{ testForStack }} 
  Sub3: {{ testForStack }}

</ng-container>

还有其他各种方法，例如，如果您不喜欢使用 *ngIf，则可以对使用 ngTemplateOutlet 的模板使用相同的方法。这种方法允许您以相同的方式创建别名变量：

<ng-template
  let-testForStack 
  [ngTemplateOutletContext]="{ $implicit: testForStack$ | async }"
  [ngTemplateOutlet]="selfie" #selfie>

  Sub1: {{ testForStack }}
  Sub2: {{ testForStack }} 
  Sub3: {{ testForStack }}

</ng-template>

此 ng-template 代码是自引用的（有效）并且完全未经测试但“应该”有效并避免使用 *ngIf

在此处阅读更多内容：

https://nitayneeman.com/posts/using-single-subscription-for-multiple-async-pipes-in-angular/

Answer 3

RxJS 7 / 2022 更新：

我发现这个问题是因为我开始将 share 添加到以 BehaviorSubject 开头的管道，但包含昂贵的中间转换，每个订阅者运行ning 一次。

我认为适合我的模式是

const transformedValues$ = subj.pipe(
  map(expensiveTransform),
  shareReplay({bufferSize: 1, refCount: true})
);

这解决了@martin 关于需要手动包含 refCount 的评论。如果 $transformedValues 没有订阅者，则转换根本不应该运行。在第一次订阅时，转换应该运行一次；随后的订阅者应该从 shareReplay 运营商处获得重播。

带有 BehaviorSubject 和异步管道的 RxJS share() 运算符 - Angular

RxJS share() operator with BehaviorSubject and async pipe - Angular

angular

rxjs

behaviorsubject