浅析Angular中的可观察对象、观察者和RxJS操作符

Observable（可观察对象）

Observable(可观察对象)，是RxJS库里面的一个对象，可以用来处理异步事件，例如HTTP请求（实际上，在Angular中，所有的HTTP请求返回的都是Observable）。【相关教程推荐：《angular教程》】

或许，你以前接触过一个叫promise的东西，它们本质上面是相同的：都是生产者主动向消费者“push”产品，而消费者是被动接收的，但是他们两者还是有很大区别的：Observable可以发送任意多值，并且，在被订阅之前，它是不会执行的！这是PRomise不具备的特点。

• Observable用于在发送方和接收方之间传输消息，你可以将这些消息看成是流
• 在创建Observable对象时，需要传入一个函数作为构造函数的参数，这个函数叫订阅者函数，这个函数也就是生产者向消费者推送消息的地方
• 在被消费者subscribe（订阅）之前，订阅者函数不会被执行，直到subscribe()函数被调用，该函数返回一个subscription对象，里面有一个unsubscribe()函数，消费者可以随时拒绝消息的接收！
• subscribe()函数接收一个observer（观察者）对象作为入参
• 消息的发送可以是同步的，也可以是异步的

observer（观察者）

有了可观察对象（发送方），就需要一个观察者（接收方）来观察可观察对象，观察者要实现observer接口，它是一个对象，其中包含三个属性，它们都是函数，如下:

订阅

只有当有人订阅 Observable 的实例时，它才会开始发布值。 订阅时要先调用可观察对象的 subscribe() 方法，并把一个观察者对象传给它，用来接收通知。如下：

为了展示订阅的原理，需要先创建新的可观察对象。它有一个构造函数可以用来创建新实例，但是为了更简明，也可以使用 Observable 上定义的一些静态方法来创建一些常用的简单可观察对象：

• of(...ITems) ：返回一个 Observable 实例，它用同步的方式把参数中提供的这些值一个一个发送出来。
• From(iterable) ： 把它的参数转换成一个 Observable 实例。 该方法通常用于把一个数组转换成一个（发送多个值的）可观察对象。

import {
 of }
     from "rxjs";
    // 1、通过 of() 方法返回一个可观察对象，并准备将1，2，3三个数据发送出去const observable = of(1, 2, 3);
	// 2、实现 observer 接口，观察者const observer = {
    		next: (num: number) =>
     console.LOG(num),	error: (err: Error) =>
     console.error('Observer got an error: ' + err),  	complete: () =>
 console.log('Observer got a complete notification'), }
    // 3、订阅。调用可观察对象的 subscribe() 方法订阅，subscribe() 方法中传入的对象就是一个观察者observable.subscribe(observer);
    

运行结果如下：

上面订阅的写法可以直接改为如下：参数不是对象

observable.subscribe(  num =>
     console.log(num),  err =>
     console.error('Observer got an error: ' + err),  () =>
     console.log('Observer got a complete notification'));
    

订阅者函数

在上面的例子中使用的是of()方法来创建可观察对象，这节使用构造函数创建可观察对象。

Observable 构造函数可以创建任何类型的可观察流。 当执行可观察对象的 subscribe() 方法时，这个构造函数就会把它接收到的参数作为订阅函数来运行。 订阅函数会接收一个 Observer 对象，并把值发布给观察者的 next()方法。

// 1、自定义订阅者函数function sequenceSubscriber(observer: Observernumber>
) {
      observer.next(1);
    	// 发送数据  observer.next(2);
    	// 发送数据  observer.next(3);
    	// 发送数据  observer.COMplete();
  return {
unsubscribe() {
}
}
    ;
}
    // 2、通过构造函数创建一个新的可观察对象，参数就是一个订阅者函数const sequence = new Observable(sequenceSubscriber);
// 3、订阅sequence.subscribe({
  next(num) {
     console.log(num);
 }
,	// 接受数据  complete() {
     console.log('Finished sequence');
 }
}
    );
    

运行结果如下：

上面一个例子演示了如何自定义订阅函数，那么既然可以自定义订阅者函数，我们就可以将异步代码封装进可观察对象的订阅者函数中，待异步代码执行完再发送数据。如下：

import {
 Observable }
 from 'rxjs'// 异步函数function fn(num) {
        return new Promise((reslove, reject) =>
 {
            setTimeout(() =>
 {
            num++            reslove(num)        }
, 1000)    }
)}
    // 创建可观察对象，并传入订阅者函数const observable = new Observable((x) =>
 {
        let num = 1    fn(num).then(    	res =>
 x.next(res)	// 异步代码执行完成，发送数据    ) }
    )// 订阅，接收数据，可以改为链式调用observable.subscribe(data =>
     console.log(data))	// 2

多播

https://angular.cn/guide/observables#multicasting

RxJS操作符

我们可以使用一系列的RxJS操作符，在这些消息被接收方接收之前，对它们进行一系列的处理、转换，因为这些操作符都是纯函数。

import {
 of }
     from 'rxjs';
import {
 map }
     from 'rxjs/operators';
    // 1、创建可观察对象，并发送数据const nums = of(1, 2, 3);
    // 2、创建函数以接受可观察对象const squarevalues = map((val: number) =>
     val * val);
    const squaredNums = squareValues(nums);
    squaredNums.subscribe(x =>
     console.log(x));
    

上面的方式我看不懂且难以接受，一般常用下面这种，使用piPE把多个操作符链接起来

import {
 map, Observable, filter }
     from 'rxjs'// 创建可观察对象，并传入订阅者函数const observable = new Observable((x) =>
 {
    x.next(1)    x.next(2)    x.next(3)    x.next(4)}
    ).pipe(    map(value =>
     value*100),		// 操作符    filter(value =>
     value == 200)	// 操作符).subscribe(data =>
     console.log(data))	// 200

错误处理

RxJS 还提供了catchError 操作符，它允许你在管道中处理已知错误。
假设你有一个可观察对象，它发起 API 请求，然后对服务器返回的响应进行映射。如果服务器返回了错误或值不存在，就会生成一个错误。如果你捕获这个错误并提供了一个默认值，流就会继续处理这些值，而不会报错。如下：

import {
 map, Observable, filter, catchError, of }
     from 'rxjs'const observable = new Observable((x) =>
 {
        x.next(1)	// 发送数据 1 和 2        x.next(2)}
    ).pipe(    map(value =>
 {
        if (value === 1) {
	// 1、当发送的数据为 1 时，将其乘以 100            return value*100        }
 else {
    	// 2、否则抛出错误            throw new Error('抛出错误');
        }
    }
    ),    // 3、此处捕获错误并处理错误，对外发送数据 0    catchError((err) =>
 {
        console.log(err)        return of(0)    }
    )).subscribe(    data =>
     console.log(data),    // 4、由于上面抛出的错误被 catchError 操作符处理（重新发送数据）了，所以这里能顺利订阅到数据而不报错    err =>
     console.log('接受不到数据：', err))

最后的运行结果如下：

更多编程相关知识，请访问：编程视频！！

以上就是浅析Angular中的可观察对象、观察者和RxJS操作符的详细内容，更多请关注其它相关文章！

声明：本文内容由网友自发贡献，本站不承担相应法律责任。对本内容有异议或投诉，请联系2913721942#qq.com核实处理，我们将尽快回复您，谢谢合作！