探索 RxJS - 做一个 github 小应用

摘要：更加优雅的风格按照上面的教程，我们在中获取到了数据发送异步请求并拿到了最新一次的返回值。之后，再通过，在监听的回调中将返回值拼接成并插入。

本文是一篇 RxJS 实战教程，利用 RxJS 和 github API 来一步步做一个 github 小应用。因此，文章的重点是解释 RxJS 的使用，而涉及的 ES6语法、webpack 等知识点不予讲解。

本例的所有代码在 github 仓库：rxjs-example

首先要注意的是，目前在 github 上有两个主流 RxJS，它们代表不同的版本：

ReactiveX - rxjs RxJS 5 beta 版

Reactive-Extensions - RxJS RxJS 4.x 稳定版

这两个版本的安装和引用稍有不同：

# 安装 4.x 稳定版
$ npm install rx --save
# 安装 5 beta 版
$ npm install rxjs --save

// 4.x 稳定版
import Rx from "rx";
// 5 beta 版
import Rx from "rxjs/Rx";

除此以外，它们的语法也稍有不同，比如在 5 beta 版里，subscribe时可以代入一个对象作为参数，也可以代入回调函数作为参数，而 4.x 版则只支持以回调函数为参数的情况：

// 5 beta
var observer = {
  next: x => console.log("Observer got a next value: " + x),
  error: err => console.error("Observer got an error: " + err),
  complete: () => console.log("Observer got a complete notification"),
};
Observable.subscribe(observer);

// 5 和 4.x 都支持：
Observable.subscribe(x => console.log(x), (err) => console.log(err), () => console.log("completed"));

其他更多语法不同可以参考：

4.x 稳定版 Document

5 beta 版 Document

从 4 到 5 的迁移

如上所说，我们要利用 RxJS 和 github API 来一步步做一个 github 小应用。首先完成其基本功能，即通过一个 input 输入文字，并实时根据 input 内值的变化去发送异步请求，调用 github API 进行搜索。如图所示（线上 Demo）：

通过RxJS，在输入过程中实时进行异步搜索：

hover到 avator 上之后异步获取用户信息

安装 webpack 配置编译环境，并使用 ES6 语法。安装如下依赖，并配置好 webpack：

webpack

webpack-dev-server

babel-loader

babel-preset-es2015

html-webpack-plugin

css-loader / postcss 及其他

jquery

rx（4.x 版本）

通过webpack-dev-server，我们将会启动一个 8080 端口的服务器，使得我们编译好的资源可以在localhost:8080/webpack-dev-server访问到。

在index.html中编写一个input，我们将在index.js中，通过 RxJS 的 Observable 监听input的keyup事件。可以使用fromEvent来创建一个基于 DOM 事件的流，并通过map和filter进一步处理。

// src/js/index.js
import Rx from "rx";

$(() => {
  const $input = $(".search");
  // 通过 input 的 keyup 事件来创建流
  const observable = Rx.Observable.fromEvent($input, "keyup")
      // 并获取每次 keyup 时搜索框的值，筛选出合法值
      .map(() => $input.val().trim())
    .filter((text) => !!text)
    // 利用 do 可以做一些不影响流的事件，比如这里打印出 input 的值
    .do((value) => console.log(value));
  // 开启监听
  observable.subscribe();
});

去 input 里随便打打字，可以看到我们已经成功监听了keyup事件，并在每次keyup时在 console 里输出 input 当前的值。

监听了 input 事件，我们就能够在每次keyup时拿到 value，那么就可以通过它来异步获取数据。将整个过程拆分一下：

用户在 input 里输入任意内容

触发keyup事件，获取到当前 value

将 value 代入到一个异步方法里，通过接口获取数据

利用返回数据渲染 DOM

也就是说，我们要把原有的 Observable 中每个事件返回的 value 进行异步处理，并使其返回一个新的 Observable。可以这么处理：

让每个 value 返回一个 Observable

通过flatMap将所有的 Observable 扁平化，成为一个新的 Observable

图解flatMap：

而既然需要异步获取数据，那么在上面的第一步时，可以通过fromPromise来创建一个 Observable：

// src/js/helper.js
const SEARCH_REPOS = "https://api.github.com/search/repositories?sort=stars&order=desc&q=";

// 创建一个 ajax 的 promise
const getReposPromise = (query) => {
  return $.ajax({
      type: "GET",
    url: `${SEARCH_REPOS}${query}`,
  }).promise();
};
// 通过 fromPromise 创建一个 Observable
export const getRepos = (query) => {
  const promise = getReposPromise(query);
  return Rx.Observable.fromPromise(promise);
};

