Angular4的Rxjs基础知识讲解

1. 下面这段话先看，留个印象就好（摘抄与中文文档）

rxjs 是一个库，它通过使用 observable 序列来编写异步和基于事件的程序。它提供了一个核心类型 observable，附属类型 (observer、 schedulers、 subjects) 和受 [array#extras] 启发的操作符 (map、filter、reduce、every, 等等)，这些数组操作符可以把异步事件作为集合来处理。

可以把 rxjs 当做是用来处理事件的 lodash 。

reactivex 结合了 观察者模式、迭代器模式 和 使用集合的函数式，以满足以一种理想方式来管理事件序列所需要的一切。

在 rxjs 中用来解决异步事件管理的的基本概念是：

observable (可观察对象): 表示一个概念，这个概念是一个可调用的未来值或事件的集合。

observer (观察者): 一个回调函数的集合，它知道如何去监听由 observable 提供的值。

subscription (订阅): 表示 observable 的执行，主要用于取消 observable 的执行。

operators (操作符): 采用函数式编程风格的纯函数 (pure function)，使用像 map、filter、concat、flatmap 等这样的操作符来处理集合。

subject (主体): 相当于 eventemitter，并且是将值或事件多路推送给多个 observer 的唯一方式。

schedulers (调度器): 用来控制并发并且是*集权的调度员，允许我们在发生计算时进行协调，例如 settimeout 或 requestanimationframe 或其他。

2. 先不去考虑实现方式，只看简单的例子

//app.component.html  
<p>  
  <h1>  
    welcome to {{ title }}!  
  </h1>  
</p>  
<button #button>first example</button>  

//app.component.ts  
import {component, elementref, oninit, viewchild} from '@angular/core';  
import { observable} from "rxjs";  
import "rxjs/rx";  
import {ibutton} from 'selenium-webdriver';  
  
@component({  
  selector: 'app-root',  
  templateurl: './app.component.html',  
  styleurls: ['./app.component.css']  
})  
export class appcomponent implements oninit {  
  title = 'app';  
  
  @viewchild('button')  
  button: elementref  
  
  ngoninit(): void {  
  }  
}  

接下来学习官方文档里的简单例子：

(1) 注册事件

常规写法：

var button = document.queryselector('button');
button.addeventlistener('click', () => console.log('clicked!'));

rxjs

observable.fromevent(this.button.nativeelement, 'click')
  .subscribe(() => console.log('clicked!'));

ps: 目前先不要管rxjs代码的具体的含义，先简单知道这样的方式可以去注册一个事件。然后我们以一种发布订阅模式来向一下这行代码的实现？

我们再来简单看一下上面的定义：

observable (可观察对象): 表示一个概念，这个概念是一个可调用的未来值或事件的集合。

observer (观察者): 一个回调函数的集合，它知道如何去监听由 observable 提供的值。

subscription (订阅): 表示 observable 的执行，主要用于取消 observable 的执行。

然后我们来类推一下：

subscription： subscribe(() => console.log('clicked!'));?

subject： (this.button.nativeelement, 'click')

observable ：observable.fromevent(返回的还是一个observable)。

observer： subscribe内部的函数就是观察者。

看到这里。能够想到的是rxjs内部使用观察者模式和函数式编程。先有个概念就好。

(2) 纯净性 (purity)

使得 rxjs 强大的正是它使用纯函数来产生值的能力。这意味着你的代码更不容易出错。

通常你会创建一个非纯函数，这个函数之外也使用了共享变量的代码会把将你的应用状态搞得一团糟。

常规写法：

let count = 0;
let button = document.queryselector('button');
button.addeventlistener('click', () => console.log(`clicked ${++count} times`));

rxjs:

observable.fromevent(this.button.nativeelement, 'click')
  .scan((count : number) => count + 1, 0)
  .subscribe(count => console.log(`clicked ${count} times`));

scan是一个工具方法，为了去计算点击次数。scan 操作符的工作原理与数组的 reduce 类似。它需要一个暴露给回调函数当参数的初始值。每次回调函数运行后的返回值会作为下次回调函数运行时的参数。

3. 拉取和推送

拉取和推送是两种不同的协议，用来描述数据生产者 (producer)如何与数据消费者 (consumer)如何进行通信的。

什么是拉取？ - 在拉取体系中，由消费者来决定何时从生产者那接收数据。生产者本身不知道数据是何时交付到消费者手中的。

什么是推送？ - 在推送体系中，由生产者来决定何时把数据发送给消费者。消费者本身不知道何时会接收到数据。在当今的?

javascript 世界中，promises 是最常见的推送体系类型。promise(生产者) 将一个解析过的值传递给已注册的回调函数(消费者)，但不同于函数的是，由 promise 来决定何时把值“推送”给回调函数。

rxjs 引入了 observables，一个新的 javascript 推送体系。observable 是多个值的生产者，并将值“推送”给观察者(消费者)。

function 是惰性的评估运算，调用时会同步地返回一个单一值。

generator 是惰性的评估运算，调用时会同步地返回零到(有可能的)无限多个值。返回一个iterators，每次的调用会返回一个值。

promise 是最终可能(或可能不)返回单个值的运算。通过resolve决定值的返回。

observable 是惰性的评估运算，它可以从它被调用的时刻起同步或异步地返回零到(有可能的)无限多个值。

这里的惰性的意义是说，需要调用者去调用才会触发结果的返回。上面的方式promise来说，对于回调函数then是在promise resolve值之后才会调用，否则永远不会调用，不管你的then是否存在，其他几种的话，只要调用就已经会返回值了，不管你何时调用，都会有相应的返回值，而不像promise可以决定这个值什么时候返回，是否返回。（自己的理解，如果有错误，望指出）。

4. observables 作为函数的泛化

对于官方文档其实还是有地方不是了解它的含义，这边只是整理出来一些自己认为的知识点。

(1) 订阅 observable 类似于调用函数

文档里谈到一句话‘一些人声称 observables 是异步的。那不是真的’。我的理解是这个地方是observables传递值之后对于回调函数的执行时同步的。只要值已经返回那么整个的运作是同步的。先看个例子吧

let foo = observable.create(function (observer) {
  console.log('hello');
  observer.next(42);
});

console.log('before');
foo.subscribe(function (x) {
  console.log(x);
});
console.log('after');
}

