将RxJS融入React项目

2024年5月31日 133次阅读来源: linshuizhaoying

媒介

近来预备毕设，手艺选型的时刻由于功用的一些需求预备将RxJs融入到项目中，斟酌RxJs的时刻由于之前的手艺栈还犹疑了一下，查了一些材料以及大略阅读了一些文档。觉得关于毕设项目RxJs的到场是有协助的，因而盘算系统的进修然后摘抄学问点以及实践一些demo做手艺积聚。

RxJS手艺积聚

RxJs经由社区的努力进修材料照样很多的，官方中文文档就已很不错，不过我们先从30 天通晓 RxJS开端感受一下RxJS.然后合营一些中文文档来补充学问点，末了再依据官方文档来校验全部学问系统。

RxJS 基本引见

RxJS是一套由Observable sequences来组合异步行动和事宜基本顺序的Library

RxJS 是Functional Programming 跟Reactive Programming 的连系

把每一个运算包成一个个差别的function，并用这些function 组合出我们要的结果，这就是最简朴的Functional Programming

Functional Programming 强调没有Side Effect，也就是function 要坚持地道，只做运算并返回一个值，没有其他分外的行动。

Side Effect

Side Effect是指一个function做了跟自身运算返回值没有关系的事，比方说修正某个全域变数，或是修正传入参数的值，以至是实行console.log都算是Side Effect。

前端罕见的Side Effect:

发送http request
在画面输出值或是log
获得用户的input
Query DOM

Reactive Programming简朴来讲就是当变数或资本发作更改时，由变数或资本自动告诉我发作更改了

Observable

Observer Pattern（视察者情势）

Observer Pattern 实在很常碰到，很多API 的设想上都用了Observer Pattern，最简朴的例子就是DOM 物件的事宜监听:


function clickHandler(event) {
    console.log('user click!');
}

document.body.addEventListener('click', clickHandler)

视察者情势:我们可以对某件事注册监听，并在事宜发作时，自动实行我们注册的监听者(listener)。

Es5版本:


function Producer() {
    
    // 这个 if 只是防止运用者不小心把 Producer 看成函数挪用
    if(!(this instanceof Producer)) {
      throw new Error('请用 new Producer()!');
    }
    
    this.listeners = [];
}

// 到场监听的要领
Producer.prototype.addListener = function(listener) {
    if(typeof listener === 'function') {
        this.listeners.push(listener)
    } else {
        throw new Error('listener 必需是 function')
    }
}

// 移除监听的要领
Producer.prototype.removeListener = function(listener) {
    this.listeners.splice(this.listeners.indexOf(listener), 1)
}

// 发送关照的要领
Producer.prototype.notify = function(message) {
    this.listeners.forEach(listener => {
        listener(message);
    })
}

es6 版本


class Producer {
    constructor() {
        this.listeners = [];
    }
    addListener(listener) {
        if(typeof listener === 'function') {
            this.listeners.push(listener)
        } else {
            throw new Error('listener 必需是 function')
        }
    }
    removeListener(listener) {
        this.listeners.splice(this.listeners.indexOf(listener), 1)
    }
    notify(message) {
        this.listeners.forEach(listener => {
            listener(message);
        })
    }
}

挪用例子:


var egghead = new Producer(); 

function listener1(message) {
    console.log(message + 'from listener1');
}

function listener2(message) {
    console.log(message + 'from listener2');
}

egghead.addListener(listener1);egghead.addListener(listener2);

egghead.notify('A new course!!')

输出:

a new course!! from listener1
a new course!! from listener2

Iterator Pattern （迭代器情势）

JavaScript 到了ES6 才有原生的Iterator
在ECMAScript中Iterator最早现实上是要采纳相似Python的Iterator范例，就是Iterator在没有元素以后，实行next会直接抛出毛病；但厥后经由一段时候议论后，决议采更functional的做法，改成在获得末了一个元素以后实行next永久都回传{ done: true, value: undefined }


var arr = [1, 2, 3];

var iterator = arr[Symbol.iterator]();