// src/js/index.js
import {getRepos} from "./helper";

// ...
const observable = Rx.Observable.fromEvent($input, "keyup")
      .map(() => $input.val())
    .filter((text) => !!text)
    .do((value) => console.log(value))
    // 调用 getRepos 方法将返回一个 Observable
    // flatMap 则将所有 Observable 合并，转为一个 Observable
    .flatMap(getRepos);
// ...

这样，每一次keyup的时候，都会根据此时 input 的 value 去异步获取数据。但这样做有几个问题：

不断打字时会连续不断触发异步请求，占用资源影响体验

如果相邻的keyup事件触发时 input 的值一样，也就是说按下了不改变 value 的按键（比如方向键），会重复触发一样的异步事件

发出多个异步事件之后，每个事件所耗费的时间不一定相同。如果前一个异步所用时间较后一个长，那么当它最终返回结果时，有可能把后面的异步率先返回的结果覆盖

所以接下来我们就处理这几个问题。

针对上面的问题，一步一步进行优化。

不断打字时会连续不断触发异步请求，占用资源影响体验

也就是说，当用户在连续打字时，我们不应该继续进行之后的事件处理，而如果打字中断，或者说两次keyup事件的时间间隔足够长时，才应该发送异步请求。针对这点，可以使用 RxJS 的debounce方法：

如图所示，在一段时间内事件被不断触发时，不会被之后的操作所处理；只有超过指定时间间隔的事件才会留下来：

// src/js/index.js
// ...
const observable = Rx.Observable.fromEvent($input, "keyup")
    // 若 400ms 内连续触发 keyup 事件，则不会继续往下处理
    .debounce(400)
      .map(() => $input.val())
    .filter((text) => !!text)
    .do((value) => console.log(value))
    .flatMap(getRepos);
// ...

如果相邻的keyup事件触发时 input 的值一样，也就是说按下了不改变 value 的按键（比如方向键），会重复触发一样的异步事件

也就是说，对于任意相邻的事件，如果它们的返回值一样，则只要取一个（重复事件中的第一个）就好了。可以利用distinctUntilChanged方法：

// src/js/index.js
// ...
const observable = Rx.Observable.fromEvent($input, "keyup")
    .debounce(400)
      .map(() => $input.val())
    .filter((text) => !!text)
    // 只取不一样的值进行异步
    .distinctUntilChanged()
    .do((value) => console.log(value))
    .flatMap(getRepos);
// ...

发出多个异步事件之后，每个事件所耗费的时间不一定相同。如果前一个异步所用时间较后一个长，那么当它最终返回结果时，有可能把后面的异步率先返回的结果覆盖

这个蛋疼的问题我相信大家很可能遇见过。在发送多个异步请求时，因为所用时长不一定，无法保障异步返回的先后顺序，所以，有时候可能早请求的异步的结果会覆盖后来请求的异步结果。

而这种情况的处理方式就是，在连续发出多个异步的时候，既然我们期待的是最后一个异步返回的结果，那么就可以把之前的异步取消掉，不 care 其返回了什么。因此，我们可以使用flatMapLatest API（类似于 RxJava 中的switchMap API，同时在 RxJS 5.0 中也已经改名为switchMap）

通过flatMapLatest，当 Observable 触发某个事件，返回新的 Observable 时，将取消之前触发的事件，并且不再关心返回结果的处理，只监视当前这一个。也就是说，发送多个请求时，不关心之前请求的处理，只处理最后一次的请求：

// src/js/index.js
// ...
const observable = Rx.Observable.fromEvent($input, "keyup")
    .debounce(400)
      .map(() => $input.val())
    .filter((text) => !!text)
    .distinctUntilChanged()
    .do((value) => console.log(value))
    // 仅处理最后一次的异步
    .flatMapLatest(getRepos);
// ...

至此，我们对 input keyup以及异步获取数据的整个事件流处理完毕，并进行了一定的优化，避免了过多的请求、异步返回结果错乱等问题。但创建了一个流之后也有对其进行监听：

// src/js/index.js
// ...
const observable = Rx.Observable.fromEvent($input, "keyup")
    .debounce(400)
      .map(() => $input.val())
    .filter((text) => !!text)
    .distinctUntilChanged()
    .do((value) => console.log(value))
    .flatMapLatest(getRepos);
// 第一个回调中的 data 代表异步的返回值
observable.subscribe((data) => {
  // 在 showNewResults 方法中使用返回值渲染 DOM
  showNewResults(data);
}, (err) => {
  console.log(err);
}, () => {
  console.log("completed");
});

