Angular2 Observable和RxJS

作者: tuacy | 来源:发表于2018-08-07 23:06 被阅读858次

看到Observable和RxJS就感觉很亲切，因为之前做Android开发的时候接触过RxJava。Observable和RxJS的相关知识最好的文档还是官方文档 https://www.angular.cn/guide/observables 强烈推荐大家看官方文档。

一、可观察对象(Observable)

可观察对象(Observable)在Angular 中使用非常广泛。可观察对象支持在应用中的发布者和订阅者之间传递消息。在需要进行事件处理、异步编程和处理多个值的时候，可观察对象相对其它技术有着显著的优点。

可观察对象的使用本质可以认为是一个观察者模式。简单的流程就是一个观察者(Observer)通过subscribe()方法订阅一个可观察对象(Observable)。订阅之后观察者(Obsever)对可观察者(Observable)发射的数据或数据序列就能作出响应(next函数发射数据)。涉及到三个东西：观察者(Observer)、可观察者(Observable)、订阅(subscribe)。

我们先给出一个简单的实例，然后再分开来讲观察者(Observer)、可观察者(Observable)、订阅(subscribe)。

import {Component} from '@angular/core';
import {of} from 'rxjs';

// 创建一个可观察者对象-Observable，发射三个数据1、2、3
const myObservable = of(1, 2, 3);

// 创建一个观察者对象-Observer(处理next、error、complete回调)
const myObserver = {
    next: x => console.log('Observer got a next value: ' + x),
    error: err => console.error('Observer got an error: ' + err),
    complete: () => console.log('Observer got a complete notification'),
};

// 通过Observable的subscribe函数，观察者去订阅可观察者的消息
myObservable.subscribe(myObserver);

@Component({
    selector: 'app-root',
    templateUrl: './app.component.html',
    styleUrls: ['./app.component.css']
})
export class AppComponent {
}

1.1、Observer(观察者)

Observer(观察者)用于接收Observable(可观察者)对象通知的处理器(说白了就是就是接收Observable发送过来的消息)。Observer(观察者)需要实现Observer接口。
观察者对象定义了一些回调函数来处理可观察对象可能会发来的三种通知(Observer接口里面的方法)。

通知类型(方法)	说明
next	必要。用来处理每个发送过来的值。在开始执行后可能执行零次或多次
error	可选。用来处理错误通知。错误会中断这个可观察对象实例的执行过程
complete	可选。用来处理执行完毕(complete)的通知。当执行完毕后，这些值就会继续传给下一个处理器

当然了观察者对象可以定义这三种处理器(next、error、complete)的任意组合。如果你不为某种通知类型提供处理器，这个观察者就会忽略相应类型的通知。

举个例子比如我们只想处理next()方法对应的通知那么观察值就可以这么写了：

// 创建一个观察者对象-Observer(只处理next回调)
const myObserver = {
    next: x => console.log('Observer got a next value: ' + x),
};

1.2、Observable(可观察者)

使用Observable构造函数可以创建任何类型的可观察流。当执行可观察对象的subscribe()方法时，这个构造函数就会把它接收到的参数作为订阅函数来运行。订阅函数需要接收一个Observer对象，并把值发布给观察者对象的next()方法。其实很好理解，比如有如下的代码，sequenceSubscriber方法是Observable构造函数的参数。当调用subscribe()方法订阅的时候就会执行sequenceSubscriber方法里面的动作发射数据。

import {Component} from '@angular/core';
import {Observable, of} from 'rxjs';

// 可观察者构造函数的参数
function sequenceSubscriber(observer) {
    // 发射三个值
    observer.next(1);
    observer.next(2);
    observer.next(3);
    observer.complete();

    return {
        unsubscribe() {
        }
    };
}

// 通过构造函数来创建一个可观察者
const sequence = new Observable(sequenceSubscriber);

// 订阅
sequence.subscribe({
    next(num) {
        console.log(num);
    },
    complete() {
        console.log('Finished sequence');
    }
});

@Component({
    selector: 'app-root',
    templateUrl: './app.component.html',
    styleUrls: ['./app.component.css']
})
export class AppComponent {
}

1.3、 Subscribing(订阅)

光有观察者和可观察者是不够的，还需要通过订阅把他两串联起来才能运作起来。只有当有人订阅Observable的实例时，它才会开始发布值。订阅就是去调用Observable对象的subscribe()方法，并把一个Observer对象传给它，用来接收通知。subscribe()方法的调用会返回一个Subscription对象，该对象具有一个unsubscribe()方法。当调用该方法时，你就会停止接收通知。

1.4、多播

默认情况下可观察对象会为每一个观察者创建一次新的、独立的执行。订阅了多少次就会有多少个独立的流(next监听器会重复调用)。