iterator.next();
// { value: 1, done: false }
iterator.next();
// { value: 2, done: false }
iterator.next();
// { value: 3, done: false }
iterator.next();
// { value: undefined, done: true }

简朴完成:


es5:

function IteratorFromArray(arr) {
    if(!(this instanceof IteratorFromArray)) {
        throw new Error('请用 new IteratorFromArray()!');
    }
    this._array = arr;
    this._cursor = 0;    
}

IteratorFromArray.prototype.next = function() {
    return this._cursor < this._array.length ?
        { value: this._array[this._cursor++], done: false } :
        { done: true };
}

es6:

class IteratorFromArray {
    constructor(arr) {
        this._array = arr;
        this._cursor = 0;
    }
  
    next() {
        return this._cursor < this._array.length ?
        { value: this._array[this._cursor++], done: false } :
        { done: true };
    }
}

上风

Iterator的特征可以拿来做耽误运算(Lazy evaluation)，让我们能用它来处置惩罚大数组。
第二由于iterator 自身是序列，所以可以第挪用要领像map, filter… 等！

耽误运算(Lazy evaluation)


function* getNumbers(words) {
        for (let word of words) {
            if (/^[0-9]+$/.test(word)) {
                yield parseInt(word, 10);
            }
        }
    }
    
    const iterator = getNumbers('30 天通晓 RxJS (04)');
    
    iterator.next();
    // { value: 3, done: false }
    iterator.next();
    // { value: 0, done: false }
    iterator.next();
    // { value: 0, done: false }
    iterator.next();
    // { value: 4, done: false }
    iterator.next();
    // { value: undefined, done: true }

把一个字串丢进getNumbersh函数时，并没有立时运算出字串中的一切数字，必需比及我们实行next()时，才会真的做运算，这就是所谓的耽误运算(evaluation strategy)

Observable简介

Observer跟Iterator有个共通的特征，就是他们都是渐进式 (progressive)的获得材料，差别只在于Observer是生产者(Producer)推送材料(push )，而Iterator是消费者(Consumer)要求材料(pull)!

Observable实在就是这两个Pattern头脑的连系，Observable具有生产者推送材料的特征，同时能像序列，具有序列处置惩罚材料的要领 (map, filter...)！

RxJS说白了就是一个中心三个重点。

一个中心是Observable 再加上相干的Operators(map, filter…)，这个部分是最主要的，其他三个重点本质上也是围绕着这个中心在转，所以我们会花快要20 天的篇数讲这个部分的观念及运用案例。

别的三个重点分别是

Observer
Subject
Schedulers

Observable 实践

Observable 同时可以处置惩罚同步与异步的行动！


同步操纵

var observable = Rx.Observable
    .create(function(observer) {
        observer.next('Jerry'); 
        observer.next('Anna');
    })
    
// 定阅 observable    
observable.subscribe(function(value) {
    console.log(value);
})

> Jerry
> Anna

异步操纵:

var observable = Rx.Observable
    .create(function(observer) {
        observer.next('Jerry'); // RxJS 4.x 之前的版本用 onNext
        observer.next('Anna');
        
        setTimeout(() => {
            observer.next('RxJS 30 days!');
        }, 30)
    })
    
console.log('start');
observable.subscribe(function(value) {
    console.log(value);
});
console.log('end');

>
start
Jerry
Anna
end
RxJS 30 days!

视察者Observer

Observable 可以被定阅(subscribe)，或说可以被视察，而定阅Observable的又称为视察者(Observer)。
视察者是一个具有三个要领(method)的对象，每当Observable 发作事宜时，便会呼唤视察者相对应的要领。

next：每当Observable 发送出新的值，next 要领就会被呼唤。
complete：在Observable 没有其他的材料可以获得时，complete 要领就会被呼唤，在complete 被呼唤以后，next 要领就不会复兴作用。
error：每当Observable 内发作毛病时，error 要领就会被呼唤。


var observable = Rx.Observable
    .create(function(observer) {
            observer.next('Jerry');
            observer.next('Anna');
            observer.complete();
            observer.next('not work');
    })
    