// 异步返回的结果是个 Array，代表搜索到的各个仓库 item
// 遍历所有 item，转化为 jQuery 对象，最后插入到 content_container 中
const showNewResults = (items) => {
  const repos = items.map((item, i) => {
    return reposTemplate(item);
  }).join("");
  $(".content_container").html(repos);
};

这样，一个通过 RxJS 监听事件的流已经完全建立完毕了。整个过程使用图像来表示则如下：

而如果我们不使用 RxJS，用传统方式监听 input 的话：

// src/js/index.js
import {getRepos} from "./helper";

$(() => {
  const $input = $(".search");
  const interval = 400;
  var previousValue = null;
  var fetching = false;
  var lastKeyUp = Date.now() - interval;
  $input.on("keyup", (e) => {
    const nextValue = $input.val();
    if (!nextValue) {
      return;
    }
    if (Date.now() - lastKeyUp <= interval) {
      return;
    }
    lastKeyUp = Date.now();
    if (nextValue === previousValue) {
      return;
    }
    previousValue = nextValue;
    if (!fetching) {
      fetching = true;
      getRepos(nextValue).then((data) => {
          fetching = false;
        showNewResults(data);
      });
    }
  });
});

挺复杂了吧？而且即便如此，这样的处理还是不够到位。上面仅仅是通过fetching变量来判断是否正在异步，如果正在异步，则不进行新的异步；而我们更希望的是能够取消旧的异步，只处理新的异步请求。

按照上面的教程，我们在 Observable 中获取到了数据、发送异步请求并拿到了最新一次的返回值。之后，再通过subscribe，在监听的回调中将返回值拼接成 HTML 并插入 DOM。

但是有一个问题：小应用的另一个功能是，当鼠标hover到头像上时，异步获取并展现用户的信息。可是用户头像是在subscribe回调中动态插入的，又该如何创建事件流呢？当然了，可以在每次插入 DOM 之后在利用fromEvent创建一个基于hover的事件流，但那样总是不太好的，写出来的代码也不够 Rx。或许我们就不应该在.flatMapLatest(getRepos)之后中断流的传递？但那样的话，又该如何把异步的返回值插入 DOM 呢？

针对这种情况，我们可以使用 RxJS 的do方法：

你想在do的回调内做什么都可以，它不会影响到流内的事件；除此以外，还可以拿到流中各个事件的返回值：

var observable = Rx.Observable.from([0, 1, 2])
    .do((x) => console.log(x))
    .map((x) => x + 1);
observable.subscribe((x) => {
  console.log(x);
});

所以，我们可以利用do来完成 DOM 的渲染：

// src/js/index.js
// ...
// $conatiner 是装载搜索结果的容器 div
const $conatiner = $(".content_container");

const observable = Rx.Observable.fromEvent($input, "keyup")
    .debounce(400)
      .map(() => $input.val())
    .filter((text) => !!text)
    .distinctUntilChanged()
    .do((value) => console.log(value))
    .flatMapLatest(getRepos)
    // 首先把之前的搜索结果清空
    .do((results) => $conatiner.html(""))
    // 利用 Rx.Observable.from 将异步的结果转化为 Observable，并通过 flatMap 合并到原有的流中。此时流中的每个元素是 results 中的每个 item
    .flatMap((results) => Rx.Observable.from(results))
    // 将各 item 转化为 jQuery 对象
    .map((repos) => $(reposTemplate(repos)))
    // 最后把每个 jQuery 对象依次加到容器里
    .do(($repos) => {
      $conatiner.append($repos);
    });

// 在 subscribe 中实际上什么都不用做，就能达到之前的效果
observable.subscribe(() => {
  console.log("success");
}, (err) => {
  console.log(err);
}, () => {
  console.log("completed");
});

简直完美！现在我们这个observable在最后通过map，依次返回了一个 jQuery 对象。那么之后如果要对头像添加hover的监听，则可以在这个流的基础上继续进行。

我们接下来针对用户头像的hover事件创建一个流。用户的详细资料是异步加载的，而hover到头像上时弹出 modal。如果是第一个hover，则 modal 里只有一个 loading 的图标，并且异步获取数据，之后将返回的数据插入到 modal 里；而如果已经拿到并插入好了数据，则不再有异步请求，直接展示：

没有数据时展示 loading，同时异步获取数据

异步返回后插入数据。且如果已经有了数据则直接展示

先不管上一个流，我们先创建一个新的事件流：

