學(xué)習(xí)資料

https://cn.rx.js.org/
https://github.com/RxJS-CN/rxjs-articles-translation
http://www.reibang.com/p/eaf28d5ce6c0

概念

Observable 和 Observer

Observable 可觀察者對(duì)象箕戳，負(fù)責(zé)發(fā)出數(shù)據(jù)步藕；Observer 觀察者肖方，負(fù)責(zé)接收數(shù)據(jù)
怎么這么抽象？試著把Observable對(duì)象理解為一種數(shù)據(jù)格式，類似于數(shù)組、鏈表等等
Observable實(shí)現(xiàn)了下?兩種設(shè)計(jì)模式：

• 觀察者模式（Observer Pattern）
• 迭代器模式（Iterator Pattern）

觀察者模式

定義：觀察者模式定義了對(duì)象之間的一對(duì)多依賴，這樣一來(lái)，當(dāng)一個(gè)對(duì)象改變狀態(tài)時(shí)敏簿，它的所有依賴著都會(huì)收到通知并自動(dòng)更新
觀察者模式對(duì)“治”這個(gè)問題提的解決?法是這樣，將邏輯分為發(fā)布者（Publisher）和觀察者（Observer）宣虾，其中發(fā)布者只管負(fù)責(zé)產(chǎn)?事件惯裕，它會(huì)通知所有注冊(cè)掛上號(hào)的觀察者，?不關(guān)?這些觀察者如何處理這些事件绣硝，相對(duì)的蜻势，觀察者可以被注冊(cè)上某個(gè)發(fā)布者，只管接收到事件之后就處理鹉胖，?不關(guān)?這些數(shù)據(jù)是如何產(chǎn)?的握玛。

發(fā)布者和觀察者的關(guān)系

// Subject
function Observable() {
  this.observers = [];
  this.data = 'data update';
  // 用于關(guān)聯(lián)observer
  this.addObserve = function(observe) {
    this.observers.push(observe)
  };
  // 用于取消關(guān)聯(lián)observer
  this.removeObserve = function(observe) {
    const index = this.observers.findIndex(item => item == observe);
    this.observers.splice(index,1);
  }

  this.notifyObservers = function () {
    for (let i=0; i<this.observers.length; i++) {
      this.observers[i].update(this.data);
    }
  }
    // 用于發(fā)送更新數(shù)據(jù)到observer
  this.updateData = function(data) {
    this.data = data;
    this.notifyObservers();
  }
}

// 所有 observer 都必須實(shí)現(xiàn)update方法
function Observer() {
  this.update = function (data) {
    console.log(data);
  }
}

const observable = new Observable();
const observer1 = new Observer();
const observer2 = new Observer();
// 建立關(guān)聯(lián)
observable.addObserve(observer1);
observable.addObserve(observer2);
observable.notifyObservers();
observable.updateData(123);

迭代器模式

定義：提供一種方法順序訪問一個(gè)聚合對(duì)象中的各個(gè)元素够傍，而又不暴露其內(nèi)部的表示
數(shù)據(jù)集合的實(shí)現(xiàn)?式很多，可以是?個(gè)數(shù)組挠铲，也可以是?個(gè)樹形結(jié)構(gòu)冕屯，也可以是?個(gè)單向鏈表……迭代器的作?就是提供?個(gè)通?的接?，讓使?者完全不?關(guān)?這個(gè)數(shù)據(jù)集合的具體實(shí)現(xiàn)?式

function Iterator(list) {
  this.list = list;
  this.porsition = 0;
  this.hasNext = function() {
    if (this.porsition > this.list.length || this.list[porsition] === null) return false;
    return true;
  }
  this.next = function() {
    const item = this.list[this.porsition];
    this.porsition++;
    return item;
  }
  this.isDone = function() {
    return this.porsition >= this.list.length;
  }
}

