轉(zhuǎn)載
終于進(jìn)到了 RxJS 的第二個(gè)重點(diǎn) Subject岛宦，不知道讀者們有沒(méi)有發(fā)現(xiàn)？ 我們?cè)谶@篇文章之前的范例耍缴，每個(gè) observable 都只訂閱了一次砾肺，而實(shí)際上 observable 是可以多次訂閱的
var source = Rx.Observable.interval(1000).take(3);
var observerA = {
next: value => console.log('A next: ' + value),
error: error => console.log('A error: ' + error),
complete: () => console.log('A complete!')
}
var observerB = {
next: value => console.log('B next: ' + value),
error: error => console.log('B error: ' + error),
complete: () => console.log('B complete!')
}
source.subscribe(observerA);
source.subscribe(observerB);
// "A next: 0"
// "B next: 0"
// "A next: 1"
// "B next: 1"
// "A next: 2"
// "A complete!"
// "B next: 2"
// "B complete!"
上面這段程式碼，分別用 observerA 跟 observerB 訂閱了 source防嗡，從 log 可以看出來(lái) observerA 跟 observerB 都各自收到了元素变汪，但請(qǐng)記得這兩個(gè) observer 其實(shí)是分開(kāi)執(zhí)行的也就是說(shuō)他們是完全獨(dú)立的，我們把 observerB 延遲訂閱來(lái)證明看看
var source = Rx.Observable.interval(1000).take(3);
var observerA = {
next: value => console.log('A next: ' + value),
error: error => console.log('A error: ' + error),
complete: () => console.log('A complete!')
}
var observerB = {
next: value => console.log('B next: ' + value),
error: error => console.log('B error: ' + error),
complete: () => console.log('B complete!')
}
source.subscribe(observerA);
setTimeout(() => {
source.subscribe(observerB);
}, 1000);
// "A next: 0"
// "A next: 1"
// "B next: 0"
// "A next: 2"
// "A complete!"
// "B next: 1"
// "B next: 2"
// "B complete!"
這裡我們延遲一秒再用 observerB 訂閱蚁趁，可以從 log 中看出 1 秒后 observerA 已經(jīng)印到了 1裙盾，這時(shí) observerB 開(kāi)始印卻是從 0 開(kāi)始，而不是接著 observerA 的進(jìn)度他嫡，代表這兩次的訂閱是完全分開(kāi)來(lái)執(zhí)行的番官，或者說(shuō)是每次的訂閱都建立了一個(gè)新的執(zhí)行。
這樣的行為在大部分的情境下適用涮瞻，但有些案例下我們會(huì)希望第二次訂閱 source 不會(huì)從頭開(kāi)始接收元素鲤拿，而是從第一次訂閱到當(dāng)前處理的元素開(kāi)始發(fā)送，我們把這種處理方式稱(chēng)為組播(multicast)署咽，那我們要如何做到組播呢近顷？
手動(dòng)建立 subject
或許已經(jīng)有讀者想到解法了，其實(shí)我們可以建立一個(gè)中間人來(lái)訂閱 source 再由中間人轉(zhuǎn)送資料出去宁否，就可以達(dá)到我們想要的效果
var source = Rx.Observable.interval(1000).take(3);
var observerA = {
next: value => console.log('A next: ' + value),
error: error => console.log('A error: ' + error),
complete: () => console.log('A complete!')
}
var observerB = {
next: value => console.log('B next: ' + value),
error: error => console.log('B error: ' + error),
complete: () => console.log('B complete!')
}
var subject = {
observers: [],
addObserver: function(observer) {
this.observers.push(observer)
},
next: function(value) {
this.observers.forEach(o => o.next(value))
},
error: function(error){
this.observers.forEach(o => o.error(error))
},
complete: function() {
this.observers.forEach(o => o.complete())
}
}
subject.addObserver(observerA)
source.subscribe(subject);
setTimeout(() => {
subject.addObserver(observerB);
}, 1000);
// "A next: 0"
// "A next: 1"
// "B next: 1"
// "A next: 2"
// "B next: 2"
// "A complete!"
// "B complete!"
從上面的程式碼可以看到窒升，我們先建立了一個(gè)物件叫 subject，這個(gè)物件具備 observer 所有的方法(next, error, complete)慕匠，并且還能 addObserver 把 observer 加到內(nèi)部的清單中饱须，每當(dāng)有值送出就會(huì)遍歷清單中的所有 observer 并把值再次送出，這樣一來(lái)不管多久之后加進(jìn)來(lái)的 observer台谊，都會(huì)是從當(dāng)前處理到的元素接續(xù)往下走蓉媳，就像范例中所示，我們用 subject 訂閱 source 并把 observerA 加到 subject 中锅铅，一秒后再把 observerB 加到 subject酪呻，這時(shí)就可以看到 observerB 是直接收 1 開(kāi)始，這就是組播(multicast)的行為盐须。
讓我們把 subject 的 addObserver 改名成 subscribe 如下
var subject = {
observers: [],
subscribe: function(observer) {
this.observers.push(observer)
},
next: function(value) {
this.observers.forEach(o => o.next(value))
},
error: function(error){
this.observers.forEach(o => o.error(error))
},
complete: function() {
this.observers.forEach(o => o.complete())
}
}
應(yīng)該有眼尖的讀者已經(jīng)發(fā)現(xiàn)玩荠，subject 其實(shí)就是用了 Observer Pattern。但這邊為了不要混淆 Observer Pattern 跟 RxJS 的 observer 就不再內(nèi)文提及。這也是為什麼我們?cè)谝婚_(kāi)始講 Observer Pattern 希望大家親自實(shí)作的原因阶冈。
RxJS 中的 Subject 確實(shí)是類(lèi)似這樣運(yùn)作的闷尿，可以在原始碼中看到
雖然上面是我們自己手寫(xiě)的 subject，但運(yùn)作方式跟 RxJS 的 Subject 實(shí)例是幾乎一樣的女坑，我們把前面的程式碼改成 RxJS 提供的 Subject 試試
var source = Rx.Observable.interval(1000).take(3);
var observerA = {
next: value => console.log('A next: ' + value),
error: error => console.log('A error: ' + error),
complete: () => console.log('A complete!')
}
var observerB = {
next: value => console.log('B next: ' + value),
error: error => console.log('B error: ' + error),
complete: () => console.log('B complete!')
}
var subject = new Rx.Subject()
subject.subscribe(observerA)
source.subscribe(subject);
setTimeout(() => {
subject.subscribe(observerB);
}, 1000);
// "A next: 0"
// "A next: 1"
// "B next: 1"
// "A next: 2"
// "B next: 2"
// "A complete!"
// "B complete!"
大家會(huì)發(fā)現(xiàn)使用方式跟前面是相同的填具，建立一個(gè) subject 先拿去訂閱 observable(source)，再把我們真正的 observer 加到 subject 中堂飞，這樣一來(lái)就能完成訂閱灌旧，而每個(gè)加到 subject 中的 observer 都能整組的接收到相同的元素。
什麼是 Subject?
雖然前面我們已經(jīng)示范直接手寫(xiě)一個(gè)簡(jiǎn)單的 subject绰筛，但到底 RxJS 中的 Subject 的概念到底是什麼呢？
首先 Subject 可以拿去訂閱 Observable(source) 代表他是一個(gè) Observer描融，同時(shí) Subject 又可以被 Observer(observerA, observerB) 訂閱铝噩，代表他是一個(gè) Observable。
總結(jié)成兩句話
- Subject 同時(shí)是 Observable 又是 Observer
- Subject 會(huì)對(duì)內(nèi)部的 observers 清單進(jìn)行組播(multicast)
補(bǔ)充： 沒(méi)事不要看窿克！其實(shí) Subject 就是 Observer Pattern 的實(shí)作并且繼承自 Observable骏庸。
今日小結(jié)
今天介紹了 RxJS 中的第二個(gè)重點(diǎn) Subject，重點(diǎn)放在 Subject 主要的運(yùn)作方式年叮，以及概念上的所代表的意思具被，如果今天還不太能夠吸收的讀者不用緊張，后面我們會(huì)講到 subject 的一些應(yīng)用只损，到時(shí)候就會(huì)有更深的體會(huì)一姿。
不知道今天讀者麼有沒(méi)有收穫呢？ 如果有任何問(wèn)題跃惫，歡迎在下方留言給我叮叹。