// 定义一个视察者
var observer = {
    next: function(value) {
        console.log(value);
    },
    error: function(error) {
        console.log(error)
    },
    complete: function() {
        console.log('complete')
    }
}

//  定阅 observable    
observable.subscribe(observer)

>
Jerry
Anna
complete


// complete实行后，next就会自动失效，所以没有印出not work。

捕捉毛病实例：

var observable = Rx.Observable
  .create(function(observer) {
    try {
      observer.next('Jerry');
      observer.next('Anna');
      throw 'some exception';
    } catch(e) {
      observer.error(e)
    }
  });

var observer = {
    next: function(value) {
        console.log(value);
    },
    error: function(error) {
        console.log('Error: ', error)
    },
    complete: function() {
        console.log('complete')
    }
}

observable    
observable.subscribe(observer)

>
Jerry
Anna
Error:  some exception

视察者可以是不完整的，他可以只具有一个next 要领

定阅一个Observable 就像是实行一个function

Operator操纵符

Operators 就是一个个被附加到Observable 型别的函数，比方像是map, filter, contactAll… 等等
每一个operator都邑回传一个新的observable，而我们可以透过create的要领竖立种种operator

Observable 有很多建立实例的要领，称为creation operator。下面我们列出RxJS 经常使用的creation operator:


create
of
from
fromEvent
fromPromise
never
empty
throw
interval
timer

当我们想要同步的通报几个值时，就可以用of这个operator来简约的表达!


var source = Rx.Observable.of('Jerry', 'Anna');

source.subscribe({
    next: function(value) {
        console.log(value)
    },
    complete: function() {
        console.log('complete!');
    },
    error: function(error) {
        console.log(error)
    }
});

// Jerry
// Anna
// complete!

用from来吸收任何可罗列的参数（Set, WeakSet, Iterator 等都可）


var arr = ['Jerry', 'Anna', 2016, 2017, '30 days'] 
var source = Rx.Observable.from(arr);

source.subscribe({
    next: function(value) {
        console.log(value)
    },
    complete: function() {
        console.log('complete!');
    },
    error: function(error) {
        console.log(error)
    }
});

// Jerry
// Anna
// 2016
// 2017
// 30 days
// complete!


var source = Rx.Observable.from('123');

source.subscribe({
    next: function(value) {
        console.log(value)
    },
    complete: function() {
        console.log('complete!');
    },
    error: function(error) {
        console.log(error)
    }
});
// 1
// 2
// 3
// complete!


var source = Rx.Observable
  .from(new Promise((resolve, reject) => {
    setTimeout(() => {
      resolve('Hello RxJS!');
    },3000)
  }))
  
source.subscribe({
    next: function(value) {
        console.log(value)
    },
    complete: function() {
        console.log('complete!');
    },
    error: function(error) {
    console.log(error)
    }
});

// Hello RxJS!
// complete!

可以用Event竖立Observable，经由过程fromEvent的要领


var source = Rx.Observable.fromEvent(document.body, 'click');

source.subscribe({
    next: function(value) {
        console.log(value)
    },
    complete: function() {
        console.log('complete!');
    },
    error: function(error) {
        console.log(error)
    }
});

fromEvent的第一个参数要传入DOM ，第二个参数传入要监听的事宜称号。上面的代码会针对body 的click 事宜做监听，每当点击body 就会印出event。

猎取 DOM 的经常使用要领：
document.getElementById()
document.querySelector()
document.getElementsByTagName()
document.getElementsByClassName()

Event来竖立Observable实例另有另一个要领fromEventPattern，这个要领是给类事宜运用

所谓的类事宜就是指其行动跟事宜相像，同时具有注册监听及移除监听两种行动，就像DOM Event有addEventListener及removeEventListener一样


class Producer {
    constructor() {
        this.listeners = [];
    }
    addListener(listener) {
        if(typeof listener === 'function') {
            this.listeners.push(listener)
        } else {
            throw new Error('listener 必须是 function')
        }
    }
    removeListener(listener) {
        this.listeners.splice(this.listeners.indexOf(listener), 1)
    }
    notify(message) {
        this.listeners.forEach(listener => {
            listener(message);
        })
    }
}