在使?RxJS的過程中絕對(duì)看不到類似這樣的代碼拂苹，實(shí)際上安聘，你都看不到上?所說的三個(gè)函數(shù)，因?yàn)槠鞍簦?所說的是“拉”式的迭代器實(shí)現(xiàn)浴韭，?RxJS實(shí)現(xiàn)的是“推”式的迭代器實(shí)現(xiàn)

Subscription

訂閱，表示建立關(guān)聯(lián)關(guān)系

理解Observable脯宿、Observe 和 Subscribtions 之間的關(guān)系

Observable 是信號(hào)源念颈、生產(chǎn)者，Observer 觀察者嗅绰、消費(fèi)者
Observable 和 Observer 之間的關(guān)系：
如果比作報(bào)社和讀者舍肠，報(bào)社是 Observable搀继，是數(shù)據(jù)源窘面，提供報(bào)紙，讀者是 Observer叽躯，負(fù)責(zé)消費(fèi)（處理）數(shù)據(jù)财边，閱讀報(bào)紙，讀者向報(bào)社訂閱（subscribe）報(bào)紙后点骑，報(bào)社將讀者列入他們派送名單酣难，定期派送報(bào)紙，報(bào)紙是數(shù)據(jù)黑滴。
可以比作前端（Observer）通過 登陸 owncloud 賬號(hào)（subscribe）時(shí)時(shí)獲取 UI（Observable）更新的 UI稿（數(shù)據(jù)）
也可以是愛奇藝會(huì)員（Observer）點(diǎn)擊播放愛奇藝視頻（subscribe）觀看愛奇藝網(wǎng)站（Observable）提供的視頻（數(shù)據(jù)）
Observable憨募、Observe 和 Subscribtions核心就是解決分工與數(shù)據(jù)傳遞。

subscribe 擴(kuò)展知識(shí)

可以通過add袁辈、remove 操作子subscription
父subscription取消訂閱菜谣，子subscription也會(huì)一起取消訂閱

var observ1 = Rx.Observable.interval(500)
var observ2 = Rx.Observable.interval(800)
var observ3 = Rx.Observable.interval(800)

var subsc1 = observ1.subscribe(x => console.log('first: ' + x))
var subsc2 = observ2.subscribe(x => console.log('second: ' + x))
var subsc3 = observ3.subscribe(x => console.log('three: ' + x))

subsc1.add(subsc2)
subsc1.add(subsc3)

setTimeout(() => {
  subsc1.remove(subsc2)
  subsc1.unsubscribe()
}, 1100)
// second: n將會(huì)一直執(zhí)行下去，我們添加了 subsc1.add(subsc2) , 在1.1S后移除了它晚缩，所以在 unsubscribe() 時(shí)尾膊，我們并沒有清除掉它

unsubscribe：釋放資源和取消Observable執(zhí)行的功能

Operators

操作符，類似于管道荞彼，對(duì)數(shù)據(jù)源發(fā)出的數(shù)據(jù)進(jìn)行過濾或其他處理冈敛，使數(shù)據(jù)源發(fā)出的數(shù)據(jù)更加滿足Observe 的需求

Subject

有一些場(chǎng)景，需要將Cold Observable 轉(zhuǎn)成 Hot Observable鸣皂，在這個(gè)場(chǎng)景下需要?個(gè)“中間?”做串接的事情抓谴，這個(gè)中間?有兩個(gè)職責(zé)：

• 中間?要提供subscribe?法暮蹂，讓其他?能夠訂閱??的數(shù)據(jù)源。
• 中間?要能夠有辦法接受推送的數(shù)據(jù)癌压，包括Cold Observable推送的數(shù)據(jù)椎侠。

上?所說的第?個(gè)職責(zé)，相當(dāng)于?個(gè)Observable措拇，第?個(gè)?作我纪，相當(dāng)于?個(gè)Observer。在RxJS中丐吓，提供了名為Subject的類型浅悉，?個(gè)Subject既有Observable的接口，也具有Observer的接口券犁，?個(gè)Subject就具備上述的兩個(gè)職責(zé)术健。

Cold Observable 和 Hot Observable的區(qū)別

