码迷,mamicode.com
首页 > Web开发 > 详细

[译]RxJS 5.X基础篇

时间:2016-08-05 17:46:46      阅读:859      评论:0      收藏:0      [点我收藏+]

标签:

欢迎指错与讨论 : )

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

   - 前言

     RxJS在ng2、react( mobx ) 中扮演一个重要角色,因此笔者想学好RxJS,提前做好准备。本文95%非原创,而是笔者对RxJS官网基础篇的翻译,如有错漏请指出。本文主要内容:简介和六大概念(Observable、Observer、Subscription、Subject、Operators、Scheduler)。

   - 安装

    import {Observable} from ‘rxjs/Observable‘ 按需打包,减轻bundle.js大小

   - 一些概念

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

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

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

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

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

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

   - 基础

    - 创建流: 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: ‘‘
});

  - 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 

http://reactivex.io/rxjs/manual/overview.html#creation-operators

 

  - 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,当我们需要以时间为基线时就使用它

 

 

 

       

    

    

     

 

 

    

 

          

     

 

  

       

[译]RxJS 5.X基础篇

标签:

原文地址:http://www.cnblogs.com/BestMePeng/p/RxJS_learn_1.html

(0)
(0)
   
举报
评论 一句话评论(0
登录后才能评论!
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码!