返回的结果如下

"before"
"hello"
42
"after"

这里的现象说明foo 的订阅完全是同步的，就像函数一样。也就是传递值之后。

(2) observables 传递值可以是同步的，也可以是异步的

这里其实是将传递值给订阅者的过程=分成了两块，现在是说传递值的过程。

那么 observable 和 函数的区别是什么呢？observable 可以随着时间的推移“返回”多个值，这是函数所做不到的。函数只能返回一个值。但是observables 可以传递多个值

同步输出

let foo = observable.create(function (observer) {
  console.log('hello');
  observer.next(42);
  observer.next(100); // “返回”另外一个值
  observer.next(200); // 还可以再“返回”值
});

console.log('before');
foo.subscribe(function (x) {
  console.log(x);
});
console.log('after');

//输出
"before"
"hello"
42
100
200
"after"

异步输出

let foo = observable.create(function (observer) {
  console.log('hello');
  observer.next(42);
  observer.next(100);
  observer.next(200);
  settimeout(() => {
    observer.next(300); // 异步执行
  }, 1000);
});

console.log('before');
foo.subscribe(function (x) {
  console.log(x);
});
console.log('after');

//输出
"before"
"hello"
42
100
200
"after"
300

这里的现象是说observable.subscribe() 意思是 "给我任意数量的值，无论是同步还是异步"，也是说传值的过程是可以定义的。

4. observable 剖析

observables 是使用 rx.observable.create 或创建操作符创建的，并使用观察者来订阅它，然后执行它并发送 next / error / complete 通知给观察者，而且执行可能会被清理。这四个方面全部编码在 observables 实例中，但某些方面是与其他类型相关的，像 observer (观察者) 和 subscription (订阅)。

observable 的核心关注点：

创建 observables

订阅 observables

执行 observables

清理 observables

(1) 创建 observables

rx.observable.create 是 observable 构造函数的别名，它接收一个参数：subscribe 函数。

下面的示例创建了一个 observable，它每隔一秒会向观察者发送字符串 'hi' 。

let observable = observable.create(function subscribe(observer) {
  let id = setinterval(() => {
    observer.next('hi');
  }, 1000);
});

observable.subscribe(x => console.log(x));

这个例子中，如果没有订阅函数，整个observables是不会执行的，大家可以测试一下，只有当订阅者存在才会执行，这也是上面说的惰性的评估运算，没有订阅者就不会执行，就不会有返回值。

(2) 订阅 observables

示例中的 observable 对象 observable 可以订阅，像这样：

observable.subscribe(x => console.log(x));

observable.subscribe 和 observable.create(function subscribe(observer) {...}) 中的 subscribe 有着同样的名字，这并不是一个巧合。在库中，它们是不同的，但从实际出发，你可以认为在概念上它们是等同的。

这表明 subscribe 调用在同一 observable 的多个观察者之间是不共享的。当使用一个观察者调用 observable.subscribe 时，observable.create(function subscribe(observer) {...}) 中的 subscribe 函数只服务于给定的观察者。对 observable.subscribe 的每次调用都会触发针对给定观察者的独立设置。

其实我们可以举一个例子去说明上面这句话：

let observable = observable.create(function subscribe(observer) {
  let number = 0;

  observer.next(number++);
  observer.next(number++);
  observer.next(number++);
});