好比視頻和電視頻道的區(qū)別，視頻沒有時(shí)間限制粘衬，任何時(shí)候想看都可以看荞估，電視有時(shí)間限制
視頻是 Cold Observable，電視是 Hot Observable
Cold Observable：你見或者不見稚新，我一直在勘伺，只要你愿意去取的話
Hot Observable：過了這個(gè)村，沒有這個(gè)店

var interval$ = Rx.Observable.interval(500);

// Cold Observable
interval$.map(val=>'a'+val).subscribe(x => console.log(x));
setTimeout(()=>{
  interval$.map(val=>'b'+val).subscribe(x => console.log(x));
}, 2000);

// Hot Observable
const subject$ = new Rx.Subject();
interval$.subscribe(subject$);
subject$.map(val=>'a'+val).subscribe(x => console.log(x));
setTimeout(() => {
  subject$.map(val=>'b'+val).subscribe(x => console.log(x));
}, 1500);

回過頭來(lái)講subject褂删，subject相當(dāng)于一個(gè)轉(zhuǎn)換器飞醉，它將 Cold Observable 轉(zhuǎn)化成 Hot Observable。這就要求subject同時(shí)是observe又是 observable屯阀。subject 既能訂閱數(shù)據(jù)源缅帘，同時(shí)本身又是數(shù)據(jù)源，能發(fā)出數(shù)據(jù)难衰。
舉個(gè)不十分恰當(dāng)?shù)睦忧瘴蓿热缡謾C(jī)，它可以接收基站信號(hào)(Observe)盖袭，同時(shí)也可以發(fā)出信號(hào)（Observable）

Schedulers

調(diào)度器
Scheduer是?種數(shù)據(jù)結(jié)構(gòu)失暂，可以根據(jù)優(yōu)先級(jí)或者其他某種條件來(lái)安排任務(wù)執(zhí)?隊(duì)列
http://www.reibang.com/p/5624c8a6bd2b

為什么要學(xué)

我們學(xué)習(xí)RxJS，并不是因?yàn)镽xJS是?項(xiàng)炫酷的技術(shù)苍凛，也不是因?yàn)镽xJS是?個(gè)最新的技術(shù)趣席，是因?yàn)镽xJS的的確確能夠幫助我們解決問題，?且這些問題長(zhǎng)期以來(lái)?直在困擾我們醇蝴，沒有好的解決辦法宣肚，這些問題包括：

• 如何控制?量代碼的復(fù)雜度；
• 如何保持代碼可讀悠栓；
• 如何處理異步操作霉涨。

Rxjs 引用了兩個(gè)重要的編程思想按价，讓代碼更加清爽，更加容易維護(hù)：
函數(shù)式
響應(yīng)式

函數(shù)式

• 聲明式

聲明式區(qū)別于命令式笙瑟，命令式強(qiáng)調(diào)的是告訴機(jī)器怎么去做（how）楼镐，一步步的告訴計(jì)算機(jī)如何完成一項(xiàng)工作
聲明式強(qiáng)調(diào)的是告訴機(jī)器你想要什么（what），不關(guān)注內(nèi)部實(shí)現(xiàn)往枷，聲明式把通用的共性抽離出來(lái)框产，避免重復(fù)代碼
應(yīng)用聲明式的困難點(diǎn)：歸納和提取完備的what，是件很困難错洁、很技術(shù)化的工作秉宿，令人望而卻步聲明式能夠應(yīng)用在特定領(lǐng)域如SQL中，是工具的編寫者屯碴，已經(jīng)歸納和提取what描睦，替你完成了

// 命令式
// how：寫for循環(huán)一一處理
function double(arr) {
  const results = []
  for (let i = 0; i < arr.length; i++){
  results.push(arr[i] * 2)
  }
  return results
}
function addOne(arr) {
  const results = []
  for (let i = 0; i < arr.length; i++){
  results.push(arr[i] + 1)
  }
  return results
}
// 聲明式
// what：通過map把每一項(xiàng)加倍或+1，不關(guān)注內(nèi)部實(shí)現(xiàn)
function double(arr) {
  return arr.map(function(item) {return item * 2});
} function addOne(arr) {
  return arr.map(function(item) {return item + 1});
}

