RxJS:万物皆为流

首先贴上一个中文Rxjs文档链接，从官网被人直译下来的，也是比较全面的介绍rxjs了。最好的Rxjs中文文档

RXJS全名Reactive Extensions for JavaScript，JavaScript的响应式扩展。什么是响应式？响应式就是跟随时间不断变化的数据、状态、事件等转换成可被观察的序列，然后订阅那些变化，一旦变化则会执行业务逻辑。适用于异步场景。

RxJS所能解决的问题：

时刻保持响应。这对于一个应用来说意味着当他处理用户的输入或者凭借AJAX从服务器接受一些数据时停止是一件不可能接受的事情。在JavaScript中解决问题的方案始终是大量运用回调函数来进行一些运行的处理。但回调的使用使内容丰富的大型应用变得凌乱，一旦你需要多块数据时你就陷入了回调地狱。Angular2中，组件间通讯@Output对应的EventEmitter实际上就是一个Subject(主题：同时为Observable和Observer);Http模块中Observable作为大部分API的交互对象使用。但是这只是官方的外部扩展，并不必须，也可以使用.toPromise的方式转换为Promise来使用或第三方扩展库或Fetch API 传统Ajax已死，Fetch永生。

RxJS初探：

RxJS是一个解决异步问题的JS开发库.它带来了观察者模式和函数式编程的相结合的最佳实践。 观察者模式是一个被实践证明的模式，基于生产者（事件的创建者）和消费者（事件的监听者）的逻辑分离关系.况且函数式编程方式的引入，如说明性编程，不可变数据结构，链式方法调用会使你极大的简化代码量。RxJS 引入了一个重要的数据类型——流（stream）。流（Streams）无非是随时间流逝的一系列事件。流（Streams）可以用来处理任何类型的事件，如：鼠标点击，键盘按下，等等。你可以把流作为变量，它有能力从数据角度对发生的改变做出反应。

Stream

流（Streams）无非是随时间流逝的一系列事件。流（Streams）可以用来处理任何类型的事件，如：鼠标点击，键盘按下等等。你可以把流作为变量，它有能力从数据角度对发生的改变做出反应。

var a = 2;
var b = 4;
var c = a + b;
console.log(c); //-> 6
a = 10;  // reassign a
console.log(c); //-> still 6

事件引发的改变总是从事件源（生产者）,向下传递到所有事件监听方（消费者）。如果变量看做流呢，又是什么结果呢？

var A$ = 2;
var B$ = 4;
var C$ = A$ + B$;
console.log(C$); //-> 6
A$ = 10;  
console.log(C$); //->  16

流的方式重新定义的变量值的动态行为.换句话说，C$是把两个流变量A$和B$进行合并操作。当一个新值被推进了A$，C$立刻响应式的变更为16。

开始了解RxJS中的几个重要成员

• Observable可观察对象：表示一个可调用的未来值或者事件的集合。
• Observer观察者：一个回调函数集合,它知道怎样去监听被Observable发送的值
• Subscription订阅： 表示一个可观察对象的执行，主要用于取消执行。
• Operators操作符： 纯粹的函数，使得以函数编程的方式处理集合比如:map,filter,contact,flatmap。
• Subject(主题)：等同于一个事件驱动器，是将一个值或者事件广播到多个观察者的唯一途径。
• Schedulers(调度者)： 用来控制并发，当计算发生的时候允许我们协调，比如setTimeout,requestAnimationFrame。

  第一个例子

  使用RxJS创建一个可观察对象：

  var button = document.querySelector('button');
  Rx.Observable.fromEvent(button, 'click')
  .throttleTime(1000)
  .scan(count => count + 1, 0)   
  .subscribe(count => console.log(`Clicked ${count} times`));

  observable可观察对象，以惰性的方式推送多值的集合

  下面的例子是一个推送1,2，3,4数值的可观察对象，一旦它被订阅1,2，3,就会被推送，4则会在订阅发生一秒之后被推送，紧接着完成推送

  var observable = Rx.Observable.create(observer=>{
      observer.next(1);
      observer.next(2);
      observer.next(3);
      setTimeout(() => {
          observer.next(4);
          observer.complete();
      }, 1000);
  });

  Pull拉取 VS Push推送

  拉和推是数据生产者和数据的消费者两种不同的交流协议(方式);

  
  

  什么是”Pull拉”？在”拉”体系中，数据的消费者决定何时从数据生产者那里获取数据，而生产者自身并不会意识到什么时候数据将会被发送给消费者。

  每一个JS函数都是一个“拉”体系，函数是数据的生产者，调用函数的代码通过“拉出”一个单一的返回值来消费该数据(return 语句)。

