创建一次性订阅

Posted 2023-02-16

【中文标题】创建一次性订阅

【英文标题】：Create one-time subscription

【发布时间】：2015-03-16 11:07:51

【问题描述】：

我需要创建一个Observable 的订阅，它在第一次调用时会立即被处理掉。

有没有类似的东西：

observable.subscribeOnce(func);

我的用例是在快速路由处理程序中创建订阅，并且每个请求都会多次调用订阅。

【参考方案1】：

不是 100% 确定您需要什么，但如果您只想观察第一个值，请使用 first() 或 take(1)：

observable.first().subscribe(func);

注意：.take(1) 和 .first() 在满足条件时都会自动取消订阅

从 RxJS 5.5+ 更新

来自Coderer的评论。

import  first  from 'rxjs/operators'
    
observable
  .pipe(first())
  .subscribe(func);

Here's why

【讨论】：

订阅后会自动清理吗？

是的。满足条件后，先取消订阅。

为什么文档没有说它会自动处理订阅？

这是一条通用的 RxJS 规则，即在可观察流结束时处理订阅。这意味着任何“缩短”流（或将其转换为不同类型的流）的操作员都将在其操作完成时取消订阅源流。我不确定文档中的位置或是否说明了这一点。

如果有人在 2018 年参加此活动，您实际上想要observable.pipe(first()).subscribe(func)，其中first 来自rxjs/operators。

【参考方案2】：

RxJS 有一些我见过的最好的文档。按照下面的链接，您将看到一个非常有用的表映射用例到操作员。例如，在“我想取第一个值”用例下是三个运算符：first、firstOrDefault 和 sample。

注意，如果一个可观察序列在没有通知的情况下完成，那么first 运算符会通知订阅者一个错误，而firstOrDefault 运算符会为订阅者提供一个默认值。

operator use case lookup

【参考方案3】：

更新（2021 年 12 月）：

由于 toPromise() 函数在 RxJS 7 中已被弃用，因此已宣布使用新函数来代替它。 firstValueFrom 和 lastValueFrom。

firsValueFrom 函数解析第一个发出的值并直接取消订阅资源。当 Observable 完成但没有发出任何值时，它会以 EmptyError 拒绝。

另一方面，lastValueFrom 函数在一定程度上与toPromise() 相同，因为它解析了 observable 完成时发出的最后一个值。但是，如果 observable 没有发出任何值，它将以 EmptyError 拒绝。与toPromise() 不同，它在没有值发出时解析undefined。

欲了解更多信息，请查看docs。

---

旧答案：

如果你只想调用一个 Observable 一次，这意味着你不会等待来自它的流。所以在你的情况下使用toPromise() 而不是subscribe() 就足够了，因为toPromise() 不需要取消订阅。

【讨论】：

非常有趣，这也意味着我们可以只使用await promise 这使它成为一个班轮

@LouieAlmeda 你能给我们举个例子吗？

嗨@AdoRen，我已经在这里为您解释了该技术，希望对您有所帮助。

toPromise() 更像observable.pipe(last()) 而不是first()。

是的，“承诺”的答案是：await observable.pipe(take(1)).toPromise()

【参考方案4】：

作为@Brandon's answer的补充，使用first()等对于基于Observable更新BehaviorSubject也是必不可少的。例如（未经测试）：

var subject = new BehaviorSubject(1:'apple',2:'banana');
var observable = subject.asObservable();

observable
  .pipe(
    first(), // <-- Ensures no stack overflow
    flatMap(function(obj) 
      obj[3] = 'pear';
      return of(obj);
    )
  )
  .subscribe(function(obj) 
    subject.next(obj);
  );

【参考方案5】：

干净方便的版本

扩展 M Fuat NUROĞLU 关于将 observable 转换为 Promise 的惊人答案，这是它的非常方便的版本。

const value = await observable.toPromise();

console.log(value)

这样做的好处是我们可以像使用普通变量一样使用该值，而无需引入另一个嵌套块！

当您需要从多个可观察对象中获取多个值时，这特别方便。干净整洁。

const content = await contentObservable.toPromise();
const isAuthenticated = await isAuthenticatedObservable.toPromise();

if(isAuthenticated)
   service.foo(content)

当然，如果你要走这条路线，你必须让你的包含函数async。如果您不希望包含函数异步，您也可以只 .then 承诺

我不确定这种方法是否存在权衡，请随时在 cmets 中告诉我，以便我们知道。

附：如果您喜欢这个答案，请不要忘记也为 M Fuat NUROĞLU 的答案投票 :)

【讨论】：

请注意，在底层 Observable 完成之前，Promise 不会解析。因此，如果 .toPromise() 不起作用，请检查您的 Observable 是否已完成。例如，“take(1)”可能会有所帮助：const value = await observable.pipe(take(1)).toPromise() 查看此评论：github.com/ReactiveX/rxjs/issues/2536#issuecomment-306041463

【参考方案6】：

我也有类似的问题。

Below 甚至后来被不同的状态改变者调用。因为我不想。

function foo() 
    // this was called many times which was not needed
    observable.subscribe(func);
    changeObservableState("new value");

我决定在订阅后尝试unsubscribe()，如下所示。

function foo() 
    // this was called ONE TIME
    observable.subscribe(func).unsubscribe();
    changeObservableState("new value");

subscribe(func).unsubscribe(); 就像subscribeOnce(func)。

希望对你也有帮助。

【参考方案7】：

observable.pipe(take(1)).subscribe() 使用 take 1 订阅一次然后退出

【讨论】：

请添加更多详细信息以扩展您的答案，例如工作代码或文档引用。