angular7 Rxjs 異步請求

Promise 和 RxJS 處理異步對比

Promise 處理異步:

let promise = new Promise(resolve => {
setTimeout(() => {
resolve('---promise timeout---');
}, 2000);
});
promise.then(value => console.log(value));

 RxJS 處理異步

import {Observable} from 'rxjs';
let stream = new Observable(observer => {
setTimeout(() => {
observer.next('observable timeout');
}, 2000);
});
stream.subscribe(value => console.log(value));

 

 從上面列子能夠看到 RxJS 和 Promise 的基本用法很是相似，除了一些關鍵詞不一樣。Promise 裏面用的是 then() 和 resolve()，而 RxJS 裏面用的是 next() 和 subscribe()。從上面例子咱們感受Promise 和 RxJS 的用法基本類似。其實Rxjs相比Promise 要強大不少。 好比 Rxjs 中能夠中途撤回、Rxjs 能夠發射多個值、Rxjs 提供了多種工具函數等等。

Rxjs unsubscribe 取消訂閱

Promise 的建立以後，動做是沒法撤回的。Observable 不同，動做能夠經過 unsbscribe() 方 法中途撤回，並且 Observable 在內部作了智能的處理。

Promise 建立以後動做沒法撤回

 

let promise = new Promise(resolve => {
setTimeout(() => {
resolve('---promise timeout---');
}, 2000);
});
promise.then(value => console.log(value));

Rxjs 能夠經過 unsubscribe() 能夠撤回 subscribe 的動做

let stream = new Observable(observer => {
let timeout = setTimeout(() => {
clearTimeout(timeout);
observer.next('observable timeout');
}, 2000);
});
let disposable = stream.subscribe(value => console.log(value));
setTimeout(() => {
//取消執行
disposable.unsubscribe();
}, 1000);

 Rxjs 訂閱後屢次執行

 若是咱們想讓異步裏面的方法屢次執行，好比下面代碼。

這一點 Promise 是作不到的，對於 Promise 來講，最終結果要麼 resole（兌現）、要麼 reject （拒絕），並且都只能觸發一次。若是在同一個 Promise 對象上屢次調用 resolve 方法， 則會拋異常。而 Observable 不同，它能夠不斷地觸發下一個值，就像 next() 這個方法的 名字所暗示的那樣。

let promise = new Promise(resolve => {
setInterval(() => {
resolve('---promise setInterval---');
}, 2000);
});
promise.then(value => console.log(value));

 

 Rxjs

et stream = new Observable<number>(observer => {
let count = 0;
setInterval(() => {
observer.next(count++);
}, 1000);
});
stream.subscribe(value => console.log("Observable>"+value));

 Angualr6.x 以前使用 Rxjs 的工具函數 map filter

 注意：Angular6 之後使用之前的 rxjs 方法，必須安裝 rxjs-compat 模塊纔可使用 map、filter 方法。

angular6 後官方使用的是 RXJS6 的新特性，因此官方給出了一個能夠暫時延緩咱們不須要修 改 rsjx 代碼的辦法。

npm install rxjs-compat

　　

et stream= new Observable<any>(observer => {
let count = 0;
setInterval(() => {
observer.next(count++);
}, 1000);
});
stream.filter(val=>val%2==0)
.subscribe(value => console.log("filter>"+value));
stream
.map(value => {
return value * value
})
.subscribe(value => console.log("map>"+value));

 Angualr6.x 之後 Rxjs6.x 的變化以及 使用

 Rxjs 的變化參考文檔：http://bbs.itying.com/topic/5bfce189b110d80f905ae545

RXJS6 改變了包的結構，主要變化在 import 方式和 operator 上面以及使用 pipe()

 

import {Observable} from 'rxjs';
import {map,filter} from 'rxjs/operators';

 

let stream= new Observable<any>(observer => {
let count = 0;
setInterval(() => {
observer.next(count++);
}, 1000);
});
stream.pipe(
filter(val=>val%2==0)
)
.subscribe(value => console.log("filter>"+value));
stream.pipe(
filter(val=>val%2==0), map(value => {
return value * value
})
)
.subscribe(value => console.log("map>"+value));

 Rxjs 延遲執行

import {Observable,fromEvent} from 'rxjs';
import {map,filter,throttleTime} from 'rxjs/operators';
var button = document.querySelector('button');
fromEvent(button, 'click').pipe(
throttleTime(1000)
)
.subscribe(() => console.log(`Clicked`));