什么是”Push推”？在推体系中，数据的生产者决定何时发送数据给消费者，消费者不会在接收数据之前意识到它将要接收这个数据。

Promise(承诺))是当今JS中最常见的Push推体系，一个Promise(数据的生产者)发送一个resolved value(成功状态的值)来注册一个回调(数据消费者)，但是不同于函数的地方的是：Promise决定着何时数据才被推送至这个回调函数。

RxJS引入了Observables(可观察对象)，一个全新的”推体系”。一个可观察对象是一个产生多值的生产者，并”推送给”Observer(观察者)。

RxJS VS Promise—3个最重要的不同点

不同点	Rxjs	Promise
动作是否可以取消？	是	否
是否可以发射多个值？	是	否
各种工具函数？	是	否

Operators操作符

create() 创建一个可观察序列，参数为一个封装数据生成逻辑的函数，该函数的参数为观察者

empty() 不需要传递参数，创建一个空序列并立即结束

Rx.Observable.empty()  =  Rx.Observable.create((o)=>{
    o.onCompleted()
})

never() 不需要传递参数，创建一个空序列，并永远不结束

Rx.Observable.never()  =  Rx.Observable.create((o)=>{
    
})

throw() 创建一个空序列，参数来声明错误并立即抛出错误

Rx.Observable.throw('error')  =  Rx.Observable.create((o)=>{
    0.onError('error')
})

range() 创建一个有线长度的整数序列，两个参数，第一个起始值，第二个元素数量

Rx.Observable.range(30,4)  //输出:30,31,32,33

interval() 创建一个无限长度的周期性序列

Rx.Observable.interval(1000) //输出:0,1,2...

timer() 指定一个额外的参数来调节第一值的静默时长，第二个参数可选，若无则仅仅在规定的静默时长后输出一个值，然后结束序列

Rx.Observable.timer(0,1000) //输出:0,1,2...

from() 可以将已有的数据转化为Observable,参数为iterable数据集对象,常见Array,String

Rx.Observable.from(iterable)

of() 不在同一个数据集中的多个来源的数据，使用of()方法直接构造：

Rx.Observable.of([1,2,3])    //    [1,2,3]
Rx.Observable.from([1,2,3])  //    1,2,3

just() 将任何数据转化为一个单值输出的Observable

Rx.Observable.just([1,2,3])    //    [1,2,3]

repeat() 创建一个重复值序列 par1:值，par2：次数

Rx.Observable.repeat('a',2)    //   a a

fromEvent() 将事件流转化为Observable,

var el = document.getElementById("btn"); //DOM对象作为事件源
Rx.Observable.fromEvent(el,"click");

toArray() 将序列还原为数组对象,只有在订阅后才还原为数组

var source = Rx.Observable.of(1,2,3,4);    //序列：1 2 3 4 
var target = source.toArray();     //序列：[1,2,3,4]
target.subscribe(function(d){
    console.log(d);             //d： [1,2,3,4]
})

delay() 推迟 参数为数字或Date对象

delaySubscription() 延迟订阅 参数同理

startWith() 可以在源序列之前添加额外的元素

var source = Rx.Observable.of(1,2,3); //序列：1 2 3
var target = source.startWith(7,8,9); //序列：7 8 9 1 2 3

map() 对源序列进行变换，并返回新的序列(改变了源)

var trandform = function(item){
    return item*2;
}
var source = Rx.Observable.of(1,2,3); //输出： 1 2 3
var target = source.map(trandform);   //输出： 2 4 6

concat() 有序拼接 merge()无序

var source = Rx.Observable.of(1,2,3); //序列：1 2 3
var target = source.concat(
    Rx.Observable.of(4,5,6),
    Rx.Observable.of(7,8,9)
); //序列：1 2 3 4 5 6 7 8 9

concatAll() 如果源序列的元素也是序列 —— 序列的序列，那么可以使用concatAll() 方法将各元素序列按顺序拼接起来

var source = Rx.Observable.of(10,20,30)
    .map(function(item){
        return Rx.Observable.range(item,3);
    }); //序列： Observable{10,11,12} Observable{20,21,22} Observable{30,31,32}
var target = source.concatAll(); //序列：10 10 12 20 21 22 30 31 32

catch() 捕捉源序列错误，返回新序列

var source = Rx.Observable.create(function(o){
    o.onNext(1);
    o.onNext(2);
    o.onError(new Error("fake error"));
    o.onNext(4);
}); //序列： 1 2 <ERROR> 4
var target = source.catch(Rx.Observable.from("abc")); //序列： 1 2 a b c