多播:多播用来让可观察对象在一次执行中同时广播给多个订阅者。借助支持多播的可观察对象，你不必注册多个监听器，而是复用第一个(next)监听器，并且把值发送给各个订阅者。我们通过一个简单的实例来看多播的代码应该怎么写，会把所有的观察者放在一个数组里面，然后复用第一个观察者的监听器。

多播代码

import {Component} from '@angular/core';
import {Observable} from 'rxjs';

function multicastSequenceSubscriber() {
    // 需要发射的数据
    const seq = [1, 2, 3];
    // 观察者数组，多播那肯定会有多个观察者
    const observers = [];
    let timeoutId;

    // 在调用Observable对应subscriber()方法的时候，会传入进来observer观察者对象
    return (observer) => {
        // observers观察者对象加入数组
        observers.push(observer);
        // 第一次有观察者订阅过来的时候
        if (observers.length === 1) {
            timeoutId = doSequence({
                next(val) {
                    // 遍历每个观察者，调用观察者的next()方法
                    observers.forEach(obs => obs.next(val));
                },
                complete() {
                    // 遍历每个观察者，调用观察者的complete()方法，调用slice(0)又冲第一个元素开始遍历。
                    // 因为前面已经调用过observers.forEach了已经移动到最后一个元素去了
                    observers.slice(0).forEach(obs => obs.complete());
                }
            }, seq, 0);
        }

        return {
            unsubscribe() {
                // 如果调用了取消订阅，则从数组里面删除
                observers.splice(observers.indexOf(observer), 1);
                // 如果是最后一个，则清除 timer out
                if (observers.length === 0) {
                    clearTimeout(timeoutId);
                }
            }
        };
    };
}

// 每秒发射一个数据
function doSequence(observer, arr, idx) {
    return setTimeout(() => {
        observer.next(arr[idx]);
        if (idx === arr.length - 1) {
            observer.complete();
        } else {
            // 继续执行
            doSequence(observer, arr, ++idx);
        }
    }, 1000);
}

// 创建一个多播的被观察者
const multicastSequence = new Observable(multicastSequenceSubscriber());

// 第一个观察者订阅
multicastSequence.subscribe({
    next(num) {
        console.log('1st subscribe: ' + num);
    },
    complete() {
        console.log('1st sequence finished.');
    }
});

// 1.5s之后，第二个观察者订阅
setTimeout(() => {
    multicastSequence.subscribe({
        next(num) {
            console.log('2nd subscribe: ' + num);
        },
        complete() {
            console.log('2nd sequence finished.');
        }
    });
}, 1500);

@Component({
    selector: 'app-root',
    templateUrl: './app.component.html',
    styleUrls: ['./app.component.css']
})
export class AppComponent {
}

二、RxJS

2.1、RxJS简单介绍

RxJS(响应式扩展的JavaScript 版)是一个使用可观察对象进行响应式编程的库。它让组合异步代码和基于回调的代码变得更简单。RxJS文档链接 https://rxjs-dev.firebaseapp.com/ 。

RxJS是一个库，一个工具，让我们写异步的代码非常的简单。

RxJS的学习关键在操作符的学习，RxJS提供了各种各样的操作符。操作符用的对很多事情能事半功倍。操作符的类型有:创建操作符、组合操作符、过滤操作符、转换操作符、多播操作符等等。

RxJS常用操作符

类别	操作
创建	from、fromPromise、fromEvent、of等
组合	combineLatest、concat、merge、startWith、withLatestFrom、zip等
过滤	debounceTime , distinctUntilChanged , filter , take , takeUntil等
转换	bufferTime , concatMap , map , mergeMap , scan , switchMap等
工具	tap等
多播	share等

如果想深入的学习RxJS可以多去了解里面的操作符。我只能说操作符非常的强大。

2.2、管道pipe

有的时候我们可能想把多个操作符连接起来就需要借助管道pipe()函数来实现。pipe() 函数以你要组合的这些函数作为参数，并且返回一个新的函数，当执行这个新函数时，就会顺序执行那些被组合进去的函数。我们用一个简单的实例来来看看pipe管道怎么使用，通过管道把filter操作符和map操作符链接起来。

要是在RxJava里面要把多个操作符链接起来，非常的简单直接...链式编程就可以实现。但是RxJS里面不支持这种操作，只能通过管道把多个操作符链接起来。

import {Component} from '@angular/core';
import {of} from 'rxjs';
import {filter, map} from 'rxjs/operators';

// 通过RxJS的of创建操作符创建一个Observable对象，并且通过管道把filter操作符和map操作符链接起来
const squareOdd = of(1, 2, 3, 4, 5)
    .pipe(
        // 只需要奇数
        filter(n => n % 2 !== 0),
        // 值平方
        map(n => n * n)
    );

// 订阅
squareOdd.subscribe(x => console.log(x));

