Angular—RxJS之Observable

目录

一、概述

二、详解


一、概述

Observable翻译为可观察对象,是RxJS中的一个对象,可以用来处理异步事件,例如http请求。实际上,在Angular中,所有的http请求返回的都是Observable。Observable和promise本质上是相同的,都是生产者主动向消费者push产品,而消费者被动接收。但两者也有很大区别,Observable可以发送任意多值,并且在被订阅钱,它是不会执行的,这是promise不具备的特点。

Observable用于在发送方和接收方之间传输消息,可以将这些消息看作流。

在创建Observable对象时,需要传入一个函数作为构造函数的参数,这个函数叫订阅者函数,是生产者向消费者推送消息的地方。在被消费者subscribe(订阅)之前,订阅者函数不会执行,直到subscribe()函数被调用,该函数返回一个subscription对象,里面有一个unsubscribe()函数,消费者可以随时拒绝消息的接收。

subscribe()函数接收一个observer(观察者)对象作为入参。

消息的发送可以是同步的,也可以是异步的。

我们可以使用一系列的RxJS操作符,在这些消息被接收方接收之前,对它进行一些列的处理、转换,因为这些操作符都是纯函数,没有副作用,可以放心使用,并不会产生期望之外的效果。

二、详解

observer观察者

observer为观察者,可以观察Observable对象。observer是一个对象,里面包含三个属性,分别是next、error和complete,它们都是函数。

  1.  
    (1) next: 以接收的值作为入参, 在正常情况下执行, 可选
  2.  
    (2) error: 出错的情况下执行, 可选
  3.  
    (3) complete: 传输完成的情况下执行, 可选

Observable被观察者

实例

通过RxJS库中的of方法创建一个Observable对象,代码如下所示。

  1.  
    onClick() {
  2.  
    let observable = of(1, 2, 3);
  3.  
    observable.subscribe({ next: num => console.log(num) });
  4.  
    // 控制台分别输出:1 2 3
  5.  
    }

解析

RxJS中的of方法用于创建一个Observable,它会将其参数一个一个的分别发送给接收方。

subscribe函数接收一个Observable对象,但这里,只定义了next方法,可以发现next方法接收一个参数,这个参数就是生产者发送过来的值,然后将其打印在控制台上。

订阅者函数

实例

  1.  
    onClick2() {
  2.  
    const observable = Observable.create(observer => {
  3.  
    try {
  4.  
    observer.next(1);
  5.  
    observer.next(2);
  6.  
    observer.next(3);
  7.  
    } catch (e) {
  8.  
    observer.error(e);
  9.  
    }
  10.  
    observer.complete();
  11.  
    });
  12.  
    const observer = {
  13.  
    next: num => console.log(num),
  14.  
    error: e => console.log(e),
  15.  
    complete: () => console.log(‘complete!!!‘)
  16.  
    }
  17.  
    observable.subscribe(observer); // 控制台分别输出:1 2 3 complete!!!
  18.  
    }

Angular—RxJS之Observable

解析

以上代码效果与用of创建的Observable是一样的,of函数的源码就是类似这种结构。

subscribe订阅

在被消费者subscribe(订阅)之前,订阅者函数不会执行,直到subscribe()函数被调用,该函数返回一个subscription对象。里面有一个unsubscribe()函数,消费者可以随时拒绝消息的接收。也就是说,在Observable调用subscribe函数之前,什么也不会发生,就像下面这段代码,控制台什么输出内容都没有。

  1.  
    onClick() {
  2.  
    let observable = of(1, 2, 3);
  3.  
    }

订阅observable对象,如下代码所示。

  1.  
    onClick() {
  2.  
    let observable = of(1, 2, 3);
  3.  
    observable.subscribe({ next: num => console.log(num) });
  4.  
    // 控制台分别输出:1 2 3
  5.  
    }

异步发送消息

