[译]RxJS 5.X基础篇

欢迎指错与讨论 : ）

　　当前RxJS版本：5.0.0-beta.10。更详细的内容尽在RxJS官网http://reactivex.io/rxjs/manual/overview.html。文章比较长，可以通过快捷键 command+f 或者 ctrl+f 搜索主要内容。　　

　　 - 前言

　　　　 RxJS在ng2、redux-observable或者前端数据层中扮演一个重要角色，因此笔者想学好RxJS，提前做好准备。本文95%非原创，而是笔者对RxJS官网基础篇的翻译，如有错漏请指出。本文主要内容：简介和六大概念（Observable、Observer、Subscription、Subject、Operators、Scheduler）。

　　　　具体Api请本博客的另外几篇中文api翻译文章：

　　　　transform（转换）、filter （过滤）、combination/multicasting （组合、广播）、ErrorHanding/Condition/Mathematical（错误处理、情况处理、数学方法）

 

 　　- 安装

　　　　import {Observable} from 'rxjs/Observable' // 1. 按需打包，减轻bundle.js大小

 　　　  import 'rxjs/add/observable/merge'; // 2. 按需导入函数，如merge

　　 - 一些概念

　　　　- Observable( 可被观察的 ) : 是一个包含来自未来、可以被使用的值( value )或事件( event )的集合

　　　　- Observe( 观察者 )：是一个知道如何监听、处理来自Obervable的值的函数集合 

　　　　- Subscription( 订阅 )：代表着Observable的执行动作，我们可以使用它来停止Obervable继续执行

　　　　- Operators( 操作 )：一系列可以操作集合的pure function，像是过滤( filter )、转换( map )等等

　　　　- Subject(  )：相当于一个事件发射器，是唯一能够向多个Observer广播值( value )的唯一手段

　　　　- Schedulers( 调度 )：是一个中央调度员，帮助我们控制并发，协调计算( setTimeout、requestAnimationFrame等 )

　　- Observable( 被观察者 )

　　　　通常而言，Observable都会延迟产生值 ，比如当我们subscribe一个observable的时候它才会向我们发送这些值 let observable = Rx.Observable.range(1,3) 

　　　　- pull( 拉取 ) 每个函数都是一个数据的生产者，每个调用函数的那个'人'，都会希望从这个函数中能够获得(pull) 唯一的返回值

　　　　- push( 推送 ) 在数据生产者中( 如函数 )，会在特定时候把数据推送至消费者，消费者在获得数据之前啥也不会做

　　　　- Observable与函数、promsise的对比：函数是当调用才同步计算，并最终只返回一个值的；promise是会或者不会返回一个值；Observable是当调用才同步或者异步地计算，并可能产生0到无穷多个值的。Observable就像一个没有参数的函数，并不断生成一些值供我们使用，因此它也像是一个事件发射机( EventEmitters )。在Observable中subscribe就像call一个函数，你订阅它，它才会被'启动'。同一个Observable对于不同的subscribe，是不会共享结果的( 通常情况下这样子的，但可以通过调用api来共享 )。

　　　　- Observable四大核心：创建 、订阅 、 执行 、销毁 。

　　　　　　　订阅( subscribe )。当对一个Observable调用多个subscribe函数并创建多个observe时，observe之间不会共享任何东西，因为在Observable.create内部是对observe列表调用各自的回调的  Observable.create(function subscribe(observe){...}) 

　　　　　　　执行( Executing )。Next函数能够将数据传递给Observer，同时在执行期间，能在Observable内部调用多个Next( )函数。同时建议在Observabl内部使用try/catch语法。

　　　　　　　销毁Observe

var observable = Rx.Observable.from([10, 20, 30]);
var subscription = observable.subscribe(x => console.log(x));
// Later:
subscription.unsubscribe();

　　　　　　销毁Observable

var observable = Rx.Observable.create(function subscribe(observer) {
  var intervalID = setInterval(() => {
    ...
  }, 1000);

  return function unsubscribe() {
    clearInterval(intervalID);
  };
});

 　　- Observer( 观察者 )