• 純函數(shù)：

函數(shù)的執(zhí)?過程完全由輸?參數(shù)決定导而，不會(huì)受除參數(shù)之外的任何數(shù)據(jù)影響忱叭，只要入?yún)⒉蛔儯祷氐膮?shù)也不會(huì)變
函數(shù)不會(huì)修改任何外部狀態(tài)今艺，?如修改全局變量或傳?的參數(shù)對(duì)象
純函數(shù)沒有副作用韵丑，是穩(wěn)定的，可以和其他純函數(shù)像搭積木一樣一起組合使用洼滚，獲得更強(qiáng)的處理能力

• 數(shù)據(jù)不可變性

有數(shù)據(jù)埂息，替換?法是通過產(chǎn)?新的數(shù)據(jù)，來(lái)實(shí)現(xiàn)這種"變化"遥巴，也就是說，當(dāng)我們需要數(shù)據(jù)狀態(tài)發(fā)?改變時(shí)享幽，保持原有數(shù)據(jù)不變铲掐，產(chǎn)??個(gè)新的數(shù)據(jù)來(lái)體現(xiàn)這種變化

JavaScript中數(shù)組的push、pop值桩、sort函數(shù)都會(huì)改變?個(gè)數(shù)組的內(nèi)容摆霉，由此引發(fā)的bug可不少。這些不純的函數(shù)導(dǎo)致JavaScript天?不是?個(gè)純粹意義上的函數(shù)式編程語(yǔ)?

響應(yīng)式

EXCEL 中的公式就是典型的響應(yīng)式奔坟，數(shù)據(jù)改變了公式計(jì)算結(jié)果也會(huì)跟著變
類似于MVVM中的M->V

體驗(yàn)兩個(gè)小例子：

測(cè)試鼠標(biāo)按住時(shí)間

const buttonDom = document.querySelector('#button');
const mouseDown$ = Rx.Observable.fromEvent(buttonDom, 'mousedown');
const mouseUp$ = Rx.Observable.fromEvent(buttonDom, 'mouseup');
const holdTime$ = mouseUp$.timestamp().withLatestFrom(mouseDown$.timestamp(), (mouseUpEvent, mouseDownEvent)=>{
    return mouseUpEvent.timestamp - mouseDownEvent.timestamp;
});
holdTime$.subscribe((ms)=>{
    document.querySelector('#holdTime').innerText = ms;
});

takeUntil携栋，統(tǒng)計(jì)5秒內(nèi)用戶點(diǎn)擊數(shù)

const click$ = Rx.Observable.fromEvent(document, 'click');
click$.bufferWhen(()=>Rx.Observable.interval(5000)).subscribe(arr=>console.log(arr.length))

彈珠圖

彈珠圖可以用來(lái)表示數(shù)據(jù)流，例如:

--a---b-c---d---X---|->
a, b, c, d 表示發(fā)出的數(shù)據(jù)
X 表示錯(cuò)誤
|表示 '結(jié)束' 信號(hào)
---> 是時(shí)間軸

彈珠圖在線演示：https://rxviz.com/

操作符

創(chuàng)建

from 和 toArray
from：數(shù)組轉(zhuǎn) Observable
toArray：Observable 轉(zhuǎn)數(shù)組
fromPromise 和 toPromise
fromPromise：promise轉(zhuǎn) Observable
toPromise：Observable 轉(zhuǎn)promise

合并

對(duì)of產(chǎn)生的數(shù)據(jù)進(jìn)行concat和merge操作哦產(chǎn)生的不同結(jié)果

例1：merge：事件的合并處理

startWith和concat的關(guān)聯(lián)關(guān)系
zip
拉鏈婉支，一對(duì)一咬合

QQ截圖20190417140650.png

QQ圖片20190417141214.png