以上代码中生产者发送消息的方式都是同步。如下代码所示,异步发送消息。

  1.  
    onClick3() {
  2.  
    const observable = Observable.create(observer => {
  3.  
    try {
  4.  
    let time = 0;
  5.  
    observer.next(1);
  6.  
    observer.next(2);
  7.  
    observer.next(3);
  8.  
    const intervalId = setInterval(() => {
  9.  
    console.log(`wait ${++time}s`);
  10.  
    }, 900)
  11.  
    setTimeout(() => {
  12.  
    observer.next(4);
  13.  
    clearInterval(intervalId)
  14.  
    }, 2000);
  15.  
    } catch (e) {
  16.  
    observer.error(e);
  17.  
    }
  18.  
    // observer.complete(); // 注意不能立即调用complete函数,不然会终止消息传输
  19.  
    setTimeout(() => observer.complete(), 3000)
  20.  
    });
  21.  
    const observer = {
  22.  
    next: num => console.log(num),
  23.  
    error: e => console.log(e),
  24.  
    complete: () => console.log(‘complete!!!‘)
  25.  
    }
  26.  
    observable.subscribe(observer);
  27.  
    }

Angular—RxJS之Observable

unsubscribe函数

调用subscribe函数,返回一个subscription对象,里面有一个unsubscribe()函数,用于取消订阅。

实例

  1.  
    onClick4() {
  2.  
    const obs = Observable.create(observer => {
  3.  
    observer.next(1);
  4.  
    setTimeout(() => observer.next(2), 2000); // 等待两秒才发送下一个值
  5.  
    });
  6.  
     
  7.  
    let suber = obs.subscribe({
  8.  
    next: x => console.log("接收到:", x)
  9.  
    });
  10.  
     
  11.  
    setTimeout(() => suber.unsubscribe(), 1000); // 在一秒后取消订阅
  12.  
    }

解析

运行代码,最终只打印了数字1,因为数字2在两秒后才会发送,而消费者在1秒后就中断了消息传输。

自定义subscribe函数

Observable的subscribe函数返回一个subscription对象,该对象有一个unsubscribe函数,用于取消订阅。

实例

如下代码所示,使用函数模拟订阅。模拟了Observable的subscribe函数的内部实现(仅仅是模拟而已)。

  1.  
    onClick5() {
  2.  
    function subscribe(observer) {
  3.  
    var intervalID = setInterval(() => {
  4.  
    observer.next(‘launch.....‘);
  5.  
    }, 1000);
  6.  
     
  7.  
    return {
  8.  
    unsubscribe: () => clearInterval(intervalID)
  9.  
    }
  10.  
    }
  11.  
    var subscription = subscribe({ next: (x) => console.log(x) });
  12.  
    setTimeout(() => subscription.unsubscribe(), 5000);
  13.  
    }

RxJS的map操作符

map操作符可以接收可观察者对象发送的值,将其转换成另外的形式,并以一个新的可观察对象发送这些新值。

实例

如下代码所示,将原来可观察对象发送的值全部转换为hello world,并订阅map操作符返回的新的可观察对象。

  1.  
    import { map } from ‘rxjs/operators‘;
  2.  
     
  3.  
    onClick3() {
  4.  
    const observable = of(1, 2, 3);
  5.  
    const opt = map(num => ‘hello‘);
  6.  
    const newObservable = opt(observable);
  7.  
    newObservable.subscribe(data => console.log(data));
  8.  
    // 控制台分别输出:hello hello hello
  9.  
    }

解析

可以发现map操作符将123转换成了三个hello,并输出到控制台上,这种方式代码结构并不是很清晰,并不推荐这种写法。

pipe管道、tap操作符

pipe管道可以将多个操作符连接起来,并将操作符返回的结果组合成一个,这样代码结构更清晰。

tap操作符可以理解为窥探,就是不打扰可观察对象发送的值,但又可以取得这些值进行处理。

  1.  
    onClick3() {
  2.  
    const observable = of(1, 2, 3);
  3.  
    const newObservable = observable.pipe(
  4.  
    tap(num => console.log(num)),
  5.  
    map(num => ‘hello world‘)
  6.  
    );
  7.  
    newObservable.subscribe(data => console.log(data));
  8.  
    }

Angular—RxJS之Observable

Angular—RxJS之Observable

上一篇:HTML基础属性


下一篇:Kubernetes 命令