　　　　什么是观察者？观察者其实是数据的消费者，把来自Observble的数据拿过来使用。同时，Observer的本质是一系列的回调函数，是来自于Observable传递数据后的回调函数。我们可以直接通过subscribe函数创建观察者

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

　　- Subscription( 订阅 )

 　　　　什么是Subscription？它其实是代表着Observable的'执行'的对象，我们可以通过它的 unsubscribe 方法销毁Observable的执行。同时我们能使用 add 方法，一次销毁多个

var subscription = observable1.subscribe(x => console.log('first: ' + x));
var childSubscription = observable2.subscribe(x => console.log('second: ' + x));

subscription.add(childSubscription);

setTimeout(() => {
  subscription.unsubscribe();
}, 1000);

 　　- Subject(  )

　　　　什么是Subject？它是在Rx中一种比较特殊的Observable( 同时它也是Observer )，它能够让值( value )同时向多个Observer传播( 广播 )。而一般的Observable都是' 单播 '形式，即：每一个订阅了同一个Observable的observer，实际上是拥有不同的、独立的Observable的执行( 原文：each subscribed Observer owns an independent execution of the Observable )，而Subject是多播的。

// Observable对比Subject
let source = Rx.Observable.create((observer)=>{
    observer.next('1');
    observer.next('2');
});
source.subscribe((x)=>{console.log(x);});
source.subscribe((x)=>{console.log(x);});
// 输出 1 2 1 2
let subject = new Rx.Subject();
subject.subscribe((x)=>{console.log(`${x}`);});
subject.subscribe((x)=>{console.log(`${x}`);});
subject.next(1);
subject.next(2);
// 输出 1 1 2 2 　　　　

// Subject可以在'多播'情景下对Observable进行优化
// 明显看到在subject下，Observable只执行了一次
var source = Rx.Observable.create((observer)=>{
    console.log(`source was called`);
    observer.next(1);observer.next(2);observer.next(3);
});
source.subscribe({next: (v) => console.log('observerA: ' + v)});
source.subscribe({next: (v) => console.log('observerB: ' + v)});
// 输出为 'source was called' observerA: 1 observerA: 2
//       'source was called'  observerB: 1 observerB: 2


var source = Rx.Observable.create((o)=>{
    console.log(`source was called`);
    o.next(1);o.next(2);
});
var subject = new Rx.Subject();
var multicasted = source.multicast(subject);

// 原理是`subject.subscribe({...})`:返回的multicasted是一个connectableObservable 
multicasted.subscribe({next: (v) => console.log('observerA: ' + v)});
multicasted.subscribe({next: (v) => console.log('observerB: ' + v)});

// 原理是 `source.subscribe(subject)`:
multicasted.connect();
// 输出 ‘source was called’ observerA: 1 observerB: 1
//                          observerA: 2 observerB: 2

　　　　可以单个取消订阅状态 observer.subscribe( ) ，也可以直接销毁Subject subscriptionConnect.unsubscribe( ) 

　　　　- refCount( ) 该api能监听当前connectableObservable的'连接状态'，当有大于0个subscribe挂在它上面它会自动执行 connect ，无subscribtion的时候会自动执行 unsubscribe( ) 

var refCounted = source.multicast(subject).refCount();

　　　　- BehaviorSubject（） 它能储存上一次从Observable发过来的值，每当有新的observer的时，会把改值立即发送给它

var subject = new Rx.BehaviorSubject(0)// 初始值

　　　　- ReplaySubject( 重现 ) 是BehaviorSubject的加强版，它能储存从Observable发送过来的一系列值，当有新的observer连接时，会把这些值发送给它。

var subject = new Rx.ReplaySubject(3); // 缓存空间为3

　　     - AsyncSubject（异步）仅将从Observable发送过来的最后一个值发送给Observe

var subject = new Rx.AsyncSubject();

　　- Operators 

　　　　什么是Operators？Operators是Rx中最有用的一系列函数，它们建立在Observable之上，并能优雅地以同步代码的写法，将不同的异步流代码链式组合到一起。同时，Operator都是pure的，因为这些函数不会直接修改传入进来的Observable，而是经过修饰之后返回一个新Observable。

　　　　在链式操作中，下一个operator会根据上一个operator修改后的Observable继续工作。

