自定義創(chuàng)建Observable
前文中我們已經(jīng)使用RxJS提供給我們的from()和of()函數(shù)創(chuàng)建Observable對象。然而Observable是如何和Observer交互的，以及如何取消訂閱，都有助于我們理解RxJS是如何運作的。
這里我們要做的是實現(xiàn)一個observable函數(shù)，接收一些參數(shù)，返回一個對象，比如叫subscription對象。我們利用這個對象來釋放資源。
const observable = dataSource => {
const INTERVAL = 1000;
let schedulerId;
return {
subscribe: observer => {//接受一個observer作為參數(shù)
schedulerId = setInterval(() => {
if(dataSource.length === 0) {
observer.complete();//通知observer數(shù)據(jù)(事件)全部發(fā)送完畢
clearInterval(schedulerId);
schedulerId = undefined;
} else {
observer.next(dataSource.shift());//把數(shù)據(jù)(事件)傳遞給observer
}
}, INTERVAL);
return {//返回一個對象，我們可以稱之為subscription，包含一個取消訂閱的函數(shù)
unsubscribe: () => {//取消訂閱，將不再發(fā)送數(shù)據(jù)(事件)給observer
if(schedulerId) {
clearInterval(schedulerId);
}
}
};
}
}
};
來看看如何使用：
let subscription = observable([1, 2, 3]).subscribe({//傳遞給subscribe函數(shù)observer對象
next: console.log,//包含next函數(shù)
complete: () => console.log('事件全部發(fā)送完畢')//包含complete函數(shù)
});
上面代碼的結(jié)果就是在控制臺每隔一秒打印一個數(shù)字，最后打印“事件全部發(fā)送完畢”。如下圖：
我們再看看如何在事件全部發(fā)送完畢之前取消操作，如下圖：
因為每一秒發(fā)送一個事件，我們在2秒后調(diào)用了取消訂閱函數(shù)，因此結(jié)果只打印了1，2，說明發(fā)送事件操作被取消了。
RxJS中提供了Observable的靜態(tài)函數(shù)create()來完成和上面一樣的功能。
const observable = Rx.Observable.create(observer => {
observer.next(1);
observer.next(2);
observer.next(3);
observer.complete();
});
const subscription = observable.subscribe(console.log);
create()函數(shù)接收一個函數(shù)作為參數(shù)，這個參數(shù)函數(shù)實際就是observable這個對象的subscribe函數(shù)，就像上面我們自定義的subscribe函數(shù)一樣。我們看到使用的時候傳給subscribe函數(shù)的并不是一個observer，而是一個函數(shù)，在RxJS內(nèi)部，subscribe函數(shù)的重載會自動為我們創(chuàng)建observer，并把console.log這個函數(shù)賦值給了next函數(shù)，也就是說，observer.next(1)這個操作實際就是調(diào)用的console.log(1)，即在控制臺打印數(shù)字1。這里值得注意的地方就是，如果一個observable包裝的數(shù)據(jù)源是有限個數(shù)的，那么你可以調(diào)用complete()函數(shù)表明所有數(shù)據(jù)(事件)都發(fā)送完畢了。
綜上，我們可以自定義Observable，自定義它發(fā)送數(shù)據(jù)的行為，并且可以在整個應(yīng)用程序中隨時重用它。
Observable是惰性求值的，不像Promise創(chuàng)建即開始運行。