var egghead = new Producer(); 


var source = Rx.Observable
    .fromEventPattern(
        (handler) => egghead.addListener(handler), 
        (handler) => egghead.removeListener(handler)
    );
  
source.subscribe({
    next: function(value) {
        console.log(value)
    },
    complete: function() {
        console.log('complete!');
    },
    error: function(error) {
        console.log(error)
    }
})

egghead.notify('Hello! Can you hear me?');

字数受限，可以去博客看完整版

Subject简介

Subject 可以拿去定阅Observable(source) 代表他是一个Observer，同时Subject 又可以被Observer(observerA, observerB) 定阅，代表他是一个Observable。

Subject 同时是Observable 又是Observer
Subject 会对内部的observers 清单举行组播(multicast)

Subject运用

Subject 在内部治理一份observer 的清单，并在吸收到值时遍历这份清单并送出值


var subject = new Rx.Subject();

var observerA = {
    next: value => console.log('A next: ' + value),
    error: error => console.log('A error: ' + error),
    complete: () => console.log('A complete!')
}

var observerB = {
    next: value => console.log('B next: ' + value),
    error: error => console.log('B error: ' + error),
    complete: () => console.log('B complete!')
}

subject.subscribe(observerA);
subject.subscribe(observerB);

subject.next(1);
// "A next: 1"
// "B next: 1"
subject.next(2);
// "A next: 2"
// "B next: 2"

这里我们可以直接用subject 的next 要领传送值，一切定阅的observer 就会吸收到，又由于Subject 自身是Observable，所以如许的运用体式格局很合适用在某些没法直接运用Observable 的前端框架中，比方在React 想对DOM 的事宜做监听


class MyButton extends React.Component {
    constructor(props) {
        super(props);
        this.state = { count: 0 };
        this.subject = new Rx.Subject();
        
        this.subject
            .mapTo(1)
            .scan((origin, next) => origin + next)
            .subscribe(x => {
                this.setState({ count: x })
            })
    }
    render() {
        return <button onClick={event => this.subject.next(event)}>{this.state.count}</button>
    }
}

BehaviorSubject

BehaviorSubject 是用来显现当前的值，而不是纯真的发送事宜。BehaviorSubject 会记着最新一次发送的元素，并把该元素看成现在的值，在运用上BehaviorSubject 建构式须要传入一个参数来代表肇端的状况


// BehaviorSubject 在竖立时就须要给定一个状况，并在以后任何一次定阅，就会先送出最新的状况。实在这类行动就是一种状况的表达而非单存的事宜，就像是岁数跟华诞一样，岁数是一种状况而华诞就是事宜；所以当我们想要用一个stream 来表达岁数时，就应该用BehaviorSubject 。

var subject = new Rx.BehaviorSubject(0); // 0
var observerA = {
    next: value => console.log('A next: ' + value),
    error: error => console.log('A error: ' + error),
    complete: () => console.log('A complete!')
}

var observerB = {
    next: value => console.log('B next: ' + value),
    error: error => console.log('B error: ' + error),
    complete: () => console.log('B complete!')
}

subject.subscribe(observerA);
// "A next: 0"
subject.next(1);
// "A next: 1"
subject.next(2);
// "A next: 2"
subject.next(3);
// "A next: 3"

setTimeout(() => {
    subject.subscribe(observerB); 
    // "B next: 3"
},3000)

ReplaySubject

在新定阅时从新发送末了的几个元素，这时候我们就可以用ReplaySubject


var subject = new Rx.ReplaySubject(2); // 反复发送末了俩个元素
var observerA = {
    next: value => console.log('A next: ' + value),
    error: error => console.log('A error: ' + error),
    complete: () => console.log('A complete!')
}

var observerB = {
    next: value => console.log('B next: ' + value),
    error: error => console.log('B error: ' + error),
    complete: () => console.log('B complete!')
}