例2：zip應(yīng)用

• 讓of產(chǎn)生的數(shù)據(jù)流交叉輸出
• 實(shí)現(xiàn)異步隊(duì)列

forkJoin
forkJoin就是RxJS界的Promise.all，Promise.all等待所有輸?的Promise對(duì)象成功之后把結(jié)果合并澜建，forkJoin等待所有輸?的Observable對(duì)象完結(jié)之后把最后?個(gè)數(shù)據(jù)合并

const testList = [
  this.httpService.post(REQUEST_URL.editCourseTestListInfo, Object.assign({testType:1},params)),
  this.httpService.post(REQUEST_URL.editCourseTestListInfo, Object.assign({testType:2},params)),
];
Observable.create(observer => {
  forkJoin(testList).subscribe((data)=>{
    observer.next(data);
  });
})subscribe((res)=>{
  if(res.some(data=>data==false)) return;
  // 1-入門測(cè)成績(jī), 2-出門測(cè)成績(jī)
  this.scoreTestType = {
    1: { scoreTestDetail: res[0] },
    2: { scoreTestDetail: res[1] }
  };
})

輔助

數(shù)學(xué)類操作符有四個(gè)：count向挖、max蝌以、min、reduce
遍歷上游Observable對(duì)象中吐出的所有數(shù)據(jù)才給下游傳遞數(shù)據(jù)何之、只有在上游完結(jié)的時(shí)候跟畅，才給下游傳遞唯?數(shù)據(jù)

過濾

takeUntil讓我們可以?Observable對(duì)象作為notifier來(lái)控制另?個(gè)Observable對(duì)象的數(shù)據(jù)產(chǎn)?徊件，使用起來(lái)非常靈活
有損回壓控制：throttle、debounce蒜危、audit庇忌、sample、throttleTime舰褪、debounceTime皆疹、auditTime、sampleTime
對(duì)比：http://www.reibang.com/p/a176d28c9eb5

例3：防抖和節(jié)流

debounce占拍，去抖動(dòng)略就。策略是當(dāng)事件被觸發(fā)時(shí)，設(shè)定一個(gè)周期延遲執(zhí)行動(dòng)作晃酒，若期間又被觸發(fā)表牢，則重新設(shè)定周期，直到周期結(jié)束贝次，執(zhí)行動(dòng)作崔兴。 這是debounce的基本思想，在后期又?jǐn)U展了前緣debounce蛔翅，即執(zhí)行動(dòng)作在前敲茄，然后設(shè)定周期，周期內(nèi)有事件被觸發(fā)山析，不執(zhí)行動(dòng)作堰燎，且周期重新設(shè)定。

// 暴力版
var debounce = (fn, wait) => {
    let timer, timeStamp=0;
    let context, args;
 
    let run = ()=>{
        timer= setTimeout(()=>{
            fn.apply(context,args);
        },wait);
    }
    let clean = () => {
        clearTimeout(timer);
    }
 
    return function(){
        context=this;
        args = arguments;
        let now = (new Date()).getTime();
 
        if(now - timeStamp < wait){
            console.log('reset',now);
            clean();  // clear running timer 
            run();    // reset new timer from current time
        } else{
            console.log('set',now);
            run();    // last timer alreay executed, set a new timer
        }
        timeStamp = now;
    }
}

// rxls
let foo$ = Rx.Observable.fromEvent(document, 'click');
foo$.debounceTime(2000).subscribe(
  console.log,
  null,
  () => console.log('complete')
);

throttling笋轨，節(jié)流的策略是秆剪，固定周期內(nèi)，只執(zhí)行一次動(dòng)作爵政，若有新事件觸發(fā)仅讽，不執(zhí)行。周期結(jié)束后钾挟，又有事件觸發(fā)洁灵，開始新的周期。 節(jié)流策略也分前緣和延遲兩種等龙。與debounce類似处渣，延遲是指 周期結(jié)束后執(zhí)行動(dòng)作伶贰，前緣是指執(zhí)行動(dòng)作后再開始周期。
throttling的特點(diǎn)在連續(xù)高頻觸發(fā)事件時(shí)罐栈，動(dòng)作會(huì)被定期執(zhí)行黍衙，響應(yīng)平滑。