let foo1 = observable.subscribe(x => console.log(x));
let foo2 = observable.subscribe(x => console.log(x));


//结果
0
1
2
0
1
2

这与像 addeventlistener / removeeventlistener 这样的事件处理方法 api 是完全不同的。使用 observable.subscribe，在 observable 中不会将给定的观察者注册为监听器。这里的意思我的理解是如果注册为监听器的话，那么他的资源是共享的，但是现在我们的现象看到并不是这样的。

*subscribe 调用是启动 “observable 执行”的一种简单方式， 并将值或事件传递给本次执行的观察者。这也是上面测试必须要有订阅者的原因所在

(3) 执行 observables

observable.create(function subscribe(observer) {...}) 中...的代码表示 “observable 执行”，它是惰性运算，只有在每个观察者订阅后才会执行。随着时间的推移，执行会以同步或异步的方式产生多个值。

observable 执行可以传递三种类型的值：

"next" 通知： 发送一个值，比如数字、字符串、对象，等等。

"error" 通知： 发送一个 javascript 错误 或 异常。

"complete" 通知： 不再发送任何值。

"next" 通知是最重要，也是最常见的类型：它们表示传递给观察者的实际数据。"error" 和 "complete" 通知可能只会在 observable 执行期间发生一次，并且只会执行其中的一个。

在 observable 执行中, 可能会发送零个到无穷多个 "next" 通知。如果发送的是 "error" 或 "complete" 通知的话，那么之后不会再发送任何通知了。

下面是 observable 执行的示例，它发送了三个 "next" 通知，然后是 "complete" 通知：

let observable = observable.create(function subscribe(observer) {
  observer.next(1);
  observer.next(2);
  observer.next(3);
  observer.complete();
});

observable 严格遵守自身的规约，所以下面的代码不会发送 "next" 通知 4：

let observable = observable.create(function subscribe(observer) {
  observer.next(1);
  observer.next(2);
  observer.next(3);
  observer.complete();
  observer.next(4); // 因为违反规约，所以不会发送
});

在 subscribe 中用 try/catch 代码块来包裹任意代码是个不错的主意，如果捕获到异常的话，会发送 "error" 通知：

observable = observable.create(function subscribe(observer) {
  try {
    observer.next(1);
    observer.next(2);
    observer.next(3);
    observer.complete();
  } catch (err) {
    observer.error(err); // 如果捕获到异常会发送一个错误
  }
});

(4) 清理 observable 执行

因为 observable 执行可能会是无限的，并且观察者通常希望能在有限的时间内中止执行，所以我们需要一个 api 来取消执行。因为每个执行都是其对应观察者专属的，一旦观察者完成接收值，它必须要一种方法来停止执行，以避免浪费计算能力或内存资源。

当调用了 observable.subscribe ，观察者会被附加到新创建的 observable 执行中。这个调用还返回一个对象，即 subscription (订阅)：

var subscription = observable.subscribe(x => console.log(x));

使用 subscription.unsubscribe() 你可以取消进行中的执行：subscription.unsubscribe();

当你订阅了 observable，你会得到一个 subscription ，它表示进行中的执行。只要调用 unsubscribe() 方法就可以取消执行。

相信看到这里，大家如果angular中使用过httpservice，在对应的生命周期ngondestroy()，是需要将我们创建的subscription清理掉，由于家里没有现成的例子，就先不写例子了。

5. observer (观察者)

什么是观察者？ - 观察者是由 observable 发送的值的消费者。观察者只是一组回调函数的集合，每个回调函数对应一种 observable 发送的通知类型：next、error 和 complete 。 和执行 observables期间传递的值相对应。

var observer = {
  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'),
};

我们将这里的方法类比到我们的httpservice的话，对于那里的观察者和这里不太一样，通常如下

var observer = {
  response => {},
  error => this.commonutilservice.showerrorinfo(error)
};

在 observable.subscribe 内部，它会创建一个观察者对象并使用第一个回调函数参数作为 next 的处理方法。所有三种类型的回调函数都可以直接作为参数来提供：

observable.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')
);

6. subscription (订阅)

什么是 subscription ？ - subscription 是表示可清理资源的对象，通常是 observable 的执行。subscription 有一个重要的方法，即 unsubscribe，它不需要任何参数，只是用来清理由 subscription 占用的资源。在上一个版本的 rxjs 中，subscription 叫做 "disposable" (可清理对象)。上面已经提到，就不在多说。

subscriptions 还有一个 remove(othersubscription) 方法，用来撤销一个已添加的子 subscription 。

由于本人也在学习过程中，这边就先不去做一些总结性的话，相信通过后面的学习会有一些心得，后面也会与angular的应用去结合学习一下，敬请关注！ <