我們知道 Rx 是 push 模式。比如 1 秒產生一個事件的流，即使我們不去使用它，它也會產生事件。如何人為的停止這個流是今天我們要解決的問題。

import React, { useRef, useEffect } from "react";import { fromEvent } from "rxjs";export default function App() { const btnRef = useRef(null); useEffect(() => { const stopBtnClick$ = fromEvent(btnRef.current, "click"); const subscription = stopBtnClick$.subscribe(() => { console.log("單擊事件"); }); return () => { subscription.unsubscribe(); }; }); return ( <div className="App"> <button ref={btnRef}>停止按鈕</button> </div> );}

然後我們使用 interval 操作符建立一個間隔為 1 秒發生的事件流：

const perSecond$ = interval(1000);const subscription = perSecond$.subscribe((value) => { console.log(value);});

我們知道 subscribe 函式會返回一個 subscription 物件，這個物件上有個 unsubscribe 函式，呼叫這個函式將會停止事件流。這也是我們為什麼要在 useEffect 的返回值中呼叫它，防止記憶體洩露。既然如此，停止 perSecond$ 就簡單了：

const stopBtnClick$ = fromEvent(btnRef.current, "click");const perSecond$ = interval(1000);const perSecondSubscription = perSecond$.subscribe(value => { console.log(value);});stopBtnClick$.subscribe(_ => { perSecondSubscription.unsubscribe();});

這樣做確實可以解決我們的問題，然而這是完全錯誤的做法。

之前的文章說過，Rx 程式設計模型是組合程式設計理念的最佳實踐。讓我們來考慮一下如何組合 stopBtnClick\\$ 和 perSecond\\$ 來解決我們的問題：

takeUntil：這個操作符接收另一個流作為引數， 等於它監控了兩個流。它做的事就是從原始的流接收事件，但不是一直接收，當引數流中開始產生事件時，它就停止接收原始流的事件了，也就是把原始流停掉了。

有了這個操作符，就可以組合使用之前的兩個流了：

const stopBtnClick$ = fromEvent(btnRef.current, "click");const perSecond$ = interval(1000);const intervalCanBeStopped$ = perSecond$.pipe(takeUntil(stopBtnClick$));const subscription = intervalCanBeStopped$.subscribe(v => console.log(v));

我們再給時間間隔流加一個開始按鈕，完整程式碼如下：

import React, { useRef, useEffect } from "react";import { fromEvent, interval } from "rxjs";import { takeUntil, switchMapTo } from "rxjs/operators";export default function App() { const stopBtnRef = useRef(null); const startBtnRef = useRef(null); useEffect(() => { const stopBtnClick$ = fromEvent(stopBtnRef.current, "click"); const startBtnClick$ = fromEvent(startBtnRef.current, "click"); const perSecond$ = interval(1000); const intervalCanBeStopped$ = perSecond$ .pipe(takeUntil(stopBtnClick$)); const subscription = startBtnClick$.pipe( switchMapTo(intervalCanBeStopped$) ).subscribe(v => console.log(v)); return () => { subscription.unsubscribe(); }; }); return ( <div className="App"> <button ref={startBtnRef}>開始按鈕</button> <button ref={stopBtnRef}>停止按鈕</button> </div> );}

switchMapTo：這個操作符接收一個事件流引數，丟棄原始的事件流，使用引數提供的事件流繼續執行。

我們沒有對之前的程式邏輯做任何修改，只是添加了一個開始按鈕的事件流，再和之前的事件流組合起來完成了開始輸出，定時輸出，停止輸出的功能。有沒有點兒搭積木的意思？