摘要：深入淺出讀書筆記遺留問題的與對(duì)應(yīng)的實(shí)際場(chǎng)景，以及在編碼中的體現(xiàn)部分測(cè)試你對(duì)時(shí)間的感覺按住我一秒鐘然后松手你的時(shí)間毫秒實(shí)現(xiàn)重置避免直接觸發(fā)事件，例如在處點(diǎn)擊然后在處實(shí)現(xiàn)獲取間隔時(shí)間你超過了的用戶的使用主要用來加載靜態(tài)資源，所

《深入淺出RxJS》讀書筆記

Observable的HOT與COLD對(duì)應(yīng)的實(shí)際場(chǎng)景，以及在編碼中的體現(xiàn)

html部分

  測(cè)試你對(duì)時(shí)間的感覺
  按住我一秒鐘然后松手
  你的時(shí)間：毫秒
  

jquery實(shí)現(xiàn)

var time = new Date().getTime();
$("#hold-me")
    .mousedown(function(event) {
        time = new Date().getTime();
    })
    .mouseup(function(event) {
        if (time) {
            var elapse = new Date().getTime() - time;
            $("#hold-time").text(elapse);
            // 重置time 避免直接觸發(fā)mouseup事件，例如在A處點(diǎn)擊然后在B處up
            time = null;
        }
    });

RxJS實(shí)現(xiàn)

const holdMeButton = document.getElementById("hold-me");
            const mouseDown$ = Rx.Observable.fromEvent(holdMeButton,"mousedown");
            const mouseUp$ = Rx.Observable.fromEvent(holdMeButton,"mouseup");
            // 獲取間隔時(shí)間
            const holdTime$ = mouseUp$.timestamp().withLatestFrom(mouseDown$.timestamp(),(mouseUpEvent,mouseDownEvent)=>{
                return mouseUpEvent.timestamp - mouseDownEvent.timestamp
            });

            holdTime$.subscribe(ms=>{
                document.getElementById("hold-time").innerText = ms;
            })

            holdTime$.flatMap(ms=>{
                return Rx.Observable.ajax("https://timing-sense-score-board.herokuapp.com/score/"+ms)
            })
            .map(e=>e.response)
            .subscribe(res=>{
                document.getElementById("rank").innerText = `你超過了${res.rank}% 的用戶`
            })

主要用來加載靜態(tài)資源，所以使用到了 koa,koa-static

const path = require("path");
const koa = require("koa");
const serve = require("koa-static");
const app = new koa();

app.use(async function (ctx,next) {
    console.log("收到請(qǐng)求...")
    await next()
    console.log(`"${ctx.path}"請(qǐng)求 已處理...`)
})

app.use(serve(path.resolve(__dirname, "../src"))).listen(3001,function(err){
    if(err) throw err;
    console.log("程序啟動(dòng)成功")
});

Observable 可被觀察的對(duì)象，Observer觀察者，Observer通過subscribe來觀察Observable對(duì)象

RxJS的數(shù)據(jù)流就是Observable對(duì)象：

觀察者模式

迭代器模式

// 使用 deep-link方式引入函數(shù)
const Observable = require("rxjs").Observable;

/*
 * 定義Observable對(duì)象的行為，會(huì)產(chǎn)生數(shù)據(jù)，調(diào)用訂閱者的next方法
 * 1. 此處的Observer與訂閱者行為 theObserver并不是同一個(gè)對(duì)象，而是對(duì)theObserver的包裝
 * 2. 如果observer.error被調(diào)用，之后的complete或者next就不會(huì)被調(diào)用啦，同理，complete被調(diào)用之后，也不會(huì)
 *    再調(diào)用next或者error
 * 3. 如果error或者complete一直未調(diào)用，則observer就一直在內(nèi)存中等待被調(diào)用
*/
const onSubscribe = observer =>{
    observer.next(1);
    observer.error(2);
    observer.complete(3);
}
// 產(chǎn)生一個(gè)Observable對(duì)象
const source$ = new Observable(onSubscribe);
// 定義觀察者的行為 消費(fèi)Observable對(duì)象產(chǎn)生的數(shù)據(jù)
const theObserver = {
    next:item => console.log(item),
    error:item => console.error(item),
    complete:item => console.log("已完成"),
}
// 建立Observable與Observer的關(guān)系
source$.subscribe(theObserver)

在訂閱一段事件之后observer不再響應(yīng)吐出的信息了，這時(shí)可以退訂，但是Observeable還會(huì)一直產(chǎn)生數(shù)據(jù)

