Rxjs笔记三：Observer的简写形式以及退订Observable

Posted 2023-03-31

参考技术A Observable对象都是一个个的对象，可以包含next、complete、以及error三个方法，用于接受Observable的三种不同事件。在创建Observable对象后，添加的事件中可以不用提供complete以及error方法。
为了简洁性，subscribe也可以直接接受函数作为参数，第一个如果是函数类型，就会被认为是next、第二个函数参数被认为是error，第三个则是complete。

如果不关心对异常的处理，但是关心Observable的终止事件，那依然要传递第二个参数来占位置，也就是null，表示不处理异常事件。

Observable和Observer之间建立的关系是通过subscribe，但是有的情况下需要切断二者之间的联系，如：Observer只需要监听一个Observable对象一分钟，一分钟之后就不再关心这个Observable这个对象的任何事件了。那么就需要“退订”了。

theSubscribe函数有返回结果，返回结果是一个对象，其中包含一个unsubscribe字段，这个unsubscribe字段的值是另一个函数，调用了clearInterval来清除setInterval产生的效果。

在输出了1、2、3之后，unsubcribe函数被调用，作为Observer不再接受到被推送的数据，但是作为Observale的 source$ 就再也不会产生新的数据了，输出结束，这是一个完整的“订阅”和“退订”过程。
虽然unsubcribe函数调用之后，作为Observer不再接受到被推送的数据，但是作为Observable的 source$ 并没有终结，因为始终没有调用complete，只不过它再也不会调用next函数了。

Angular 2 / Rxjs：我真的需要退订吗？

【中文标题】Angular 2 / Rxjs：我真的需要退订吗？

【英文标题】：Angular 2 / Rxjs : do I really need to unsubscribe?

【发布时间】：2017-07-30 09:02:48

【问题描述】：

我明白，当组件被销毁时，我必须取消订阅某些 Observable（即：具有 infinite 值的 Observable），以防止内存泄漏。我不需要为 finite Observables 这样做，因为它们会自动完成并自动 unsubscribe。

但是如果我在我的组件中创建一个infinite Observable（例如FormGroup.valueChanges，或QueryList.changes），这个会被包含它的组件一起销毁，所以我 认为即使我不退订，它们也不会发生内存泄漏。

这是一个简单的例子：

@Component()
export class DummyComponent 
    form: FormGroup;

    constructor(private fb: FormBuilder) 
        this.form = this.fb.group(
            firstName: [''],
            lastName: ['']
        );
        this.form.valueChanges.subscribe(
            x => console.log(x)
        );
    

这里，我不是来自this.form.valueChanges的unsubscribe；当我的组件被销毁时，this.form.valueChanges 也会被销毁。

这种情况会不会出现内存泄漏？

【问题讨论】：

你有没有尝试给 subscribe(...) 添加一个完整的回调，并检查它是否在组件被销毁时被调用？

我做了，它没有被调用（我只用this.form.valueChanges做了测试）。

我想这样会更好，但没有必要取消订阅。我不希望出现内存泄漏，因为当组件被销毁时，它会被垃圾收集，并且你的订阅也会随之而来。如果您在组件之外订阅或传递订阅（例如服务），那么这可能会阻止组件被 GC，但如果所有组件都在组件内，它应该不会受到伤害。

【参考方案1】：

正如 Babar 所提到的，您需要取消订阅以停止这些订阅以继续关注更改。

对于你的具体情况，我认为你有一个观点。

当我在同一个组件中有很多订阅时，我会做以下事情。

首先我创建“订阅”，一个订阅类型的空数组。

private subscriptions: Subscription[] = [];

然后每次我需要订阅时，我都会将其推送到数组中

this.subscriptions.push(this.form.valueChanges.subscribe(x => console.log(x)));

在 ngOnDestroy 中，我取消订阅数组中的每个订阅。

ngOnDestroy(): void 
  this.subscriptions.forEach((elem) =>  elem.unsubscribe(); )

【讨论】：

您可以使用 subscription.add() 方法相互添加订阅。那么你只需要退订top订阅即可。

【参考方案2】：

当你的组件被销毁时，你的订阅者不是还在等待任何事件发生，这也会消耗内存，有时它也可能会扰乱你的逻辑。例如路由器事件订阅，它将在您的应用程序中的任何地方触发并执行您在订阅中执行的代码。

第二种情况，如果您在组件之间切换并且从未取消订阅，您的应用将在一段时间后挂起，因为每当您加载组件时，新订阅者都会绑定。

@Component()
export class DummyComponent implements OnDestroy 
 form: FormGroup;
 subscription: Subscription; // from rxjs
 constructor(private fb: FormBuilder) 
    this.form = this.fb.group(
        firstName: [''],
        lastName: ['']
    );
    this.subscription = this.form.valueChanges.subscribe(
        x => console.log(x)
    );
 

 ngOnDestroy(): void 
  this.subscription.unsubscribe();