原文鏈接: https://medium.com/@benlesh/on-the-subject-of-subjects-in-rxjs-2b08b7198b93
本文為 RxJS 中文社區(qū) 翻譯文章膜钓，如需轉(zhuǎn)載，請注明出處，謝謝合作沾瓦！
如果你也想和我們一起释移，翻譯更多優(yōu)質(zhì)的 RxJS 文章以奉獻(xiàn)給大家关顷，請點(diǎn)擊【這里】
RxJS 中的 Subjects 經(jīng)常被誤解餐济。因?yàn)樗鼈冊试S你命令式地向 Observable 流中推送值侨赡，當(dāng)人們不太清楚如何將某個(gè)東西變成 Observable 時(shí)嗦嗡，他們傾向于濫用 Subjects 勋锤。此模式看起來有點(diǎn)像這樣...
// 當(dāng)人們發(fā)現(xiàn)不太清楚如何做時(shí)，通常會首先使用 Subjects
// (不要這樣做)
const subject = new Subject();
button.addEventListener('click', () => subject.next('click');
subject.subscribe(x => console.log(x));
雖然這對于 RxJS 新手 (對于這個(gè)階段來說太正常不過了) 很有幫助侥祭，但這不是以 “Rx 的方式”在處理問題叁执。理想的是在 Observable 中包裝事件注冊茄厘，既可以監(jiān)聽事件，又可以取消事件監(jiān)聽谈宛〈喂看起來像這樣:
// 這樣好些了，但請使用 Observable.fromEvent(button, 'click') 來替代
const clicks = new Observable(observer => {
const handler = (e) => observer.next(e);
button.addEventListener('click', handler);
return () => button.removeEventListener('click', handler);
});
為什么展示這個(gè)跟 Subjects 沒半點(diǎn)關(guān)系的示例吆录？好吧窑滞，一點(diǎn)是它展示了為什么不總是需要使用 Subject，另外一點(diǎn)這有個(gè)隱藏的 subject... (某種程度上可以說是 subject )恢筝。這里要注意的一點(diǎn)是哀卫，Observable 是通過 addEventListener 來包裝按鈕的處理函數(shù)的注冊，而 addEventListener 本身就是一個(gè) subject 撬槽〈烁模…至少根據(jù) “Gang Of Four” 的觀察者模式來說是這樣的矢渊。
觀察者模式
你可能知道另萤，RxJS 主要是關(guān)于 Observables 和 Observers 的，但它也是 Subjects 相關(guān)的挑秉。然而在 Gof 的 設(shè)計(jì)模式中是找不到 observables 的勋拟，Subjects 和 Observers 是 觀察者模式中的根本勋磕。
模式本身很簡單。Observers 是具有通知方法的類敢靡，Subject 也是類挂滓，它具有向內(nèi)部觀察者列表添加或刪除觀察者的方法和通知觀察者列表的方法。
RxJS 中的 Subjects 并沒有太差區(qū)別啸胧。當(dāng)使用 observer 對 Rx Subject 調(diào)用 subscribe 時(shí)赶站，Subject 會將該 observer 添加到內(nèi)部的觀察者列表中。同樣的纺念，如果使用一到三個(gè)函數(shù)來調(diào)用 subscribe贝椿，Subject 會將它們包裝成一個(gè) observer，然后添加到觀察者列表中陷谱。當(dāng)調(diào)用 Subject 的 next(value) 時(shí)烙博，它會遍歷觀察者列表并將 value 傳遞給 next 方法。對于 error 和 complete 也是同樣的烟逊。要想從 subject 的觀察者列表中移除 observer渣窜，只需簡單調(diào)用 subscription 的 unsubscribe 方法即可，subscription 是將 observer 添加到觀察者列表中時(shí)返回的宪躯。
const subject = new Subject();
// 將 observer1 添加到觀察者列表
const sub1 = subject.subscribe(observer1);
// 將 observer2 添加到觀察者列表
const sub2 = subject.subscribe(observer2);
// 使用 "hi there" 來通知列表中的所有觀察者
subject.next('hi there');
// 將 observer1 從觀察者列表中移除
sub1.unsubscribe();
Subject 相比較于 Observable
實(shí)際上乔宿，RxJS 中的 Subjects 不同于 GoF 觀察者模式中的 Subjects，但它們的 API 是 Observable 的鴨子類型访雪。實(shí)際上详瑞，在 RxJS 中 Subjects 更甚掂林，它們繼承自 Observable 。優(yōu)點(diǎn)是所有 Subjects 都具有與 Observable 相同的操作符和方法坝橡。
大概 Subject 和 Observable 之間一個(gè)很重要的區(qū)別就是 Subject 是有狀態(tài)的泻帮，它維護(hù)觀察者列表。另一方面驳庭，Observable 真的只是一個(gè)函數(shù)刑顺，它建立了觀察本身。
雖然 Subjects 是 Observables饲常，但 Subjects 還實(shí)現(xiàn)了 Observer 接口。也就是說狼讨，它們擁有 next贝淤、error 和 complete 方法。這些方法用來通知 subject 內(nèi)部觀察者列表中的 observers 政供。這意味著 subject 可以用作訂閱任何 observable 的 observer 播聪。
// 為了使兩個(gè)觀察者 observer1 和 observer2 “共享” tick$，
// 我們可以通過 Subject 來傳輸所有通知布隔，像這樣
const tick$ = Observable.interval(1000);
const subject = new Subject();
subject.subscribe(observer1);
subject.subscribe(observer2);
tick$.subscribe(subject);
上面的示例是將 observable tick$ “多播” 給兩個(gè)觀察者: observer1 和 observer2 离陶。這其實(shí)就是 RxJS 中大多數(shù)多播操作符內(nèi)部所做的事情。例如 publish衅檀、publishReplay招刨、multicast、share哀军，等等沉眶。真的，這才是 RxJS 中 Subjects 的主要用法杉适。
Subjects 是不可重用的
在 RxJS 中谎倔，Subjects 不能重用。也就是說猿推，當(dāng)一個(gè) Subject 完成或報(bào)錯(cuò)時(shí)片习，便不可再使用了。如果你嘗試在已關(guān)閉的 Subject (調(diào)用過 complete 或 error 方法)上調(diào)用 next蹬叭，它會默認(rèn)忽略通知藕咏。如果想 Subject 在完成后調(diào)用 next 時(shí)進(jìn)行顯示地報(bào)錯(cuò)，你可以在 subject 實(shí)例上直接調(diào)用 unsubscribe 具垫。
// Subject 之死
const subject = new Subject();
subject.subscribe(x => console.log(x));
subject.next(1); // 1
subject.next(2); // 2
subject.complete();
subject.next(3); // 悄悄地忽略
subject.unsubscribe();
subject.next(4); // Unhandled ObjectUnsubscribedError
RxJS 中的陷阱
但是在 RxJS 的當(dāng)前版本中侈离，會帶來了一些令人困惑的痛點(diǎn)。因?yàn)?Rx observables 不會“捕獲”錯(cuò)誤筝蚕，我們會遭遇一些奇怪的行為卦碾。我曾經(jīng)嘲笑過 Promises 實(shí)現(xiàn)了錯(cuò)誤“捕獲”铺坞，但在多播場景中它或許是正確的。我的意思是當(dāng)我說 Rx observable 不“捕獲”錯(cuò)誤時(shí)洲胖，是表示當(dāng)錯(cuò)誤滲透到觀察者鏈末端并的末端時(shí)济榨，如果錯(cuò)誤未被處理，它會被重新拋出绿映。
// 演示缺少錯(cuò)誤處理時(shí)會進(jìn)行重新拋出
const badObservable = Observable.throw(new Error('haha'));
try {
badObservable.subscribe({
next: x => console.log(x),
error: null,
complete: () => console.log('done')
});
} catch (err) {
console.error(err); // 輸出自定義錯(cuò)誤: "haha"
}
現(xiàn)在我們來想想擒滑，當(dāng)你循環(huán)觀察者列表并通知它們時(shí)會發(fā)生什么(正如 subject 所做的)。
for (let observer of observers) {
observer.next('notify'); // 如果在這里調(diào)用 throw 會發(fā)生什么叉弦？
}
// 提示: 會報(bào)錯(cuò)會打破循環(huán)
// 注意: 好吧丐一，這不僅僅是個(gè)提示
假設(shè)一些操作符是同步處理的(map、filter淹冰、scan 等等)库车，如果其中一個(gè)或任何其它同步操作符報(bào)錯(cuò)了，你會在多播(使用 Subject 來循環(huán)觀察者列表并通知它們)的下游得到一些詭異的行為:
// 會發(fā)生奇怪的行為
const source$ = Observable.interval(1000).share();
const mapped$ = source$.map(x => {
if (x === 1) {
throw new Error('oops');
}
return x;
});
source$.subscribe(x => console.log('A', x));
mapped$.subscribe(x => console.log('B', x));
source$.subscribe(x => console.log('C', x));
// "A" 0
// "B" 0
// "C" 0
// "A" 1
// Uncaught Error: "oops"
在上面的示例中樱拴，大多數(shù)用戶會期望 A 和 C 能繼續(xù)通知柠衍。輸出 B 的 observable 死了是可以理解的，它報(bào)錯(cuò)了晶乔，但其他流和源流也死了令人想當(dāng)困惑珍坊。任意的第三方都可以殺掉共享的 observable 流以及未知數(shù)量的兄弟流，不應(yīng)該是這樣的正罢。這是一個(gè)脆弱的抽象阵漏，我們需要在 RxJS 接下來的版本中修復(fù)它。
臨時(shí)解決上述場景中的問題很簡單腺怯，感謝調(diào)度器( schedulers )袱饭。你可以在多播后使用 observeOn，這樣就可以解決此問題呛占，因?yàn)殄e(cuò)誤不再是同步拋出的虑乖。
const source$ = Observable.interval(1000)
.share()
.observeOn(Rx.Scheduler.asap); // 點(diǎn)睛之筆
const mapped$ = source$.map(x => {
if (x === 1) {
throw new Error('oops');
}
return x;
});
source$.subscribe(x => console.log('A', x));
mapped$.subscribe(x => console.log('B', x));
source$.subscribe(x => console.log('C', x));
// "A" 0
// "B" 0
// "C" 0
// "A" 1
// Uncaught Error: "oops"
// "C" 1
// "A" 2
// "C" 2
// "A" 3
// "C" 3
// ... 等等
還有另一種臨時(shí)解決方案，如果你可以管理它的話晾虑，它的性能會更好一些疹味，方法是只需為所有的 subscriptions 添加錯(cuò)誤處理方法。
const source$ = Observable.interval(1000)
.share()
.observeOn(Rx.Scheduler.asap); // 點(diǎn)睛之筆
const mapped$ = source$.map(x => {
if (x === 1) {
throw new Error('oops');
}
return x;
});
source$.subscribe(x => console.log('A', x));
mapped$.subscribe(
x => console.log('B', x),
err => console.log('Error handled: ' + err.message)
);
source$.subscribe(x => console.log('C', x));
// "A" 0
// "B" 0
// "C" 0
// "A" 1
// "Error handled: oops"
// "C" 1
// "A" 2
// "C" 2
// "A" 3
// "C" 3
// ... 等等
Observable 的未來
TC39 Observable 提議 新的化身帜篇，不包括 CancelToken 業(yè)務(wù)糙捺，提議本身就是一整篇文章，它可能會在沒有錯(cuò)誤處理方法的情況下通過“捕獲”錯(cuò)誤來解決這個(gè)問題笙隙。也就是說洪灯，它不再會到達(dá)觀察者鏈末端并重新拋出錯(cuò)誤。在 RxJS 未來的版本中竟痰，我想我們也會做同樣的事情签钩，因?yàn)檫@才是正確的掏呼。當(dāng)然，這個(gè)問題是公開討論的铅檩，但我個(gè)人認(rèn)為這不會有太多阻力憎夷。
總結(jié)
- Subjects 既是 observer，又是 observable
- Subjects 對內(nèi)部的觀察者列表進(jìn)行“多播”
- Observables 只是建立觀察的函數(shù)
- Observables 當(dāng)前不會捕獲錯(cuò)誤昧旨，但它們應(yīng)該捕獲
- 向 Subject 的下游同步拋出錯(cuò)誤會殺掉此 Subject
- 你可以使用錯(cuò)誤處理方法或 observeOn 來解決 #4 中的問題
- 關(guān)于 Promise 的錯(cuò)誤捕獲拾给，我錯(cuò)了。這是個(gè)好主意兔沃，因?yàn)?promises 是多播的 *
- 未來版本的 RxJS 很可能會捕獲錯(cuò)誤
- 雖然可能不是完全必要的蒋得，因?yàn)?promises 永遠(yuǎn)是異步的。 (聳肩臉)
