响应式编程实战——RxJS 手动停止事件流的正确方式

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

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

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