代码之家 › 专栏 › 技术社区 › Alexander Abakumov

对每个可观察到的值发射运行一次操作

reactivex reactive-programming rxjs typescript javascript

Alexander Abakumov · 技术社区 · 8 年前

假设我们有以下代码( Stackblitz ):

  public counter: number = 0; // Some arbitrary global state

  public ngOnInit(): void {
    const subj: Subject<number> = new Subject<number>();

    const obs: Observable<number> = subj.asObservable().pipe(
      tap((): void => {
        ++this.counter;
      }));

    // this.counter increments for each subscription.
    // But supposed to increment once per obs new value emition.
    obs.subscribe();
    obs.subscribe();
    obs.subscribe();

    subj.next(1);

    // Now this.counter === 3
  }

结果是 this.counter 等于 3 执行上述操作后,每次订阅都会递增一次。

但是我们怎么能每 Observable 以一种被动的方式释放新的价值?

我知道我们应该避免被动流之外的状态,但在现实世界的应用程序中仍然需要时间来计时。

2 回复 | 直到 8 年前

martin 8 年前

你看到数字的原因 3 有点不同。当你使用 of(1) 它发出一个 next 通知之后 complete 通知。每次你打电话都是这样 obs.subscribe() .

所以如果你想看 1 您需要创建一个中间主题,在这里您进行所有订阅,然后这个中间主题订阅可观察的源(使用 publish() 最容易为您创建主题的接线员):

const obs: ConnectableObservable<number> = of(1).pipe(
  tap((): void => {
    ++this.counter;
  }),
  publish(),
) as ConnectableObservable<number>;

obs.subscribe();
obs.subscribe();
obs.subscribe();

obs.connect();

您的更新演示: https://stackblitz.com/edit/rxjs-update-global-state-ejdkie?file=app/app.component.ts

请注意 (1) 将发射 完成 不管怎样,通知意味着里面的主题 发布() 也将完成。这对您来说可能是个问题,但这确实取决于您的用例。

编辑 :这是更新的演示,其中源是主题: https://stackblitz.com/edit/rxjs-update-global-state-bws36m?file=app/app.component.ts

Nandita Sharma 8 年前

请试一下这样的东西。我哪儿也跑不了。如果有错误请告诉我

  public counter: number = 0;
  public items: any = [];

  public ngOnInit(): void {
    const obs: Observable<number> = of(1).pipe(
      tap((value): void => {
        if(! Array.prototype.includes(value)) {
           this.items.push(value);
            ++this.counter;
         }
      }));

    // this.counter increments for each subscription.
    // How to make it increment once per obs new value emission.
    obs.subscribe(1);
    obs.subscribe(2);
    obs.subscribe(1);

    // Now this.counter === 3
  }

推荐文章

Diafotismos · 启动RxJava2 Observable。间隔()没有初始延迟?

8 年前

user6791424 · 如何返回可观察到的。mongo光标查询中的fromEvent?

8 年前

Manuel Huber · RxJS并非所有订阅者都收到所有事件

8 年前

thiagoh · RxJS可观察concat不工作

9 年前

PerrierCitror · Rx:可观察到的力至少需要N秒才能完成

9 年前

TardigradeX · 如何使用ReactiveX观察对象/主题处理进度更新?

9 年前

rumdrums · 可观察的ReactiveX批处理结果

9 年前

Per Hornshøj-Schierbeck · 我如何使用可观察项而不是承诺?

9 年前

Alex X · RxJS谷歌地图标记

9 年前

Timofey Solonin · 订阅UIButton.rx。点击位于UITableViewDataSource中的UITableViewCell

9 年前