rxjs簡介

Observable

Observable相似於函數聲明

var observable = Rx.Observable.create(function subscribe(observer) {
  var id = setInterval(() => {
    observer.next('hi')
  }, 1000)
});

subscribe相似於函數調用(相似於function.call),調用時傳入回調函數observer對象來執行

observable.subscribe({next:x => console.log(x)})

執行後的返回值稱爲Subscription,能夠中止後續Observable的執行,Subscription還能夠合在一塊兒,一塊兒終止多個Observable的執行

var subscription = observable.subscribe({next:x => console.log(x)})
subscription.unsubscribe()

Observable是多個值的惰性推送集合。

Observable和普通函數的區別:

普通的function會同步的返回一個值,屬於一種拉取值的操做,你能夠肯定這個值是存在的,而Observable能夠同步或異步的返回一個或多個值,屬於推送值的操做,相似於Promise,你並不肯定這個值是否存在,也不知道什麼時候會接收到數據,可是能夠經過回調函數的方式,明確當值可用的時候作什麼操做.

爲何要用Observable呢,由於它能夠完成一些普通function完成不了的工做,好比:

純淨性

純淨性使你的代碼無需引入外部狀態,更不容易出錯。

var count = 0;
var button = document.querySelector('button');
button.addEventListener('click', () => console.log(`Clicked ${++count} times`));

// 使用rxjs
var button = document.querySelector('button');
Rx.Observable.fromEvent(button, 'click')
  .scan(count => count + 1, 0)
  .subscribe(count => console.log(`Clicked ${count} times`));

流動性

RxJS 提供了一整套操做符來幫助你控制事件如何流經 observables ,代碼更簡潔,更可讀。

var count = 0;
var rate = 1000;
var lastClick = Date.now() - rate;
var button = document.querySelector('button');
button.addEventListener('click', () => {
  if (Date.now() - lastClick >= rate) {
    console.log(`Clicked ${++count} times`);
    lastClick = Date.now();
  }
});

var button = document.querySelector('button');
Rx.Observable.fromEvent(button, 'click')
  .throttleTime(1000)
  .scan(count => count + 1, 0)
  .subscribe(count => console.log(`Clicked ${count} times`));

Subject

Subject相似於事件監聽,會在內部維護多個observer

var subject = new Rx.Subject();

subject.subscribe({
  next: (v) => console.log('observerA: ' + v)
});
subject.subscribe({
  next: (v) => console.log('observerB: ' + v)
});

調用Subject.next時發送事件通知,將依次執行Subject內部維護的所有observer

var subject = new Rx.Subject();

subject.subscribe({
  next: (v) => console.log('observerA: ' + v)
});
subject.subscribe({
  next: (v) => console.log('observerB: ' + v)
});

subject.next(1)
subject.next(2)

在某個Observable聲明以後,進行subscribe調用時也能夠傳入Subject,這將依次執行Subject內部維護的所有observer

var subject = new Rx.Subject();

subject.subscribe({
  next: (v) => console.log('observerA: ' + v)
});
subject.subscribe({
  next: (v) => console.log('observerB: ' + v)
});

var observable = Rx.Observable.from([1, 2, 3]);

observable.subscribe(subject);

BehaviorSubject

BehaviorSubject是Subject的一個變體,它有一個「當前值」的概念。當前值是發送給消費者的最後一個值。在new BehaviorSubject時能夠初始化一個當前值。當有新的觀察者訂閱時，會當即從 BehaviorSubject 那接收到「當前值」。

var subject = new Rx.BehaviorSubject(0); // 0是初始的當前值

subject.subscribe({
  next: (v) => console.log('observerA: ' + v)
});

subject.next(1);
subject.next(2);

subject.subscribe({
  next: (v) => console.log('observerB: ' + v)
});

subject.next(3);

ReplaySubject

ReplaySubject相似於BehaviorSubject,它有一個「緩衝值」的概念,能夠想象成多個"當前值"。在new ReplaySubject時能夠初始化一個"緩衝值"的數量,當有新的觀察者訂閱時，會當即從 ReplaySubject 那接收到每個「緩衝值」。

除了緩衝數量，你還能夠指定 window time (以毫秒爲單位)來肯定多久以前的值能夠記錄。

var subject = new Rx.ReplaySubject(3); // 爲新的訂閱者緩衝3個值

subject.subscribe({
  next: (v) => console.log('observerA: ' + v)
});

subject.next(1);
subject.next(2);
subject.next(3);
subject.next(4);

subject.subscribe({
  next: (v) => console.log('observerB: ' + v)
});

subject.next(5);

AsyncSubject

AsyncSubject 是另外一個 Subject 變體，只有當 Observable 執行完成時(執行 complete())，它纔會將執行的最後一個值發送給觀察者。

var subject = new Rx.AsyncSubject();

subject.subscribe({
  next: (v) => console.log('observerA: ' + v)
});

subject.next(1);
subject.next(2);
subject.next(3);
subject.next(4);

subject.subscribe({
  next: (v) => console.log('observerB: ' + v)
});

subject.next(5);
subject.complete();

總結

observable的執行相似函數調用,subject的執行相似事件監聽.

observer是一些待執行的函數,Observable能夠執行observer,Subject也能夠執行observer,只不過是兩種不一樣的運行模式,究竟用哪一種視狀況而定

操做符

操做符就是以特定的observer來訂閱原observable,而後返回一個以特定操做結果做爲next值的新的Observable

function multiplyByTen(input) {
    // 原observable被封裝在新的observable內部,只有外部的observable被訂閱時纔會執行內部的訂閱
    const output = Observable.create(function subscribe(observer) {
        input.subscribe(val => {
            observer.next(val * 10)
        })
    })
    // 返回新的observable
    return output;
}

var source = Observable.create(function subscribe(observer) {
    observer.next(1)
    observer.next(2)
});

multiplyByTen(source).subscribe(val => console.log(val))