// 簡(jiǎn)單版： 定時(shí)器期間荠诬，只執(zhí)行最后一次操作
var throttling = (fn, wait) => {
    let timer;
    let context, args;
 
    let run = () => {
        timer=setTimeout(()=>{
            fn.apply(context,args);
            clearTimeout(timer);
            timer=null;
        },wait);
    }
 
    return function () {
        context=this;
        args=arguments;
        if(!timer){
            console.log("throttle, set");
            run();
        }else{
            console.log("throttle, ignore");
        }
    }
}

// rxjs
let foo$ = Rx.Observable.fromEvent(document, 'click');
foo$.throttleTime(2000).subscribe(
  console.log,
  null,
  () => console.log('complete')
);

轉(zhuǎn)化

scan可能是RxJS中對(duì)構(gòu)建交互式應(yīng)?程序最重要的?個(gè)操作符有决，因?yàn)樗軌蚓S持應(yīng)?的當(dāng)前狀態(tài)闸拿，???可以根據(jù)數(shù)據(jù)流持續(xù)更新這些狀態(tài)，另???可以持續(xù)把更新的狀態(tài)傳給另?個(gè)數(shù)據(jù)流?來(lái)做必要處理书幕。
定義：public scan(accumulator: function(acc: R, value: T, index: number): R, seed: T | R): Observable對(duì)源 Observable 使用累加器函數(shù)新荤， 返回生成的中間值， 可選的初始值index 是賦給 acc 的初始值

let foo$ = Rx.Observable.interval(1000);
// acc 為上次返回值
// cur 更新的值台汇，此處由foo$提供
foo$.scan((acc, cur) => {
    return cur
}, 0).subscribe((data)=>console.log(data));

acc 是上一個(gè) scan 的返回值
subscript data 顯示的是當(dāng)前值

scan 和 reduce 的區(qū)別

reduce需要數(shù)據(jù)結(jié)束才能輸出結(jié)果
scan可以輸出中間狀態(tài)

無(wú)損回壓控制

數(shù)據(jù)組合成數(shù)組：bufferTime苛骨、bufferCount、bufferWhen苟呐、bufferToggle痒芝、buffer
數(shù)據(jù)組合成Observable：windowTime、windowCount掠抬、windowToggle 和window

bufferCount
支持兩個(gè)參數(shù) bufferSize 和 startBufferEvery
bufferSize 表示 緩存區(qū)長(zhǎng)度吼野，緩存區(qū)長(zhǎng)度達(dá)到bufferSize的時(shí)候傳新的數(shù)據(jù)給下游
startBufferEvery 可選，表示 新的緩存區(qū)長(zhǎng)度两波，即新數(shù)據(jù)個(gè)數(shù)，從上次bufferCount觸發(fā)以后闷哆，上游每發(fā)出startBufferEvery個(gè)數(shù)據(jù)后向下游傳出數(shù)據(jù)腰奋，數(shù)組中舊數(shù)據(jù)個(gè)數(shù)為bufferSize- startBufferEvery
如果不填startBufferEvery，則默認(rèn)值為 bufferSize抱怔，都是新數(shù)據(jù)
如果startBufferEvery大于bufferSize劣坊，則會(huì)丟失startBufferEvery-bufferSize個(gè)數(shù)據(jù)

例4： 判斷連續(xù)輸入是否正確

召喚隱藏英雄

const code = [
  "ArrowUp",
  "ArrowUp",
  "ArrowDown",
  "ArrowDown",
  "ArrowLeft",
  "ArrowRight",
  "ArrowLeft",
  "ArrowRight",
  "KeyB",
  "KeyA",
  "KeyB",
  "KeyA"
]

