响应式编程实战—— RxJS 中的 combineLatest 操作符

之前文章介绍的例子都是处理一个流中的事件。然而在实际的业务中我们往往会遇到同时处理两个流的需求。比如我们需要从两个不同的 api 获取数据，然后合并数据在前端显示等等。

首先为我们之前的例子添加一个文本输入框 input，并获取它的输入事件流：

const input$ = fromEvent(inputRef.current, "input");

然而我们把输入流中的事件变换为输入值（默认是输入事件对象），同时把之前的代码做下整理：

const input$ = fromEvent(inputRef.current, "input").pipe(
  map(e => e.target.value),
);

const timer$ = time$.pipe(
  switchMap(addOneOrReset),
  startWith({ count: 0 }),
  scan((acc, current) => current(acc)),
  map(obj => obj.count)
  tap(v => setTxt(v))
);

tap：它的作用就是对流过的数据进行处理，然后原封不动的再把原数据传递给接下来的操作符。我们一般用它来进行产生负效果的操作（之前的负效果代码是写在 subscribe 函数中的），比如写日志啊，更新页面等等。这里其实遵循的是某一种 Rx 编程模型最佳实践。也就是在 subscribe 函数中不做任何操作，有点儿类似函数式编程中 IO Monad。当然，现在我们关注的重点是使用操作符完成功能。

准备工作做好了，现在我们要做的是如何同时使用输入流(input$)和定时器流(timer$)中的数据呢？

combineLatest：这个操作符有很多方法重载，我们这里用到的是接收多个流作为参数的方法。这里先不讲，直接看效果。

combineLatest(timer$, input$).pipe(
  tap(console.log)
).subscribe();

我们观察控制台，发现一开始什么输出都没有，按理说定时器流中的 startWith 操作符应该会流出事件啊。当我们在 input 输入框输入数据时，控制台终于有了输出。再点击各种按钮试试，发现规律了吗？

combineLatest 是符如其名，组合流中最后的事件。意思是（以这里的例子为例）：

首先文本输入流和定时器流都得有事件流出。
combineLatest 捕获是两个事件流中的最新值，如果文本输入流有新值，那么将输出 [定时器流最后一个值，文本输入流新值]；如果定时器流有新值，将输出 [定时器流新值，文本输入流最后一个值]。因此，只要任意一个流有新值产生，combineLatest 就会有输出。

因此，一开始我们的定时器流中有值，但文本流没有值，所以没有输出，这符合第一点。然后，当我们开始在文本框输入时，有值输出；当我们点击定时器按钮开始计时时，控制台将会以定时器的频率持续输出，并且输入肯定是两个流中的最新值，或者说是最后那个值。这符合第二点。

我们看到 combineLatest 操作符以数组的方式组合了各个流中的数据，一般来说我们肯定要对这些数据进行加工产生新的数据类型，比如对象啊，文本啊，可以在接下来使用 map 操作符进行数据变换。其实 combineLatest 的重载为我们提供了更方便的变换数据的方式，传入额外的函数参数，这个函数接收各个流中的值作为输入参数，返回值作为下一个操作符操作流中的值。使用方式如下：

combineLatest(
  timer$,
  input$,
  (timeValue, inputValue) => ({count: timeValue, input: inputValue}) // 下一个操作符操作的值就为一个对象，包含两个属性
)

完整代码如下：

import React, { useRef, useEffect, useState } from "react";

import { fromEvent, interval, merge, combineLatest } from "rxjs";
import {
  takeUntil,
  switchMap,
  scan,
  startWith,
  mapTo,
  tap,
  map
} from "rxjs/operators";

export default function App() {
  const [txt, setTxt] = useState("");

  const pauseBtnRef = useRef(null);
  const startBtnRef = useRef(null);
  const resetBtnRef = useRef(null);
  const halfBtnRef = useRef(null);
  const quarterBtnRef = useRef(null);
  const inputRef = useRef(null);

  interface Count {
    count: number;
  }

  const addOne = (acc: Count) => ({ count: acc.count + 1 });
  const reset = (acc: Count) => ({ count: 0 });

  useEffect(() => {
    const pauseBtnClick$ = fromEvent(pauseBtnRef.current, "click");
    const startBtnClick$ = fromEvent(startBtnRef.current, "click");
    const resetBtnClick$ = fromEvent(resetBtnRef.current, "click");
    const halfBtnClick$ = fromEvent(halfBtnRef.current, "click");
    const quarterBtnClick$ = fromEvent(quarterBtnRef.current, "click");

    const addOneOrReset = (time = 1000) =>
      merge(
        interval(time).pipe(
          takeUntil(pauseBtnClick$),
          mapTo(addOne)
        ),
        resetBtnClick$.pipe(mapTo(reset))
      );
    const time$ = merge(
      startBtnClick$.pipe(mapTo(1000)),
      halfBtnClick$.pipe(mapTo(500)),
      quarterBtnClick$.pipe(mapTo(250))
    );

    const input$ = fromEvent(inputRef.current, "input").pipe(
      map(e => e.target.value)
    );

    const timer$ = time$.pipe(
      switchMap(addOneOrReset),
      startWith({ count: 0 }),
      scan((acc, current) => current(acc)),
      map(obj => obj.count),
      tap(v => setTxt(v))
    );

    const subscription = combineLatest(
      timer$,
      input$,
      (timeValue, inputValue) => ({count: timeValue, input: inputValue})
    )
      .pipe(tap(console.log))
      .subscribe();

    return () => {
      subscription.unsubscribe();
    };
  }, []);

  return (
    <div className="App">
      <div style={{ fontSize: "30px" }}>{txt}</div>
      <button ref={startBtnRef}>开始</button>
      <button ref={pauseBtnRef}>暂停</button>
      <button ref={resetBtnRef}>重置</button>
      <button ref={halfBtnRef}>1/2秒</button>
      <button ref={quarterBtnRef}>1/4秒</button>
      <div>
        <input type="text" ref={inputRef} />
      </div>
    </div>
  );
}

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