pluck() 针对元素为json对象的源序列，返回指定属性的值的序列

var source = Rx.Observable.of({name:'john',age:33},{name:'lee',age:22});
var target = source.pluck('name');  //序列 john lee

flatMap() 平坦化映射:首先将一个序列的各元素映射为序列，然后将各序列融合 参数是一个映射函数，返回值为序列

var source = Rx.Observable.of(10,20);
var mf = function(item){
    return Rx.Observable.range(item,3)
}
var target = source.flatMap(mf)  // 序列：10,11,20,12,21,22

flatMapLatest() 与flatMap()的区别在于将最新的序列中的元素输出

concatMap() 将源序列各元素映射为序列，然后按顺序拼接 (与flatMap的区别所在)

var source = Rx.Observable.of(1,2,3);
var mf = function(item){
    return Rx.Observable.range(item,3);
}
var target = source.concatMap(mf)  //序列：1,2,3,2,3,4,3,4,5

flatMap与map异同点

Map用于对自身对象数值进行映射，将发射对象转换成另一个发射对象发射，返回一个包含映射结果的Observable对象。而flatMap是把自身对象里的数值进行映射并转换成一个新的Observable对象.当从其他类型对象中构建Observable对象时，需要使用flatMap方法

flatMap与concatMap异同点

merge和map结合与concat和map结合的区别，归根结底为merge与concat的区别，一个无序一个有序

filter() 筛选源序列中满足条件的元素，并返回新的序列

var source = Rx.Observable.of(1,2,3,4,5); //序列： 1 2 3 4 5
var target = source.filter(x => x<4)  //序列： 1 2 3

skip(num) 抑制序列头部元素数量输出 skipLast(num)尾部 skipWhile(if)指定一个条件

take(num) 截取序列头部元素数量输出 takeLast(num)尾部 takeWhile(if)指定一个条件

distinct 去重,并返回一个新序列

var source = Rx.Observable.of(1,2,3,4,2,1); //序列： 1 2 3 4 2 1
var target = source.distinct(); //序列：1 2 3 4

distinctUntilChanged 去重,并返回一个新序列

var source = Rx.Observable.of(1,2,2,3,4,2,1); //序列： 1 2 3 4 2 1
var target = source.distinctUntilChanged(); //序列：1 2 3 4 2 1

debounce 去抖动,一段时间内只取最新数据作为一次发射数据，其他数据取消发射

throttle(和debounce唯一区别是debounce取一段时间内最新的，而throttle忽略这段时间后，发现新值才发送, 通俗讲，都设定一个时间周期，持续触发事件，throttle为每到时间周期便会触发一次，bebounce为触发周期小于设定时间周期不予事件触发）

buffer() 使用第二个序列触发源序列中多个元素的打包

var source = Rx.Observable.timer(0,1000);    //序列：0 1 2 3 ...
var boundaries = Rx.Observable.timer(2500); //延时2500ms触发
var target = source.buffer(boundaries); //序列： [0,1,2]

bufferWithTime() 按固定时间间隔对源序列进行打包

var source = Rx.Observable.timer(0,1000); //序列：0 1 2 3 ...
var target = source.bufferWithTime(2500); //序列：[0,1,2] [3,4] ...

zip() 支持可变数量的序列作为参数，最后一个参数应当是一个组合函数， 其返回值将作为目标序列的元素

var source1 = Rx.Observable.of(1,2,3); //序列： 1 2 3
var source2 = Rx.Observable.of(4,5,6); //序列：4 5 6
var cf = function(d1,d2){ return d1 + '-' + d2;};
var target = Rx.Observable.zip(source1,source2,cf); //序列： 1-4 2-5 3-6

forkJoin() 将多个序列的最后一个元素组合为一个数组后，作为目标序列的唯一元素

var source1 = Rx.Observable.of(1,2,3); //序列： 1 2 3
var source2 = Rx.Observable.of(4,5,6); //序列： 4 5 6
var target = Rz.Observable.forkJoin(source1,source2); //序列：[3,6]

combineLatest() 将多个序列的最后一个元素，使用组合函数构成目标序列的一个新元素

var source1 = Rx.Observable.interval(200).map(x => 'First:' + x)
var source2 = Rx.Observable.interval(100).map(x => 'Second:' + x)
var source = Rx.Observable.combineLatest(
    source1,
    source2
).take(4);   // ["First:0", "Second:0"]  ["First:0", "Second:1"]  ["First:0", "Second:2"]   ["First:1", "Second:2"]