Rx.Observable.fromEvent(document, 'keyup')
  .map(e => e.code)
  .bufferCount(12, 1)
  .subscribe(last12key => {
    if (_.isEqual(last12key, code)) {
      console.log('隱藏的彩蛋 \(^o^)/~')
    }
  });

bufferToggle
利?Observable來(lái)控制緩沖窗口的開和關(guān)
有兩個(gè)參數(shù)openings 和 closingSelector，openings 是一個(gè)Observable,控制每個(gè)緩沖窗口的開始時(shí)間屈留，closingSelector是一個(gè)返回Observable的函數(shù)（這樣能夠靈活控制取值范圍）,控制每個(gè)緩沖窗口的結(jié)束時(shí)間（相對(duì)于開始時(shí)間而言）

clipboard.png

 const code = [
   "ArrowUp",
   "ArrowUp",
   "ArrowDown",
   "ArrowDown",
   "ArrowLeft",
   "ArrowRight",
   "ArrowLeft",
   "ArrowRight",
   "KeyB",
   "KeyA",
   "KeyB",
   "KeyA"
 ]

Rx.Observable.fromEvent(document, 'keyup')
.map(e => e.code)
.bufferToggle(Rx.Observable.timer(0, 3000), i=>Rx.Observable.interval(3000))
.subscribe(last12key => {
console.log(last12key);
if (_.isEqual(last12key, code)) {
console.log('隱藏的彩蛋 \(^o^)/~')
}
});

高階Observable

QQ圖片20190417115019.png

高階Observable和一階Observable的關(guān)系
正如二維數(shù)組和一維數(shù)組的關(guān)系

相關(guān)的操作符

打平：concatAll局冰、mergeAll测蘑、zipAll、combineAll康二、forkJoin碳胳、switch、exhaust
組合：windowTime沫勿、windowCount挨约、windowToggle 和window、groupBy分組
(前四個(gè)是按順序分組产雹，最后一個(gè)打亂了順序)
轉(zhuǎn)化：concatMap诫惭、mergeMap（flatMap）、switchMap蔓挖、exhaustMap夕土、mergeScan
cancatMap=一對(duì)多的map+concatAll
映射：concatMapTo、mergeMapTo瘟判、switchMapTo
生成高階函數(shù)

const ho$ = Rx.Observable.interval(1000)
  .take(2)
  .concat(Rx.Observable.never())  // 添加了一個(gè)never數(shù)據(jù)流
  .map(x => Rx.Observable.interval(1500).map(y => x+':'+y).take(3));
ho$.subscribe(
  console.log,
  null,
  () => console.log('complete')
);

// 打平
ho$.zipAll().subscribe(
  console.log,
  null,
  () => console.log('complete')
);

例4：拖拽

const box = document.querySelector('.box');
const mousedown$ = Rx.Observable.fromEvent(box, 'mousedown');
const mousemove$ = Rx.Observable.fromEvent(box, 'mousemove');
const mouseup$ = Rx.Observable.fromEvent(box, 'mouseup');
const mouseout$ = Rx.Observable.fromEvent(box, 'mouseout$');

mousedown$.mergeMap((md) => {
  const stop$ = mouseup$.merge(mouseout$);
  return mousemove$.takeUntil(stop$).map((mm) =>{
    return {
      target: md.target,
      x: mm.clientX - md.offsetX,
      y: mm.clientY - md.offsetY
    }
  });
}).subscribe((obj) => {
  console.log(obj);
  obj.target.style.top = obj.y + 'px';
  obj.target.style.left = obj.x + 'px';
});

綜上：在RxJS中怨绣，創(chuàng)建類操作符是數(shù)據(jù)流的源頭，其余所有操作符最重要的三類就是合并類荒适、過濾類和轉(zhuǎn)化類梨熙。不夸張地說，使?RxJS解決問題絕大部分時(shí)間就是在使?這三種操作符

多播

Observable和Observer的關(guān)系刀诬，就是前者在播放內(nèi)容咽扇，后者在收聽內(nèi)容。播放內(nèi)容的?式分為三種：

