本文是一篇 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'));其他更多语法差别可以参考：
Let’s start
如上所说，我们要运用 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访问到。
初始化 DOM 事宜流
在index.html中编写一个input，我们将在index.js中，经由历程 RxJS 的 Observable 监听input的keyup事宜。可以运用fromEvent来竖立一个基于 DOM 事宜的流，并经由历程map和filter进一步处置惩罚。
<!-- index.html -->
<input class="search" type="text" maxlength="1000" required placeholder="search in github"/>// 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变量来推断是不是正在异步，假如正在异步，则不举行新的异步；而我们更愿望的是可以作废旧的异步，只处置惩罚新的异步要求。
越发文雅的 Rx 作风
根据上面的教程，我们在 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事宜竖立一个流。用户的详细资料是异步加载的，而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 ！
APIS
栗子中运用到的 RxJS API：
from经由历程一个可迭代对象来竖立流fromEvent经由历程 DOM 事宜来竖立流debounce假如在肯定时候内流中的某个事宜不停被触发，则不会举行以后的事宜操纵map遍历流中一切事宜，返回新的流filter挑选流中一切事宜，返回新的流flatMap对各个事宜返回的值举行处置惩罚并返回 Observable，然后将一切的 Observable 扁平化，成为一个新的 ObservableflatMapLatest对各个事宜返回的值举行处置惩罚并返回 Observable，然后将一切的 Observable 扁平化，成为一个新的 Observable。但只会猎取末了一次返回的 Observable，其他的返回结果不予处置惩罚distinctUntilChanged流中假如相邻事宜的结果一样，则仅挑选出一个（剔除反复值）do可以顺次拿到流上每一个事宜的返回值，运用其做一些无关流通报的事变takeWhile赋予流一个推断，只需当takeWhile中的回调返回true时，流才会继承实行；否则将中缀以后的事宜