// src/js/index.js
// ...
const initialUserInfoSteam = () => {
  const $avator = $(".user_header");
  // 通过头像 $avator 的 hover 事件来创建流
  const avatorMouseover = Rx.Observable.fromEvent($avator, "mouseover")
    // 500ms 内重复触发事件则会被忽略
    .debounce(500)
    // 只有当满足了下列条件的流才会继续执行，否则将中断
    .takeWhile((e) => {
      // 异步获取的用户信息被新建到 DOM 里，该 DOM 最外层是 infos_container
      // 因此，如果已经有了 infos_container，则可以认为我们已经异步获取过数据了，此时 takeWhile 将返回 false，流将会中断
      const $infosWrapper = $(e.target).parent().find(".user_infos_wrapper");
      return $infosWrapper.find(".infos_container").length === 0;
    })
    .map((e) => {
      const $infosWrapper = $(e.target).parent().find(".user_infos_wrapper");
      return {
        conatiner: $infosWrapper,
        url: $(e.target).attr("data-api")
      }
    })
    .filter((data) => !!data.url)
    // getUser 来异步获取用户信息
    .flatMapLatest(getUser)
    .do((result) => {
      // 将用户信息组建成为 DOM 元素，并插入到页面中。在这之后，该用户对应的 DOM 里就会拥有 infos_container 这个 div，所以 takeWhile 会返回 false。也就是说，之后再 hover 上去，流也不会被触发了
      const {data, conatiner} = result;
      showUserInfo(conatiner, data);
    });

  avatorMouseover.subscribe((result) => {
      console.log("fetch user info succeed");
  }, (err) => {
    console.log(err);
  }, () => {
    console.log("completed");
  });
};

上面的代码中有一个 API 需要讲解：takeWhile

由图可知，当takeWhile中的回调返回true时，流可以正常进行；而一旦返回false，则之后的事件不会再发生，流将直接终止：

var source = Rx.Observable.range(1, 5)
    .takeWhile(function (x) { return x < 3; });

var subscription = source.subscribe(
    function (x) { console.log("Next: " + x); },
    function (err) { console.log("Error: " + err); },
    function () { console.log("Completed"); });
// Next: 0
// Next: 1
// Next: 2
// Completed

创建好针对hover的事件流，我们可以把它和上一个事件流结合起来：

// src/js/index.js
// ...
const initialUserInfoSteam = ($repos) => {
  const $avator = $repos.find(".user_header");
  // ...
}

const observable = Rx.Observable.fromEvent($input, "keyup")
    // ...
    .do(($repos) => {
      $conatiner.append($repos);
      initialUserInfoSteam($repos);
    });
// ...

现在这样就已经可以使用了，但依旧不够好。目前总共有两个流：监听 input keyup的流和监听mouseover的流。但是，因为用户头像是动态插入的 ，所以我们必须在$conatiner.append($repos);之后才能创建并监听mouseover。不过鉴于我们已经在最后的do方法里插入了获取的数据，所以可以试着把两个流合并到一起：

// src/js/index.js
// ...
const initialUserInfoSteam = ($repos) => {
  const $avator = $repos.find(".user_header");
  const avatorMouseover = Rx.Observable.fromEvent($avator, "mouseover")
  // ... 流的处理跟之前的一样
  // 但我们不再需要 subscribe 它，而是返回这个 Observable
  return avatorMouseover;
};

const observable = Rx.Observable.fromEvent($input, "keyup")
    // ...
    .do(($repos) => {
      $conatiner.append($repos);
      // 不再在 do 里面创建新的流并监听
      // initialUserInfoSteam($repos);
    })
    // 相反，我们继续这个流的传递，只是通过 flatMap 将原来的流变成了监听 mouseover 的流
    .flatMap(($repos) => {
      return initialUserInfoSteam($repos);
    });
// ...

DONE ！

栗子中使用到的 RxJS API：

from 通过一个可迭代对象来创建流

fromEvent 通过 DOM 事件来创建流

debounce 如果在一定时间内流中的某个事件不断被触发，则不会进行之后的事件操作

map 遍历流中所有事件，返回新的流

filter 筛选流中所有事件，返回新的流

flatMap 对各个事件返回的值进行处理并返回 Observable，然后将所有的 Observable 扁平化，成为一个新的 Observable

flatMapLatest 对各个事件返回的值进行处理并返回 Observable，然后将所有的 Observable 扁平化，成为一个新的 Observable。但只会获取最后一次返回的 Observable，其他的返回结果不予处理

distinctUntilChanged 流中如果相邻事件的结果一样，则仅筛选出一个（剔除重复值）

do 可以依次拿到流上每个事件的返回值，利用其做一些无关流传递的事情

takeWhile 给予流一个判断，只有当takeWhile中的回调返回true时，流才会继续执行；否则将中断之后的事件

What does RxJS observable debounce do

How do I work with jQuery and RxJS

Introduction of observable operators

文章源码 - rxjs-example