我们知道 Rx 是 push 模式。比如 1 秒产生一个事件的流，即使我们不去使用它，它也会产生事件。如何人为的停止这个流是今天我们要解决的问题。

先把场景搭建好，从页面中得到按钮的 click 事件流：

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：这个操作符接收一个事件流参数，丢弃原始的事件流，使用参数提供的事件流继续运行。

我们没有对之前的程序逻辑做任何修改，只是添加了一个开始按钮的事件流，再和之前的事件流组合起来完成了开始输出，定时输出，停止输出的功能。有没有点儿搭积木的意思？

有问题请添加微信公众号“读一读我”。