摘要：技術(shù)積累經(jīng)過社區(qū)的努力學(xué)習(xí)資料還是很多的，官方中文文檔就已經(jīng)很不錯(cuò)，不過我們先從天精通初步感受一下然后配合一些中文文檔來補(bǔ)充知識(shí)點(diǎn)，最后再根據(jù)官方文檔來校驗(yàn)整個(gè)知識(shí)體系。資料學(xué)習(xí)操作符的時(shí)候可以對(duì)照彈珠圖的交互彈珠圖的中文版中文文檔

最近準(zhǔn)備畢設(shè)，技術(shù)選型的時(shí)候因?yàn)楣δ艿囊恍┬枨鬁?zhǔn)備將RxJs融入到項(xiàng)目中，考慮RxJs的時(shí)候因?yàn)橹暗募夹g(shù)棧還猶豫了一下，查了一些資料以及粗略瀏覽了一些文檔。感覺對(duì)于畢設(shè)項(xiàng)目RxJs的加入是有幫助的，因此打算系統(tǒng)的學(xué)習(xí)然后摘抄知識(shí)點(diǎn)以及實(shí)踐一些demo做技術(shù)積累。

RxJs經(jīng)過社區(qū)的努力學(xué)習(xí)資料還是很多的，官方中文文檔就已經(jīng)很不錯(cuò)，不過我們先從30 天精通 RxJS初步感受一下RxJS.然后配合一些中文文檔來補(bǔ)充知識(shí)點(diǎn)，最后再根據(jù)官方文檔來校驗(yàn)整個(gè)知識(shí)體系。

RxJS是一套由Observable sequences來組合異步行為和事件基礎(chǔ)程序的Library

RxJS 是Functional Programming 跟Reactive Programming 的結(jié)合

把每個(gè)運(yùn)算包成一個(gè)個(gè)不同的function，并用這些function 組合出我們要的結(jié)果，這就是最簡單的Functional Programming

Functional Programming 強(qiáng)調(diào)沒有Side Effect，也就是function 要保持純粹，只做運(yùn)算并返回一個(gè)值，沒有其他額外的行為。

Side Effect

Side Effect是指一個(gè)function做了跟本身運(yùn)算返回值沒有關(guān)系的事，比如說修改某個(gè)全域變數(shù)，或是修改傳入?yún)?shù)的值，甚至是執(zhí)行console.log都算是Side Effect。

前端常見的Side Effect:

發(fā)送http request

在畫面輸出值或是log

獲得用戶的input

Query DOM

Reactive Programming簡單來說就是當(dāng)變數(shù)或資源發(fā)生變動(dòng)時(shí)，由變數(shù)或資源自動(dòng)告訴我發(fā)生變動(dòng)了

Observer Pattern 其實(shí)很常遇到，許多API 的設(shè)計(jì)上都用了Observer Pattern，最簡單的例子就是DOM 物件的事件監(jiān)聽:

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

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

觀察者模式:我們可以對(duì)某件事注冊(cè)監(jiān)聽，并在事件發(fā)生時(shí)，自動(dòng)執(zhí)行我們注冊(cè)的監(jiān)聽者(listener)。

Es5版本:

function Producer() {
    
    // 這個(gè) if 只是避免使用者不小心把 Producer 當(dāng)做函數(shù)調(diào)用
    if(!(this instanceof Producer)) {
      throw new Error("請(qǐng)用 new Producer()!");
    }
    
    this.listeners = [];
}

// 加入監(jiān)聽的方法
Producer.prototype.addListener = function(listener) {
    if(typeof listener === "function") {
        this.listeners.push(listener)
    } else {
        throw new Error("listener 必須是 function")
    }
}

// 移除監(jiān)聽的方法
Producer.prototype.removeListener = function(listener) {
    this.listeners.splice(this.listeners.indexOf(listener), 1)
}

// 發(fā)送通知的方法
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);
        })
    }
}

調(diào)用例子:

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

