Observable（可观察对象）

基本用法

创建一个Observable实例，其中定义了订阅函数(subscribe)，供消费者调用。

定义观察者

用于接收可观察对象通知的处理器要实现 Observer 接口。
这个对象定义了一些回调函数来处理可观察对象可能会发来的三种通知：next()、error()、complete()。

订阅

只有订阅Observable的实例时，它才会发布值。订阅时首先要调用该实例的subscribe()方法，并把一个观察者对象传给它，用来接收通知。（消费者订阅观察者对象）

// Create simple observable that emits three valuesconst myObservable = of(1, 2, 3);// Create observer object
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'),
};// Execute with the observer object
myObservable.subscribe(myObserver);
// Logs:
// Observer got a next value: 1
// Observer got a next value: 2
// Observer got a next value: 3
// Observer got a complete notification

下面的subscribe()调用跟前面的是等价的：

myObservable.subscribe(x => console.log('Observer got a next value: ' + x),err => console.error('Observer got an error: ' + err),() => console.log('Observer got a complete notification')
);

注意：
1.next()都是必要的，而error()和complete()是可选的。
2.使用of()要这样导入：import { of } from ‘rxjs’;

错误处理

由于可观察对象会异步生成值，所以用 try/catch 是无法捕获错误的。你应该在观察者中指定一个 error 回调来处理错误。发生错误时还会导致可观察对象清理现有的订阅，并且停止生成值。可观察对象可以生成值（调用 next 回调），也可以调用 complete 或 error 回调来主动结束。

myObservable.subscribe({next(num) { console.log('Next num: ' + num)},error(err) { console.log('Received an errror: ' + err)}
});

RxJS

RxJS（响应式扩展的 JavaScript 版）是一个使用可观察对象进行响应式编程的库，它让组合异步代码和基于回调的代码变得更简单 (RxJS Docs)。
RxJS 提供了一些用来创建可观察对象的函数：
定时器：

import { interval } from 'rxjs';// Create an Observable that will publish a value on an interval
const secondsCounter = interval(1000);
// Subscribe to begin publishing values
secondsCounter.subscribe(n =>console.log(`It's been ${n} seconds since subscribing!`));

操作符

操作符是基于可观察对象构建的一些对集合进行复杂操作的函数。RxJS 定义了一些操作符，比如 map()、filter()、concat() 和 flatMap()。

import { map } from 'rxjs/operators';
import { of } from 'rxjs';
const nums = of(1, 2, 3);const squareValues = map((val: number) => val * val);
const squaredNums = squareValues(nums);squaredNums.subscribe(x => console.log(x));// Logs
// 1
// 4
// 9

可以使用管道把操作符连接起来，管道把由多个由操作符返回的函数组合成一个。即pipe()把这些函数作为参数，然后返回一个新的函数。pipe()是RxJS的Observable上的一个方法。

import { filter, map } from 'rxjs/operators';const squareOdd = of(1, 2, 3, 4, 5).pipe(filter(n => n % 2 !== 0),map(n => n * n));// Subscribe to get values
squareOdd.subscribe(x => console.log(x));

错误处理

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

import { ajax } from 'rxjs/ajax';
import { map, catchError } from 'rxjs/operators';
// Return "response" from the API. If an error happens,
// return an empty array.
const apiData = ajax('/api/data').pipe(map(res => {if (!res.response) {throw new Error('Value expected!');}return res.response;}),catchError(err => of([]))
);apiData.subscribe({next(x) { console.log('data: ', x); },error(err) { console.log('errors already caught... will not run'); }
});

重试失败的可观察对象

catchError 提供了一种简单的方式进行恢复，而 retry 操作符让你可以尝试失败的请求。
可以在 catchError 之前使用 retry 操作符。它会订阅到原始的来源可观察对象，它可以重新运行导致结果出错的动作序列。如果其中包含 HTTP 请求，它就会重新发起那个 HTTP 请求。

import { ajax } from 'rxjs/ajax';
import { map, retry, catchError } from 'rxjs/operatorsconst apiData = ajax('/api/data').pipe(retry(3), // Retry up to 3 times before failingmap(res => {if (!res.response) {throw new Error('Value expected!');}return res.response;}),catchError(err => of([]))
);apiData.subscribe({next(x) { console.log('data: ', x); },error(err) { console.log('errors already caught... will not run'); }
});

Angular中的可观察对象

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

事件发送器 EventEmitter

EventEmitter扩展了Observable，并添加了emit()方法，这样它就能发送任意了。当你调用 emit() 时，就会把所发送的值传给订阅上来的观察者的 next() 方法。

HTTP

Angular 的 HttpClient 从 HTTP 方法调用中返回了可观察对象。与Promise相比有以下优点：
1）可观察对象不会修改服务器的响应（和在承诺上串联起来的 .then() 调用一样）。反之，你可以使用一系列操作符来按需转换这些值。
2）HTTP 请求是可以通过 unsubscribe() 方法来取消的。
3）请求可以进行配置，以获取进度事件的变化。
4）失败的请求很容易重试。

Async管道

AsyncPipe 会订阅一个可观察对象或承诺，并返回其发出的最后一个值。当发出新值时，该管道就会把这个组件标记为需要进行变更检查的（因此可能导致刷新界面）。

@Component({selector: 'async-observable-pipe',template: `<div><code>observable|async</code>:Time: {{ time | async }}</div>`
})
export class AsyncObservablePipeComponent {time = new Observable(observer =>setInterval(() => observer.next(new Date().toString()), 1000));
}

路由器

Router.events 以可观察对象的形式提供了其事件。 你可以使用 RxJS 中的 filter() 操作符来找到感兴趣的事件，并且订阅它们，以便根据浏览过程中产生的事件序列作出决定。

响应式表单

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