const Observable = require("rxjs").Observable;

const onSubscribe = observer =>{
    let n = 1;
    const handle = setInterval(()=>{
        console.log(`in onSubscribe ${n}`)
        // if(n>3){
        //     observer.complete()
        // }
        observer.next(n++);
    },1000)
    return {
        unsubscribe(){
            // clearInterval(handle)
        }
    }
}

const source$ = new Observable(onSubscribe);

const theObserver = {
    next:item => console.log(item)
}

let subscription = source$.subscribe(theObserver)

setTimeout(()=>{
      // 此處的unsubscribe也是封裝過的
    subscription.unsubscribe()
},3500)

在node中執(zhí)行，會(huì)一直打印 in onSubscribe *，但是source$不會(huì)再響應(yīng)

const Observable = require("rxjs/Observable").Observable;
const of = require("rxjs/observable/of").of;
const map = require("rxjs/operator/map").map;
// 新建一個(gè)操作符
// 此處this是外部變量，導(dǎo)致此operator不再是純函數(shù)
Observable.prototype.double = function(){
    // return this::map(x=>x*2)
    return map.call(this,x=>x*2)
}

const source$ = of(1,3,4);
const result$ = source$.double();

result$.subscribe(value=>console.log(value))

解決需要使用call或者bind改變this的操作，這樣是依賴外部環(huán)境的，不屬于純函數(shù)，也會(huì)喪失TS的類型檢查優(yōu)勢(shì)

lettable將Observable對(duì)象傳遞給下文，避免使用this

const Observable = require("rxjs/Observable").Observable;
require("rxjs/add/observable/of").of;
require("rxjs/add/operator/map").map;
require("rxjs/add/operator/let").let;

const source$ = Observable.of(1,2,3);
const double$ = obs$ => obs$.map(v=>v*2);
// 接受上文，傳遞到下文
const result$ = source$.let(double$);

result$.subscribe(console.log)

  不引入`map`補(bǔ)丁,開發(fā)**lettable**寫法的操作符

// ES5實(shí)現(xiàn)
function map(project){

return function(obj$){
  // 通過上面的Observable生成一個(gè)新Observable
  return new Observable(observer=>{
    return obj$.subscribe({
      next:value=>observer.next(project(value)),
      error:err=>observer.error(err),
      complete:()=>observer.complete()
    })
  })
}

}
// 添加操作符
var result$ = source$.let(map(x => x * 3));

// ES6實(shí)現(xiàn)
const map6 = fn => obj$ =>

  new Observable(observer =>
      obj$.subscribe({
          next: value => observer.next(fn(value)),
          error: err => observer.error(err),
          complete: () => observer.complete()
      })
  );

// 添加操作符
var result$ = source$.let(map6(x => x * 4));

`pipeable`是`lettable`的別稱，方便對(duì)于`lattable`的理解,V6以上才支持

## Chapter4 創(chuàng)建數(shù)據(jù)流

> 大多數(shù)的操作符是靜態(tài)操作符

### 基礎(chǔ)操作符

1.  `create`簡單的返回一個(gè)Observable對(duì)象
    