JavaScript 到了ES6 才有原生的Iterator
在ECMAScript中Iterator最早其實(shí)是要采用類似Python的Iterator規(guī)范，就是Iterator在沒有元素之后，執(zhí)行next會(huì)直接拋出錯(cuò)誤；但后來經(jīng)過一段時(shí)間討論后，決定采更functional的做法，改成在取得最后一個(gè)元素之后執(zhí)行next永遠(yuǎn)都回傳{ 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 }

簡單實(shí)現(xiàn):

es5:

function IteratorFromArray(arr) {
    if(!(this instanceof IteratorFromArray)) {
        throw new Error("請(qǐng)用 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 };
    }
}

優(yōu)勢(shì)

Iterator的特性可以拿來做延遲運(yùn)算(Lazy evaluation)，讓我們能用它來處理大數(shù)組。

第二因?yàn)閕terator 本身是序列，所以可以第調(diào)用方法像map, filter... 等！

延遲運(yùn)算(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 }

把一個(gè)字串丟進(jìn)getNumbersh函數(shù)時(shí)，并沒有馬上運(yùn)算出字串中的所有數(shù)字，必須等到我們執(zhí)行next()時(shí)，才會(huì)真的做運(yùn)算，這就是所謂的延遲運(yùn)算(evaluation strategy)

Observer跟Iterator有個(gè)共通的特性，就是他們都是漸進(jìn)式 (progressive)的取得資料，差別只在于Observer是生產(chǎn)者(Producer)推送資料(push )，而Iterator是消費(fèi)者(Consumer)請(qǐng)求資料(pull)!

Observable其實(shí)就是這兩個(gè)Pattern思想的結(jié)合，Observable具備生產(chǎn)者推送資料的特性，同時(shí)能像序列，擁有序列處理資料的方法 (map, filter...)！

RxJS說白了就是一個(gè)核心三個(gè)重點(diǎn)。

一個(gè)核心是Observable 再加上相關(guān)的Operators(map, filter...)，這個(gè)部份是最重要的，其他三個(gè)重點(diǎn)本質(zhì)上也是圍繞著這個(gè)核心在轉(zhuǎn)，所以我們會(huì)花將近20 天的篇數(shù)講這個(gè)部份的觀念及使用案例。

另外三個(gè)重點(diǎn)分別是

Observer

Subject

Schedulers

Observable 同時(shí)可以處理同步與異步的行為！

同步操作

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!

Observable 可以被訂閱(subscribe)，或說可以被觀察，而訂閱Observable的又稱為觀察者(Observer)。
觀察者是一個(gè)具有三個(gè)方法(method)的對(duì)象，每當(dāng)Observable 發(fā)生事件時(shí)，便會(huì)呼叫觀察者相對(duì)應(yīng)的方法。

next：每當(dāng)Observable 發(fā)送出新的值，next 方法就會(huì)被呼叫。

complete：在Observable 沒有其他的資料可以取得時(shí)，complete 方法就會(huì)被呼叫，在complete 被呼叫之后，next 方法就不會(huì)再起作用。

error：每當(dāng)Observable 內(nèi)發(fā)生錯(cuò)誤時(shí)，error 方法就會(huì)被呼叫。

var observable = Rx.Observable
    .create(function(observer) {
            observer.next("Jerry");
            observer.next("Anna");
            observer.complete();
            observer.next("not work");
    })
    
// 定義一個(gè)觀察者
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執(zhí)行后，next就會(huì)自動(dòng)失效，所以沒有印出not work。

捕獲錯(cuò)誤實(shí)例：

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

觀察者可以是不完整的，他可以只具有一個(gè)next 方法

訂閱一個(gè)Observable 就像是執(zhí)行一個(gè)function

Operators 就是一個(gè)個(gè)被附加到Observable 型別的函數(shù)，例如像是map, filter, contactAll... 等等
每個(gè)operator都會(huì)回傳一個(gè)新的observable，而我們可以透過create的方法建立各種operator