@Component({
    selector: 'app-root',
    templateUrl: './app.component.html',
    styleUrls: ['./app.component.css']
})
export class AppComponent {
}

2.3、RxJS错误处理

RxJS除了可以在订阅时提供error()处理器外，RxJS 还提供了catchError操作符来处理一些不是致命的错误。什么意思，我们知道一旦走到error()方法去了之后整个数据流就直接断了，比如我们顺序发送100个数据，第一个数据发送的时候就发生了错误后面的99个数据都没办法再发送了。RxJS里面的catchError操作符可以避免这种情况，他让你有一个修复的机会，我们可以在catchError里面做一些特殊的处理，当第一个数据发送的时候的错误，我们可以通过某种方式让数据可以继续发送。比如如下的实例，第一个数据发射的时候产生了错误，我们接着从2开始发送数据。

RxJS里面的catchError 操作符让我们对一些错误可以做一些修复。

import {Component} from '@angular/core';
import {range} from 'rxjs';
import {catchError, map} from 'rxjs/operators';

const apiData = range(1, 100).pipe(
    map(value => {
        if (value === 1) {
            // 第一个数据就发生错误了
            throw new Error('Value expected!');
        }
        return value;
    }),
    // 当有错误返回的时候，我们又从2开始发送
    catchError(err => range(2, 99))
);

apiData.subscribe({
    next(x) {
        console.log('data: ', x);
    },
    error(err) {
        console.log('errors already caught... will not run');
    }
});

@Component({
    selector: 'app-root',
    templateUrl: './app.component.html',
    styleUrls: ['./app.component.css']
})
export class AppComponent {
}

三、Angular里面的可观察对象

Angular通常使用可观察对象作为处理各种常用异步操作的接口。

3.1、事件发送器 EventEmitter

EventEmitter类我们在前一篇文章(组件交互)有提过，当子组件想向父组件发送消息的时候我们就用到了EventEmitter。EventEmitter用来从组件的 @Output() 属性中发布一些值。EventEmitter扩展了Observable，并添加了一个 emit()方法，这样它就可以发送任意值了。当你调用emit() 时，就会把所发送的值传给订阅上来的观察者的next()方法。

子组件代码

import {Component, EventEmitter, Output} from '@angular/core';

@Component({
  selector: 'app-data-child',
  template: `
    <button (click)="vote(true)">点击</button>
  `
})
export class DataChildComponent {

  // @Output定义一个准备回调父组件的事件EventEmitter也是可以传递参数的
  @Output() voted = new EventEmitter<boolean>();


  vote(agreed: boolean) {
    // 把事件往上抛出去，可以带参数
    this.voted.emit(agreed);
  }
}

父组件代码

import {Component} from '@angular/core';

@Component({
  selector: 'app-data-parent',
  styleUrls: ['./data-parent.component.css'],
  template: `
    <p>点击 {{clickCount}} 次</p>
    <app-data-child (voted)="onVoted($event)"></app-data-child>
  `
})
export class DataParentComponent {

  clickCount = 0;

  /**
   * 子组件抛上来的事件
   */
  onVoted(agreed: boolean) {
    this.clickCount++;
  }
}

上面代码中我们在子组件里面通过@Output() voted = new EventEmitter<boolean>();定义了一个EventEmitter类型的输出变量voted。并且父组件里面绑定到了这个输出变量(其实就是订阅的关系)。我们可以理解下他们是怎么工作的。当子组件里面的按钮被点击之后，调用了EventEmitter的emit()函数。这个时候EventEmitter会去检测有那些观察值订阅了这个EventEmitter。然后调用这些订阅者的next()函数从而调用父组件模板里面写的函数。

3.2、HTTP

Angular的HttpClient从HTTP 方法调用中返回了可观察对象。例如，http.get('url')、http.post('url')返回的对象就是可观察对象。我们还是用一个非常简单的实例来说明。组件里面有一个按钮，点击这个按钮的时候通过htt.get(url)获取url上对应的数据。

service对应代码

import {Injectable} from '@angular/core';
import {HttpClient} from '@angular/common/http';

@Injectable()
export class NetworkProtocolsService {

  constructor(private http: HttpClient) {
  }

  /**
   * 获取配置信息-返回Observable
   */
  getConfig() {
    return this.http.get('assets/config.json');
  }
}

组件对应代码

import {Component} from '@angular/core';
import {NetworkProtocolsService} from '../network-protocols.service';

@Component({
  selector: 'app-data-parent',
  styleUrls: ['./data-parent.component.css'],
  template: `
    <button (click)="onButtonClick()">点击获取配置信息</button>
    <p>配置信息:{{message}}</p>
  `,
  providers: [NetworkProtocolsService]
})
export class DataParentComponent {

  message = '';