　　　　- 具体请本博客的另外几篇中文api翻译文章：

　　　　transform（请点击）、filter （请点击）、combination/multicasting （请点击）、ErrorHanding/Condition/Mathematical（请点击）

　　- Scheduler

　　　　Scheduler有咩用？Scheduler能控制一个订阅的开始、数据的传递。它由三部分组成：

　　　　　- scheduler是一种数据结构，它知道如何基于优先级或者其他标准对任务队列进行储存

　　　　　- scheduler是一个执行环境，它指何时何地地执行任务

　　　　　- scheduler有一个' 时钟 '的概念，它能让我们自己定义，当Observable向Observer时会处于什么环境下( context )

// observeOn( RX.Scheduler.async )
var observable = Rx.Observable.create(function (observer) {
  observer.next(1);observer.next(2);
})
.observeOn(Rx.Scheduler.async);

console.log('just before subscribe');
observable.subscribe({next: x => console.log('got value ' + x)});
console.log('just after subscribe');
// 输出
// just before subscribe just after subscribe
// got value 1 got value 2

　　　　　　- Scheduler类型

　　　　　　　　- Scheduler.queue 基于当前的事件框架在一个队列上工作，当我们需要遍历操作时可以使用它

　　　　　　　　- Scheduler.asap 

　　　　　　　　- Scheduler.async 在内部，Schedules会使用setInterval，当我们需要以时间为基线时就使用它

　　 - 基础

　　　　- 创建流: Observable.create、of、from、fromEvent( target,eventType )、fromPromise、bindCallback( 把callback写法转换为链式写法 )　　　　　

let exists = Rx.Observable.bindCallback(fs.exists);
exists('file.txt').subscribe( exist=>console.log(`exist? : ${exist} `) )

　　　　- 事件流: filter、delay( 延时 )、throttleTime( 时间间隔 )、debounceTime( 事件暂停x毫秒后 )、take( 执行x次后停止 )、takeUtil( 取消订阅 )

let inputStream = Rx.Observable.fromEvent(document.querySelector('input'), 'keyup')
let stopStream = Rx.Observable.fromEvent(document.querySelector('button'), 'click')
inputStream.throttleTime(200) // 每隔200毫秒才释放一次
                .takeUtil(stopStream) // 当触发时，停止订阅
                .subscribe(($event)=> {console.log(`${$event.target.value}`)}) // 事件对象

　　　　- 转换流、过滤流：map、distinct( 过滤重复 )、distince( 过滤连续重复 )

// 输入hello world
input.pluck('target', 'value').distinctUntilChanged()
  .subscribe(value => console.log(value)); // "helo wrd"
input.pluck('target', 'value').distinct()
  .subscribe(value => console.log(value)); // "helo world"

　　- 创建一个应用

　　　　- RxJS的工具函数都是pure、无状态的一类函数。但我们的应用中往往需要记录大量的state，那在Rx如何记录状态呢？类似于redux，rx中的scan函数能够使用reduce函数，将流的结果合并。我们需要做的仅仅是保存这个状态( state )值

// 基础模式
var button = document.querySelector('button');
Rx.Observable.fromEvent(button, 'click')
  // 内部的reduce函数能够积累(此处是相加，若是对象可以是Object.assign),并相应变化
  .scan(count => count + 1, 0)
  // Set the count on an element each time it changes
  .subscribe(count => document.querySelector('#count').innerHTML = count);

　　　　- 一个更好的模式

var button = document.querySelector('button');
Rx.Observable.fromEvent(button, 'click')
    // 返回了一个, 类似redux中 经过switch(action.type)的reducer 与函数
   .map(() => state => Object.assign({}, state, {count: state.count + 1}))
   // 直接把函数作用于初始化state，scan同时能保存这个更新后的state
   .scan((state, changeFn) => changeFn(state), {count: 0})
   .subscribe(state => document.querySelector('#count').innerHTML = state.count);

　　　　 - 对于某一个组件还可以

var state = Rx.Observable.merge(
  increase,
  decrease,
  input
).scan((state, changeFn) => changeFn(state), {
  count: 0,
  inputValue: ''
});