详细介绍RxJS在Angular中的应用

rxjs是一种针对异步数据流编程工具，或者叫响应式扩展编程；可不管如何解释rxjs其目标就是异步编程，angular引入rxjs为了就是让异步可控、更简单。

而今就是要探讨什么是observable、observer、operator、submit、eventemmit，以及如何去使用它们。

什么是observable？

observable只是一个普通函数，要想让他有所作为，就需要跟observer一起使用；前者是受后者是攻。而这个observer（后面我们会介绍）只是一个带有 next、error、complete 的简单对象而已。最后，还需要通过 subscribe 订阅来启动observable；否则它是不会有任何反应；可以理解为陌*为了他们能在一起而提供的环境，而订阅也会返回一个可用于取消操作（在rxjs里叫 unsubscribe）。

当observable设置观察者后，而连接并获取原始数据的这个过程叫生产者，可能是dom中的 click 事件、input 事件、或者更加复杂的http通信。

为了更好理解，先从一个简单的示例开始：

import { component } from '@angular/core';
import { observable, subscription } from 'rxjs';

@component({
 selector: 'app-home',
 template: `<input type="text"> `
})
export class homecomponent {
 ngoninit() {
  const node = document.queryselector('input[type=text]');

  // 第二个参数 input 是事件名，对于input元素有一个 oninput 事件用于接受用户输入
  const input$ = observable.fromevent(node, 'input');
  input$.subscribe({
   next: (event: any) => console.log(`you just typed ${event.target.value}!`),
   error: (err) => console.log(`oops... ${err}`),
   complete: () => console.log(`complete!`)
  });
 }
}

示例中 observable.fromevent() 会返回一个observable，并且监听 input 事件，当事件被触发后会发送一个 event 给对应的observer观察者。

什么是observer？

observer非常简单，像上面示例中 subscribe 订阅就是接收一个 observer 方法。

一般在angular我们 subscribe 会这么写：

input$.subscribe((event: any) => {

});

从语法角度来讲和 subscribe({ next, error, complete }) 是一样的。

当observable产生一个新值时，会通知 observer 的 next()，而当捕获失败可以调用 error()。

当observable被订阅后，除非调用observer的 complete() 或 unsubscribe() 取消订阅两情况以外；会一直将值传递给 observer。

observable的生产的值允许经过一序列格式化或操作，最终得到一个有价值的数据给观察者，而这一切是由一序列链式operator来完成的，每一个operator都会产生一个新的observable。而我们也称这一序列过程为：流。

什么是operator？

正如前面说到的，observable可以链式写法，这意味着我们可以这样：

observable.fromevent(node, 'input')
 .map((event: any) => event.target.value)
 .filter(value => value.length >= 2)
 .subscribe(value => { console.log(value); });

下面是整个顺序步骤：

• 假设用户输入：a
• observable对触发 oninput 事件作出反应，将值以参数的形式传递给observer的 next()。
• map() 根据 event.target.value 的内容返回一个新的 observable，并调用 next() 传递给下一个observer。
• filter() 如果值长度 >=2 的话，则返回一个新的 observable，并调用 next() 传递给下一个observer。
• 最后，将结果传递给 subscribe 订阅块。

你只要记住每一次 operator 都会返回一个新的 observable，不管 operator 有多少个，最终只有最后一个 observable 会被订阅。

不要忘记取消订阅

为什么需要取消订阅

observable 当有数据产生时才会推送给订阅者，所以它可能会无限次向订阅者推送数据。正因为如此，在angular里面创建组件的时候务必要取消订阅操作，以避免内存泄漏，要知道在spa世界里懂得擦屁股是一件必须的事。

unsubscribe

前面示例讲过，调用 subscribe() 后，会返回一个 subscription 可用于取消操作 unsubscribe()。最合理的方式在 ngondestroy 调用它。

ngondestroy() {
  this.inputsubscription.unsubscribe();
}

takewhile

如果组件有很多订阅者的话，则需要将这些订阅者存储在数组中，并组件被销毁时再逐个取消订阅。但，我们有更好的办法：

使用 takewhile() operator，它会在你传递一个布尔值是调用 next() 还是 complete()。

private alive: boolean = true;
ngoninit() {
 const node = document.queryselector('input[type=text]');

 this.s = observable.fromevent(node, 'input')
  .takewhile(() => this.alive)
  .map((event: any) => event.target.value)
  .filter(value => value.length >= 2)
  .subscribe(value => { console.log(value) });
}

ngondestroy() {
 this.alive = false;
}

简单有效，而且优雅。

subject

如果说 observable 与 observer 是攻受结合体的话，那么 subject 就是一个人即攻亦受。正因为如此，我们在写一个service用于数据传递时，总是使用 new subject。

@injectable()
export class messageservice {
  private subject = new subject<any>();

  send(message: any) {
    this.subject.next(message);
  }

  get(): observable<any> {
    return this.subject.asobservable();
  }
}

当f组件需要向m组件传递数据时，我们可以在f组件中使用 send()。

constructor(public srv: messageservice) { }

ngoninit() {
  this.srv.send('w s k f m?')
}

而m组件只需要订阅内容就行：

constructor(private srv: messageservice) {}

message: any;
ngoninit() {
  this.srv.get().subscribe((result) => {
    this.message = result;
  })
}

eventemitter

其实eventemitter跟rxjs没有直接关系，因为他是angular的产物，而非rxjs的东西。或者我们压根没必要去谈，因为eventemitter就是subject。

eventemitter的作用是使指令或组件能自定义事件。

@output() changed = new eventemitter<string>();

click() {
  this.changed.emit('hi~');
}

@component({
 template: `<comp (changed)="subscribe($event)"></comp>`
})
export class homecomponent {
 subscribe(message: string) {
   // 接收：hi~
 }
}

上面示例其实和上一个示例中 messageservice 如出一辙，只不过是将 next() 换成 emit() 仅此而已。

结论

rxjs最难我想就是各种operator的应用了，这需要一些经验的积累。

rxjs很火很大原因我认还是提供了丰富的api，以下是摘抄：

创建数据流：

• 单值：of, empty, never
• 多值：from
• 定时：interval, timer
• 从事件创建：fromevent
• 从promise创建：frompromise
• 自定义创建：create

转换操作：

• 改变数据形态：map, mapto, pluck
• 过滤一些值：filter, skip, first, last, take
• 时间轴上的操作：delay, timeout, throttle, debounce, audit, buffertime
• 累加：reduce, scan
• 异常处理：throw, catch, retry, finally
• 条件执行：takeuntil, delaywhen, retrywhen, subscribeon, observeon
• 转接：switch

组合数据流：

• concat，保持原来的序列顺序连接两个数据流
• merge，合并序列
• race，预设条件为其中一个数据流完成
• forkjoin，预设条件为所有数据流都完成
• zip，取各来源数据流最后一个值合并为对象
• combinelatest，取各来源数据流最后一个值合并为数组

另，最好使用 $ 结尾的命名方式来表示observable，例：input$。

以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持。