• 單播（unicast）：微信發(fā)給朋友陕壹，只有一個(gè)接收者
• ?播（broadcast）：朋友圈廣告质欲，所有人都能看得見
• 多播（multicast）：群聊天，發(fā)給一群人糠馆，只有選中的朋友才能看見

clipboard.png

前面的例??都是單播
RxJS是?持?個(gè)Observable被多次subscribe的嘶伟，所以，RxJS?持多播又碌，但是九昧，表?上看到的是多播，實(shí)質(zhì)上還是單播

const tick$ = Rx.Observable.interval(1000).take(3);

tick$.subscribe(value => console.log('observer 1: ' + value));

setTimeout(() => {
    tick$.subscribe(value => console.log('observer 2: ' + value));
}, 2000);

第?個(gè)Observer依然接收到了0毕匀、1铸鹰、2總共三個(gè)數(shù)據(jù)。為什么會(huì)是這樣的結(jié)果皂岔？因?yàn)閕nterval這個(gè)操作符產(chǎn)?的是?個(gè)Cold Observable對(duì)象蹋笼。
Cold Observable，就是每次被subscribe都產(chǎn)??個(gè)全新的數(shù)據(jù)序列的數(shù)據(jù)流,例如對(duì)interval產(chǎn)?的Observable對(duì)象每subscribe?次，都會(huì)產(chǎn)??個(gè)全新的遞增整數(shù)序列剖毯，從0開始產(chǎn)?Hot Observable：fromPromise圾笨、fromEvent、fromEventPattern就是異步的創(chuàng)建操作符真正的多播逊谋，必定是?論有多少Observer來(lái)subscribe擂达，推給Observer的都是?樣的數(shù)據(jù)源把Cold Observable變成Hot Observable，用的是Subject

Subject

clipboard.png

var interval$ = Rx.Observable.interval(500);

const subject$ = new Rx.Subject();
    interval$.subscribe(subject$);
    subject$.map(val=>'a'+val).subscribe(x => console.log(x));
    setTimeout(() => {
      subject$.map(val=>'b'+val).subscribe(x => console.log(x));
    }, 1500);

Subject不能重復(fù)使?
Subject可以有多個(gè)上游

例5：scan管理react狀態(tài)

class Counter extends React.Component {
  state = {count: 0}
  onIncrement() {
    this.setState({count: this.state.count + 1});
  }
  onDecrement() {
    this.setState({count: this.state.count - 1});
  }
  render() {
    return (
      <CounterView
        count={this.state.count}
        onIncrement={this.onIncrement.bind(this)}
        onDecrement={this.onDecrement.bind(this)}
      />
    );
  }
}
export default Counter;

// subject作為橋梁進(jìn)行狀態(tài)維護(hù)
class RxCounter extends React.Component {
  constructor() {
    super(...arguments);
    this.state = {count: 0};
    this.counter = new Subject();
    const observer = value => this.setState({count: value});
    this.counter.scan((result, inc) => result + inc, 0)
    .subscribe(observer);
  }
  render() {
    return <CounterView
      count={this.state.count}
      onIncrement={()=> this.counter.next(1)}
      onDecrement={()=> this.counter.next(-1)}
    />
  }
}
export default RxCounter;

例6：買房放租

const house$ = new Rx.Subject();
const housecount$ = house$.scan((has, one) => has = has+one, 0).startWith(0);

const month$ = Rx.Observable.interval(1000);
const salary$ = month$.mapTo(1);
const rent$ = month$.withLatestFrom(housecount$).map(arr=>arr[1]*0.5);

// 月收入累加
const income$ = salary$.merge(rent$);

const cash$ = income$.scan((has, one)=>{
  has = has + one;
  if (has >= 100) {
    has -= 100;
    console.log('買房啦');
    house$.next(1);
  }
  return has;
}, 0)

cash$.subscribe(
  (data)=>{
    console.log('進(jìn)賬,余額:',data)
  },
  null,
  ()=>{
    console.log('complete');
  }
)