  constructor(private protocolsService: NetworkProtocolsService) {
  }


  // 点击按钮获取配置信息,简单的把配置信息显示出来
  onButtonClick() {
    this.protocolsService.getConfig()
      .subscribe(
        data => this.message = data.toString(),
        err => console.error('network get error: ' + err));

  }
}

3.3、AsyncPipe(异步管道)

AsyncPipe(异步管道)订阅一个 Observable或Promise对象(这里我们只讲Observable)，并返回它发出的最新值。当通过next()方法发出新值时，订阅的异步管道就知道数据发送变化了，然后标识组件需执行变化检测。当组件被销毁时，异步管道自动取消订阅，以避免潜在的内存泄漏。

如下实例，time是一个Observable，并且并且每一秒都发射一个新的值。

import {Component} from '@angular/core';
import {Observable} from 'rxjs';

@Component({
  selector: 'app-data-parent',
  styleUrls: ['./data-parent.component.css'],
  template: `
    <div><code>observable|async</code>:
      Time: {{ time | async }}
    </div>`
})
export class DataParentComponent {

  time = new Observable(observer => {
      setInterval(() => observer.next(new Date().toString()), 1000);
    }
  );
}

3.4、路由器(router)

Router.events路由器的事件都是以可观察对象的形式提供的。而且路由器的事件有以下七种。如下所示:

路由器事件	说明
NavigationStart	会在导航开始时触发
RoutesRecognized	会在路由器解析完 URL，并识别出了相应的路由时触发
RouteConfigLoadStart	会在 Router 对一个路由配置进行惰性加载之前触发
RouteConfigLoadEnd	会在路由被惰性加载之后触发
NavigationEnd	会在导航成功结束之后触发
NavigationCancel	会在导航被取消之后触发。这可能是因为在导航期间某个路由守卫返回了 false
NavigationError	会在导航由于意料之外的错误而失败时触发

因为路由器事件是以Observable的形式提供的，我们也可以使用RxJS里面的filter()操作符来找到感兴趣的事件，并且订阅它们。如下代码我们只关心NavigationStart事件。

import {Component, OnInit} from '@angular/core';
import {Observable} from 'rxjs';
import {NavigationStart, Router} from '@angular/router';
import {filter} from 'rxjs/operators';

@Component({
  selector: 'app-data-parent',
  styleUrls: ['./data-parent.component.css'],
  template: ``
})
export class DataParentComponent implements OnInit {

  navStart: Observable<NavigationStart>;

  constructor(private router: Router) {
    // 我们只是关心NavigationStart事件
    this.navStart = router.events.pipe(
      filter(evt => evt instanceof NavigationStart)
    ) as Observable<NavigationStart>;
  }

  ngOnInit(): void {
    this.navStart.subscribe(evt => console.log('Navigation Started!'));
  }
}

3.5、响应式表单 (reactive forms)

响应式表单具有一些属性，它们使用可观察对象来监听表单控件的值。FormControl的valueChanges属性和statusChanges属性包含了会发出变更事件的可观察对象。订阅可观察的表单控件属性是在组件类中触发应用逻辑的途径之一。

用一个简单的实例来说明，当input里面的值变化的时候，我们能获取input的值的变化。

import {Component, OnInit} from '@angular/core';
import {FormControl, FormGroup} from '@angular/forms';

@Component({
  selector: 'app-data-parent',
  styleUrls: ['./data-parent.component.css'],
  template: `
    <h2>Hero Detail</h2>
    <h3><i>FormControl in a FormGroup</i></h3>
    <form [formGroup]="heroForm" novalidate>
      <div class="form-group">
        <label class="center-block">Name:
          <input class="form-control" formControlName="name">
        </label>
      </div>
    </form>
  `
})
export class DataParentComponent implements OnInit {

  heroForm = new FormGroup({
    name: new FormControl()
  });

  ngOnInit() {
    const nameControl = this.heroForm.get('name');
    // 当表单里面数据变化的时候，可以收到通知。nameControl.valueChanges就是一个Observable
    nameControl.valueChanges.forEach(
      (value: string) => console.log(value)
    );
  }


}

以上就是对Angular2以上Observable和RxJS的简答介绍。最后还是想告诉大家最好的文档还是官方文档 https://www.angular.cn/guide/observables

Angular2 Observable和RxJS

一、可观察对象(Observable)

1.1、Observer(观察者)

1.2、Observable(可观察者)

1.3、 Subscribing(订阅)

1.4、多播

二、RxJS

2.1、RxJS简单介绍

2.2、管道pipe

2.3、RxJS错误处理

三、Angular里面的可观察对象

3.1、事件发送器 EventEmitter

3.2、HTTP

3.3、AsyncPipe(异步管道)

3.4、路由器(router)

3.5、响应式表单 (reactive forms)

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读

我爱编程