subject.subscribe(observerA);
subject.next(1);
// "A next: 1"
subject.next(2);
// "A next: 2"
subject.next(3);
// "A next: 3"

setTimeout(() => {
    subject.subscribe(observerB);
    // "B next: 2"
    // "B next: 3"
},3000)

AsyncSubject

在subject完毕后送出末了一个值


var subject = new Rx.AsyncSubject();
var observerA = {
    next: value => console.log('A next: ' + value),
    error: error => console.log('A error: ' + error),
    complete: () => console.log('A complete!')
}

var observerB = {
    next: value => console.log('B next: ' + value),
    error: error => console.log('B error: ' + error),
    complete: () => console.log('B complete!')
}

subject.subscribe(observerA);
subject.next(1);
subject.next(2);
subject.next(3);
subject.complete();
// "A next: 3"
// "A complete!"

setTimeout(() => {
    subject.subscribe(observerB);
    // "B next: 3"
    // "B complete!"
},3000)

Observable and Subject

multicast

multicast 可以用来挂载subject 并回传一个可贯穿连接(connectable)的observable


var source = Rx.Observable.interval(1000)
             .take(3)
             .multicast(new Rx.Subject());

var observerA = {
    next: value => console.log('A next: ' + value),
    error: error => console.log('A error: ' + error),
    complete: () => console.log('A complete!')
}

var observerB = {
    next: value => console.log('B next: ' + value),
    error: error => console.log('B error: ' + error),
    complete: () => console.log('B complete!')
}

source.subscribe(observerA); // subject.subscribe(observerA)

source.connect(); // source.subscribe(subject)

setTimeout(() => {
    source.subscribe(observerB); // subject.subscribe(observerB)
}, 1000);

必需真的比及实行connect()后才会真的用subject定阅source，并最先送出元素，假如没有实行connect()observable是不会真正实行的。



var source = Rx.Observable.interval(1000)
             .do(x => console.log('send: ' + x))
             .multicast(new Rx.Subject()); // 无限的 observable 

var observerA = {
    next: value => console.log('A next: ' + value),
    error: error => console.log('A error: ' + error),
    complete: () => console.log('A complete!')
}

var observerB = {
    next: value => console.log('B next: ' + value),
    error: error => console.log('B error: ' + error),
    complete: () => console.log('B complete!')
}

var subscriptionA = source.subscribe(observerA);

var realSubscription = source.connect();

var subscriptionB;
setTimeout(() => {
    subscriptionB = source.subscribe(observerB);
}, 1000);

setTimeout(() => {
    subscriptionA.unsubscribe();
    subscriptionB.unsubscribe(); 
    // 虽然A,B退订，然则时候流照样继承实行
}, 5000);

setTimeout(() => {
    realSubscription.unsubscribe();
    // 这里才会真正的退订
}, 7000);

refCount

竖立一个只需有定阅就会自动connect 的observable


var source = Rx.Observable.interval(1000)
             .do(x => console.log('send: ' + x))
             .multicast(new Rx.Subject())
             .refCount();

var observerA = {
    next: value => console.log('A next: ' + value),
    error: error => console.log('A error: ' + error),
    complete: () => console.log('A complete!')
}

var observerB = {
    next: value => console.log('B next: ' + value),
    error: error => console.log('B error: ' + error),
    complete: () => console.log('B complete!')
}

var subscriptionA = source.subscribe(observerA); // 当source 一被observerA 定阅时(定阅数从0 变成1)，就会马上实行并发送元素


var subscriptionB;
setTimeout(() => {
    subscriptionB = source.subscribe(observerB);

}, 1000);

setTimeout(() => {
    subscriptionA.unsubscribe(); // 定阅减一    subscriptionB.unsubscribe(); // 定阅为0，住手发送
}, 5000);

publish

等价于 multicast(new Rx.Subject())


var source = Rx.Observable.interval(1000)
             .publish() 
             .refCount();
             
// var source = Rx.Observable.interval(1000)
//             .multicast(new Rx.Subject()) 
//             .refCount();

var source = Rx.Observable.interval(1000)
             .publishReplay(1) 
             .refCount();
             
