2. RxJS - 观察者和订阅者

爱被打了一巴掌 2024-03-16 12:23 72阅读 0赞

欢迎大家回来，  
今天文章的主题是Observers和Subscriptions。

**观察员**  
观察者是价值的消费者。观察者是对可观察者发出的值感兴趣的东西。在 RxJS 中，Observer 只是一组回调（next、error、complete）。  
这是一个观察者的简单例子

import {
         Observer } from "rxjs";
    
    const observer: Observer<string> = {
        
      next: (value: string) =>
        console.log(`[observer] next`, value),
      error: (error: Error) =>
        console.error(`[observer] error`, error),
      complete: () =>
        console.log(`[observer] complete!`),
    };

如您所见，一个观察者有三个回调，一个对应于 Observable 可能发出的每种类型的通知。每个回调都可以在可观察事件通知的基础上做出反应。  
观察者可能没有定义所有三个回调，因为它可能不想对所有通知做出反应。在这些情况下，您可以创建一个只包含您需要的回调的观察者。

import {
      NextObserver,
      ErrorObserver,
      CompletionObserver,
    } from "rxjs";
    
    const nextObserver: NextObserver<string> = {
      next: (value: string) =>
        console.log(`[nextObserver] next`, value),
    };
    
    const errorObserver: ErrorObserver<string> = {
      error: (error: Error) =>
        console.error(`[errorObserver] error`, error),
    };
    
    const completeObserver: CompletionObserver<string> = {
      complete: () =>
        console.log(`[completeObserver] complete!`),
    };

可以定义一个没有 next 属性但仅指示 next 方法主体的 NextObserver，在这种情况下，观察者默认为 NextObserver 类型。

const defaultNextObserver: (value: string) => void = (value: string) =>
        console.log(`${
        new Date().toISOString()} - [defaultNextObserver] next`, value)

**订阅**  
订阅是一个参与者，它决定何时必须收听 Observable 以及我们何时可以停止收听它。  
在您的代码中没有订阅可观察对象之前，什么都不会发生，但是一旦您创建订阅，魔法就会开始。  
RxJS 中的订阅是一个使用“订阅”方法创建的对象，它有一个主要方法：“取消订阅”；此方法允许您停止侦听可观察事件。  
在您的代码中，当您不再需要订阅时调用“取消订阅”很重要，这可以防止内存泄漏等问题。  
订阅示例

import {
         Subscription } from "rxjs";
    
    const observer = (value: string) => console.log(`[unsubscribe method] next`, value)
    
    const subscription: Subscription = observable.subscribe(observer);
    subscription.unsubscribe();

RxJS 订阅中的另一个漂亮的特性是“添加”方法；此方法允许您在一个订阅实例中收集更多订阅，之后，您可以一次取消订阅所有订阅。

import {
         Subscription } from "rxjs";
    
    const subscription: Subscription = observable.subscribe((value: string) =>
      console.log(
        `[unsubscribe 1 method] next`,
        value
      )
    );
    subscription.add(
      observable.subscribe((value: string) =>
        console.log(
          `[unsubscribe 2 method] next`,
          value
        )
      )
    );
    
    subscription.unsubscribe();

在某些情况下，当您调用取消订阅方法时，您需要在您的可观察对象中运行一些特殊代码：RxJS 让我们在可观察对象声明中使用特殊语法来执行此操作。当您创建可观察对象时，您可以返回库在取消订阅期间调用的函数。下面是一个简单的例子，可以更好地理解问题和解决方案：

import {
        
      Observable,
      Subscriber,
      Subscription,
    } from "rxjs";
    
    const observableWithCallback = new Observable<string>(
      (subscriber: Subscriber<string>) => {
        
        let count = 0;
        const id = setInterval(() => {
        
          subscriber.next(`Count: ${
        ++count}`);
        }, 300);
    
        return () => {
        
          console.log("On UnSubscription");
          clearInterval(id);
        };
      }
    );
    
    const subscriptionObservableWithCallback: Subscription = observableWithCallback.subscribe({
        
      next: (value: string) =>
        console.log(`[observableWithCallback] Next: ${
        value}`),
    });
    setTimeout(() => {
        
      subscriptionObservableWithCallback.unsubscribe();
    }, 3000);
    
    [observableWithCallback] Next: Count: 1
    [observableWithCallback] Next: Count: 2
    [observableWithCallback] Next: Count: 3
    [observableWithCallback] Next: Count: 4
    [observableWithCallback] Next: Count: 5
    [observableWithCallback] Next: Count: 6
    [observableWithCallback] Next: Count: 7
    [observableWithCallback] Next: Count: 8
    [observableWithCallback] Next: Count: 9
    On UnSubscription

在下一篇文章中，我们将看到 RxJS 的最后一个基础：运算符，稍后，我们可以移动到这个库的功能。  
如果你喜欢我的文章，记得关注获取更多的信息。感谢您的阅读，祝您有美好的一天！