使用 Rxjs 的 `takeUntil` 自动取消订阅 - 无需等待发射？答案

【问题标题】：Using Rxjs's `takeUntil` to auto unsubscribe - without waiting for emission?使用 Rxjs 的 `takeUntil` 自动取消订阅 - 无需等待发射？
【发布时间】：2018-08-20 13:00:43
【问题描述】：

我已经阅读了 Ben 的 article 关于取消订阅 via 、takeUntil 、takeWhile（谓词版本）

I've used it like this example in my Angular app

但是有些东西我不明白。

假设我有一个 Observable，它会在很长一段时间后发出值：

const source = Rx.Observable.interval(10000);

var isContinue=true;

const example = source.takeWhile(val => isContinue)
                      .subscribe(val => {},()=>{},()=>console.log('complete'));

setTimeout(function (){isContinue=false},3000) //somewhere in destructor

嗯，这只会在 10 秒后显示“完成”，而不是在 3 秒后。

所以基本上如果我有一个订阅 Observable 的组件并且该 observable 长时间不发出值，它仍然会引用我的对象，从而导致缓慢的内存泄漏。

问题：

如何在我设置好后立即使用 takeWhile 运算符取消订阅 isContinue 值。

我不想信任可能不会发出值的服务并保留对我的组件的引用。

JSBIN

【问题讨论】：

您可以通过example 订阅自己致电unsubscribe。如果您想确保在完成或取消订阅时发生某些操作，您可以使用 finally() 运算符或使用 example.add(() => { console.log('complete') }) 手动添加处置处理程序
@martin 谢谢。所以基本上这些运营商不能依赖。 在 takeWhile 值改变之后，它必须发出至少一个值....为了注意到变化
是的，这就是takeWhile 的用途。但老实说，我不明白takeUntil 有什么问题。我认为这是最简单的方法。
takeWhile 应该使用你的布尔值，试试这个：source.takeWhile(this.isContinue)
Observable.subscribe 返回 Subscription github.com/ReactiveX/rxjs/blob/… 但实际上它返回订阅者 github.com/ReactiveX/rxjs/blob/master/src/internal/util/… 扩展 Subscription 并具有 complete 方法 github.com/ReactiveX/rxjs/blob/master/src/internal/…

标签： javascript angular rxjs

【解决方案1】：

我相信您的代码会产生正确的排放 - 但您关心的是完成的及时性（它应该在 3 秒后完成，而不是 10 秒）。

这需要 10 秒，因为 takeWhile 仅在源 observable 发出时测试其谓词。

takeUntil 会解决这个问题，但它需要一个 observable 来等待：

const source = Rx.Observable.interval(10000);

const abort = new Subject();

const example = source.takeUntil(abort)
                      .subscribe(val=>{}, ()=>{}, ()=>console.log('complete'));

setTimeout(function () {abort.next()}, 3000) //somewhere in destructor

【讨论】：

我明白了。这种差异应该在文档中提到。因为（像我自己一样），设置 Subject 只会发出一个值 - 看起来很奇怪。这就是为什么我认为听boolean (takeWhile) 是更好的选择，但正如你所说，它们的行为不同。 takeUntil 一直在关注 subject ，而不管发射，而 takeWhile 只会在 next(!) 发射（可能永远不会出现）时考虑布尔值。我对诊断的看法是对的吗？
takeUntil 的主要优点是它本身就是订阅的。从这个意义上说，它并不总是在监视——它会收到通知，因此它会立即采取行动，而无需花费任何精力进行监视。我推荐这种方法。

【解决方案2】：

您需要将isContinue 变量设置为Observable 并订阅您的两个可观察对象：source 和isContinue。

【讨论】：

takeWhile 也可以返回一个布尔值。那么为什么它没有注意到 boolean 的变化，而只是在发射之后呢？而 takeUntil ，是否看到发生的变化？
它可以返回一个布尔值，但它始终应该应用于Observable。
你能举个例子吗？