Observable 有許多創(chuàng)建實(shí)例的方法，稱為creation operator。下面我們列出RxJS 常用的creation operator:

create
of
from
fromEvent
fromPromise
never
empty
throw
interval
timer

當(dāng)我們想要同步的傳遞幾個(gè)值時(shí)，就可以用of這個(gè)operator來簡潔的表達(dá)!

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來接收任何可枚舉的參數(shù)（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的第一個(gè)參數(shù)要傳入DOM ，第二個(gè)參數(shù)傳入要監(jiān)聽的事件名稱。上面的代碼會(huì)針對(duì)body 的click 事件做監(jiān)聽，每當(dāng)點(diǎn)擊body 就會(huì)印出event。

獲取 DOM 的常用方法：
document.getElementById()
document.querySelector()
document.getElementsByTagName()
document.getElementsByClassName()

Event來建立Observable實(shí)例還有另一個(gè)方法fromEventPattern，這個(gè)方法是給類事件使用

所謂的類事件就是指其行為跟事件相像，同時(shí)具有注冊(cè)監(jiān)聽及移除監(jiān)聽兩種行為，就像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?");

字?jǐn)?shù)受限，可以去博客看完整版

Subject 可以拿去訂閱Observable(source) 代表他是一個(gè)Observer，同時(shí)Subject 又可以被Observer(observerA, observerB) 訂閱，代表他是一個(gè)Observable。

Subject 同時(shí)是Observable 又是Observer
Subject 會(huì)對(duì)內(nèi)部的observers 清單進(jìn)行組播(multicast)

Subject 在內(nèi)部管理一份observer 的清單，并在接收到值時(shí)遍歷這份清單并送出值

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 就會(huì)接收到，又因?yàn)镾ubject 本身是Observable，所以這樣的使用方式很適合用在某些無法直接使用Observable 的前端框架中，例如在React 想對(duì)DOM 的事件做監(jiān)聽

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  this.subject.next(event)}>{this.state.count}
    }
}

BehaviorSubject

BehaviorSubject 是用來呈現(xiàn)當(dāng)前的值，而不是單純的發(fā)送事件。BehaviorSubject 會(huì)記住最新一次發(fā)送的元素，并把該元素當(dāng)作目前的值，在使用上BehaviorSubject 建構(gòu)式需要傳入一個(gè)參數(shù)來代表起始的狀態(tài)

// BehaviorSubject 在建立時(shí)就需要給定一個(gè)狀態(tài)，并在之后任何一次訂閱，就會(huì)先送出最新的狀態(tài)。其實(shí)這種行為就是一種狀態(tài)的表達(dá)而非單存的事件，就像是年齡跟生日一樣，年齡是一種狀態(tài)而生日就是事件；所以當(dāng)我們想要用一個(gè)stream 來表達(dá)年齡時(shí)，就應(yīng)該用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

在新訂閱時(shí)重新發(fā)送最后的幾個(gè)元素，這時(shí)我們就可以用ReplaySubject

var subject = new Rx.ReplaySubject(2); // 重復(fù)發(fā)送最后倆個(gè)元素
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結(jié)束后送出最后一個(gè)值

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)

multicast

multicast 可以用來掛載subject 并回傳一個(gè)可連結(jié)(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);

必須真的等到執(zhí)行connect()后才會(huì)真的用subject訂閱source，并開始送出元素，如果沒有執(zhí)行connect()observable是不會(huì)真正執(zhí)行的。

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退訂，但是時(shí)間流還是繼續(xù)執(zhí)行
}, 5000);

setTimeout(() => {
    realSubscription.unsubscribe();
    // 這里才會(huì)真正的退訂
}, 7000);

refCount

建立一個(gè)只要有訂閱就會(huì)自動(dòng)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); // 當(dāng)source 一被observerA 訂閱時(shí)(訂閱數(shù)從0 變成1)，就會(huì)立即執(zhí)行并發(fā)送元素


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

}, 1000);