// var source = Rx.Observable.interval(1000)
//             .multicast(new Rx.ReplaySubject(1)) 
//             .refCount();


var source = Rx.Observable.interval(1000)
             .publishBehavior(0) 
             .refCount();
             
// var source = Rx.Observable.interval(1000)
//             .multicast(new Rx.BehaviorSubject(0)) 
//             .refCount();

var source = Rx.Observable.interval(1000)
             .publishLast() 
             .refCount();
             
// var source = Rx.Observable.interval(1000)
//             .multicast(new Rx.AsyncSubject(1)) 
//             .refCount();

share

等价于 publish + refCount


var source = Rx.Observable.interval(1000)
             .share();
             
// var source = Rx.Observable.interval(1000)
//             .publish() 
//             .refCount();

// var source = Rx.Observable.interval(1000)
//             .multicast(new Rx.Subject()) 
//             .refCount();

Scheduler

Scheduler简介

Scheduler 掌握一个observable 的定阅什么时刻最先，以及发送元素什么时刻送达，主要由以下三个元素所构成


Scheduler 是一个对象构造。它晓得怎样依据优先级或其他规范来贮存并实行使命。
Scheduler 是一个实行环境。它意味着使命何时何地被实行，比方像是马上实行、在回调(callback)中实行、setTimeout 中实行、animation frame 中实行
Scheduler是一个假造时钟。它透过now()这个要领供应了时候的观点，我们可以让使命在特定的时候点被实行。


// Scheduler 会影响Observable 最先实行及元素送达的机遇

var observable = Rx.Observable.create(function (observer) {
    observer.next(1);
    observer.next(2);
    observer.next(3);
    observer.complete();
});

console.log('before subscribe');
observable.observeOn(Rx.Scheduler.async) // 设为 async
.subscribe({
    next: (value) => { console.log(value); },
    error: (err) => { console.log('Error: ' + err); },
    complete: () => { console.log('complete'); }
});
console.log('after subscribe');

项目中的RxJs

在项目中RxJs可以经由过程库的情势援用，也可以援用连系了框架的组合。

经由过程之前的进修，对RxJs有了肯定的相识。对我而言RxJS最好的运用场景就是庞杂的UI交互。

而且在进修RxJS的材估中，很多典范的Demo都是:

拖拽交互
Auto Complete
等等

应用RxJS能把我们之前须要写很多推断，很多逻辑的UI交互都简化了，经由过程它自带的一套Stream的用法，可以应用更少的代码完成之前的庞杂的工作量，供应了开辟效力。

RxJS同时能运用在组件状况治理中，可以参考Reactive 视角审阅 React 组件

在React中，内部经由过程setState治理状况。状况的变动可以依靠RxJS流,在须要的Response中setState即可。

其他计划可以自行依据项目需求到场，须要就引入，不须要就不要加，不要为RxJS而RxJS.

还要注重的是RxJS的操纵符异常壮大，然则数目极多，因而一最先开辟入门的时刻先设想好逻辑再去查文档。

官方的example有很多例子可以参考运用。

认识一下 redux-observable

redux-observable，则是经由过程建立epics中间件，为每一个dispatch增加响应的附加结果。相较于thunk中间件，运用redux-observable来处置惩罚异步action，有以下两个长处：

不须要修正reducer，我们的reducer可以依旧坚持简朴的纯函数形状。
epics中间件会将action封装成Observable对象，可以运用RxJs的响应api来掌握异步流程。

比起redux-thunk,redux-observable可以强有力的支撑更加庞杂的异步逻辑。用更少的代码来完成需求。

总结

经由过程几天的进修，对RxJS有了肯定的相识，以后就是将其运用到现实项目中。

材料

进修操纵符的时刻可以对比弹珠图

Rx Observables 的交互弹珠图

Learn RxJS 的中文版
redux-observable中文文档

    原文作者：linshuizhaoying
    原文地址: https://segmentfault.com/a/1190000013141856
    本文转自网络文章，转载此文章仅为分享知识，如有侵权，请联系博主进行删除。