Observable.create = function(subscribe){
  return new Observable(subscribe)
}
```

of列舉數(shù)據(jù)

import {Observable} from "rxjs/Observable";
import "rxjs/add/observable/of"
// 依次吐出數(shù)據(jù)，一次性emit
const source$ = Observable.of(1,2,3)；
// 訂閱
// 第一個(gè)參數(shù)是next，第二個(gè)參數(shù)是error回調(diào)，第三個(gè)參數(shù)是complete回調(diào)
source$.subscribe(console.log,null,()=>{console.log("Complete")})

range產(chǎn)生指定范圍的數(shù)據(jù)

const sourc$ = Observable.range(/*初始值*/1,/*個(gè)數(shù)*/100);
// 每次只能步進(jìn) 1

generate循環(huán)創(chuàng)建

相當(dāng)于for循環(huán)

const source$ = Observable.generate(
  // 初始值
  2,
  // 判斷條件
  value=> value < 10,
  // 步進(jìn)
  value=> value+0.5,
  // 函數(shù)體,產(chǎn)生的結(jié)果
  value=> value*value
)

使用generate代替range

const range = function(min,count){
  const max = min + count;
  return Observable.generate(min,v=>vv+1,v=>v*v)
}

repeat重復(fù)數(shù)據(jù)的數(shù)據(jù)流

實(shí)例操作符，通過import "rxjs/add/operator/repeat"引入

V4版本中repeat是靜態(tài)屬性，這樣在使用Observable.repeat(1,2)重復(fù)1兩次，這樣數(shù)據(jù)就夠靈活
V5版本中改為實(shí)例屬性之后，Observable.of(1,2,4).repeat(2),將產(chǎn)生的1,2,3重復(fù)兩次，功能更加強(qiáng)大

const Observable = require("rxjs").Observable;
require("rxjs/add/operator/repeat");

const source$ = Observable.create(observer => {
    setTimeout(() => {
        observer.next(1);
    }, 1000);
    setTimeout(() => {
        observer.next(2);
    }, 2000);
    setTimeout(() => {
        observer.next(3);
    }, 3000);
    setTimeout(() => {
        observer.complete(1);
    }, 4000);
    return {
        unsubscribe(){
            console.log("on Unsubscribe")
        }
    }
});

const repeat$ = source$.repeat(2)

repeat$.subscribe(console.log,null,()=>{console.log("Complete")})

// 1
// 2
// 3
// on Unsubscribe
// 1
// 2
// 3
// Complete
// on Unsubscribe

如果沒有observer.complete()repeat不會(huì)被調(diào)用

repeat以complete為契機(jī)會(huì)再次執(zhí)行數(shù)據(jù)源，如果上游一直沒有complete下游就不會(huì)執(zhí)行

因?yàn)?b>repeat的存在，第一次數(shù)據(jù)源執(zhí)行完(以complete為契機(jī))后并不會(huì)執(zhí)行observer的complete回調(diào)

empty,throw,never

interval和timer

interval類似于setInterval

require("rxjs/add/observable/interval")
// 每隔1000ms產(chǎn)生一個(gè)數(shù)據(jù)，初始值為0，步進(jìn)為1
Observable.interval(1000)"

timer 是setTimeout的超集

// 1000ms后開始產(chǎn)生數(shù)據(jù),之后每隔1000ms產(chǎn)生一個(gè)數(shù)據(jù),功能相當(dāng)于interval
Observable.timer(1000,1000)
// 指定日期
Observable.time(new Date(new Date().getTime() + 12000))

from 把一切轉(zhuǎn)化為Observable

將所有的Iterable的對(duì)象都轉(zhuǎn)化為Observable對(duì)象
可以將Promise對(duì)象轉(zhuǎn)化為Observable對(duì)象，功能與fromPromise相同

fromPromise異步處理的對(duì)接

const Observable = require("rxjs").Observable;
require("rxjs/add/observable/fromPromise");

const promise = Promise.resolve(123);
Observable.fromPromise(promise).subscribe(console.log, null, () =>
    console.log("Complete")
);
//123
//Complete
const promise1 = Promise.reject("error");
Observable.from(
    console.log,
    err => console.log("catch", err),
    () => console.log("Complete!")
);
// 未捕獲的Promise錯(cuò)誤
// (node:765) UnhandledPromiseRejectionWarning: error
// (node:765) UnhandledPromiseRejectionWarning: Unhandled promise rejection. This error originated either by throwing
// inside of an async function without a catch block, or by rejecting a promise which was not handled with .catch(). (
// rejection id: 1)
// (node:765) [DEP0018] DeprecationWarning: Unhandled promise rejections are deprecated. In the future, promise rejections that are not handled will terminate the Node.js process with a non-zero exit code.

fromEvent連接DOM與RxJS的橋梁

const event$ = Observable.fromEvent(document.getElementById("btn"),"click");
event$.subscribe(event=>{
  // Do something
})

在NodeJs中可以與EventEmitter交互

const Observable = require("rxjs").Observable;
const EventEmitter = require("events");
require("rxjs/add/observable/fromEvent")

const emitter = new EventEmitter();

const source$ = Observable.fromEvent(emitter,"msg");
source$.subscribe(console.log,null,()=>console.log("Complete"))

emitter.emit("msg",1)
// 1
emitter.emit("msg","haha")
// haha
emitter.emit("a-msg","haha")
//
emitter.emit("msg","nihao")
// nihao

fromEvent是Hot Observable,也就是數(shù)據(jù)的產(chǎn)生和訂閱無關(guān)，對(duì)于fromEvent來說，數(shù)據(jù)源是外部產(chǎn)生的，不受RxJS控制，這是Hot Observable對(duì)象的特點(diǎn)

fromEventPattern針對(duì)不規(guī)范的事件源

規(guī)范的事件源：DOM事件，EventEmitter事件

ajax見最上面的例子

repeatWhen

例如 在上游事件結(jié)束之后的一段時(shí)間再重新訂閱

const Observable = require("rxjs").Observable;
require("rxjs/add/operator/repeatWhen")

const notifier = ()=>{
    return Observable.interval(1000);
}

const source$ = Observable.of(1,2,3);
// const source$ = Observable.create(observer=>{
//     observer.next(111);
//     return {
//         unsubscribe(){
//             console.log("on Unsubscribe")
//         }
//     }
// });
const repeat$ = source$.repeatWhen(notifier);

repeat$.subscribe(console.log,null,()=>console.log("Complete"))
// 每隔一秒產(chǎn)生一次
// 1
// 2
// 3
// 1
// 2
// 3
// 1
// 2
// 3
// 1
// 2
// 3
// 1
// 2
// 3
// ^C

defer延遲創(chuàng)建Observable

針對(duì)Observable占用內(nèi)存比較大的情況，懶加載

const Observable = require("rxjs").Observable;
require("rxjs/add/observable/defer");
require("rxjs/add/observable/of");

const observableFactory = ()=>Observable.of(1,2,3);
const source$ = Observable.defer(observableFactory)

功能需求
操作符



把多個(gè)數(shù)據(jù)流以首尾相連的方式合并

concat,concatAll



把多個(gè)數(shù)據(jù)流以先到先得的方式合并

merge,mergeAll



把多個(gè)數(shù)據(jù)流中的數(shù)據(jù)以一一對(duì)應(yīng)的方式合并

zip和zipAll



持續(xù)合并多個(gè)數(shù)據(jù)流中最新產(chǎn)生的數(shù)據(jù)


combineLatest,combineAll,withLatestFrom



從多個(gè)數(shù)據(jù)流中選取第一個(gè)產(chǎn)生內(nèi)容的數(shù)據(jù)流
race


在數(shù)據(jù)流前面添加一個(gè)指定數(shù)據(jù)
startWith


只獲取多個(gè)數(shù)據(jù)流最后產(chǎn)生的數(shù)據(jù)
forkJoin


從高階數(shù)據(jù)流中切換數(shù)據(jù)源

switch,exhaust

concat

實(shí)例方法
靜態(tài)方法，如果兩個(gè)數(shù)據(jù)沒有先后關(guān)系，推薦使用此方法

實(shí)例方法

const Observable = require("rxjs").Observable;
require("rxjs/add/operator/of")
require("rxjs/add/operator/concat")

const source$1 = Observable.of(1,2,3);
const source$2 = Observable.of(4,5,6);

source$1.concat(source$2).subscribe(console.log,null,()=>console.log("Complete"))

靜態(tài)方法

const Observable = require("rxjs").Observable;
require("rxjs/add/operator/of")
require("rxjs/add/observable/concat")

const source$1 = Observable.of(1,2,3);
const source$2 = Observable.of(4,5,6);

Observable
  .concat(source$1,source$2)
  .subscribe(console.log,null,()=>console.log("Complete"))

`concat`在將上一個(gè)數(shù)據(jù)源傳遞下去的時(shí)候會(huì)調(diào)用上一個(gè)`Observable`的`unsubscribe`,如果上一個(gè)`Observable`一直為完結(jié)，后續(xù)的都不會(huì)被調(diào)用

```javascript
const source$1 = Observable.internal(1000);
const source$2 = Observable.of(1);
const concated$ = Observable.concat(source$1,source$2);
// 此時(shí) source$2永遠(yuǎn)不會(huì)被調(diào)用
```

在此推測(cè)：`rxjs/add/operator/*`下的屬性都是實(shí)例屬性，`rxjs/add/observable/*`下的屬性都是實(shí)例屬性

merge先到先得

merge用在同步數(shù)據(jù)的情況下和concat表現(xiàn)只,不建議使用

const Observable = require("rxjs").Observable;
require("rxjs/add/operator/merge");
require("rxjs/add/operator/map");
require("rxjs/add/observable/timer");

const source$1 = Observable.timer(0, 1000).map(x => x + "A");
const source$2 = Observable.timer(500, 1000).map(x => x + "B");
const source$3 = Observable.timer(1000, 1000).map(x => x + "C");

// 此時(shí) source$1與source$2永遠(yuǎn)不會(huì)停止，所以
source$1
    .merge(source$2, source$3, /*此參數(shù)限制了合并的Observable的個(gè)數(shù)*/ 2)
    .subscribe(console.log, null, () => console.log("Complete"));

// 0A
// 0B
// 1A
// 1B
// 2A
// 2B
// 3A
// 3B
// 4A
// 4B
// ^C