setTimeout(() => {
    subscriptionA.unsubscribe(); // 訂閱減一    subscriptionB.unsubscribe(); // 訂閱為0，停止發(fā)送
}, 5000);

publish

等價(jià)于 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

等價(jià)于  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 控制一個(gè)observable 的訂閱什么時(shí)候開始，以及發(fā)送元素什么時(shí)候送達(dá)，主要由以下三個(gè)元素所組成

Scheduler 是一個(gè)對(duì)象結(jié)構(gòu)。它知道如何根據(jù)優(yōu)先級(jí)或其他標(biāo)準(zhǔn)來儲(chǔ)存并執(zhí)行任務(wù)。
Scheduler 是一個(gè)執(zhí)行環(huán)境。它意味著任務(wù)何時(shí)何地被執(zhí)行，比如像是立即執(zhí)行、在回調(diào)(callback)中執(zhí)行、setTimeout 中執(zhí)行、animation frame 中執(zhí)行
Scheduler是一個(gè)虛擬時(shí)鐘。它透過now()這個(gè)方法提供了時(shí)間的概念，我們可以讓任務(wù)在特定的時(shí)間點(diǎn)被執(zhí)行。

// Scheduler 會(huì)影響Observable 開始執(zhí)行及元素送達(dá)的時(shí)機(jī)

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) // 設(shè)為 async
.subscribe({
    next: (value) => { console.log(value); },
    error: (err) => { console.log("Error: " + err); },
    complete: () => { console.log("complete"); }
});
console.log("after subscribe");

在項(xiàng)目中RxJs可以通過庫的形式引用，也可以引用結(jié)合了框架的組合。

通過之前的學(xué)習(xí)，對(duì)RxJs有了一定的了解。對(duì)我而言RxJS最好的應(yīng)用場(chǎng)景就是復(fù)雜的UI交互。

而且在學(xué)習(xí)RxJS的資料中，很多典型的Demo都是:

拖拽交互

Auto Complete

等等

利用RxJS能把我們以前需要寫很多判斷，很多邏輯的UI交互都簡化了，通過它自帶的一套Stream的用法，可以利用更少的代碼完成以前的復(fù)雜的工作量，提供了開發(fā)效率。

RxJS同時(shí)能應(yīng)用在組件狀態(tài)管理中，可以參考Reactive 視角審視 React 組件

在React中，內(nèi)部通過setState管理狀態(tài)。狀態(tài)的變更可以依賴RxJS流,在需要的Response中setState即可。

其他方案可以自行根據(jù)項(xiàng)目需求加入，需要就引入，不需要就不要加，不要為RxJS而RxJS.

還要注意的是RxJS的操作符非常強(qiáng)大，但是數(shù)量極多，因此一開始開發(fā)入門的時(shí)候先設(shè)計(jì)好邏輯再去查文檔。

官方的example有很多例子可以參考應(yīng)用。

redux-observable，則是通過創(chuàng)建epics中間件，為每一個(gè)dispatch添加相應(yīng)的附加效果。相較于thunk中間件，使用redux-observable來處理異步action，有以下兩個(gè)優(yōu)點(diǎn)：

不需要修改reducer，我們的reducer可以依然保持簡單的純函數(shù)形態(tài)。
epics中間件會(huì)將action封裝成Observable對(duì)象，可以使用RxJs的相應(yīng)api來控制異步流程。

比起redux-thunk,redux-observable能夠強(qiáng)有力的支持更為復(fù)雜的異步邏輯。用更少的代碼來實(shí)現(xiàn)需求。

通過幾天的學(xué)習(xí)，對(duì)RxJS有了一定的了解，之后就是將其應(yīng)用到實(shí)際項(xiàng)目中。

學(xué)習(xí)操作符的時(shí)候可以對(duì)照彈珠圖

Rx Observables 的交互彈珠圖

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