构建流式应用—RxJS详解

Question

构建流式应用—RxJS详解

joeyguo opened this issue 8 years ago · 24 comments

最近在 Alloyteam Conf 2016 分享了《使用RxJS构建流式前端应用》，会后在线上线下跟大家交流时发现对于 RxJS 的态度呈现出两大类:有用过的都表达了 RxJS 带来的优雅编码体验，未用过的则反馈太难入门。所以，这里将结合自己对 RxJS 理解，通过 RxJS 的实现原理、基础实现及实例来一步步分析，提供 RxJS 较为全面的指引，感受下使用 RxJS 编码是怎样的体验。

常规方式实现搜索功能
RxJS · 流 Stream
RxJS 实现原理简析
- 观察者模式
- 迭代器模式
- RxJS 的观察者 + 迭代器模式
RxJS 基础实现
- Observable
- Observer
RxJS · Operators
- Operators ·入门
- 一系列的 Operators 操作
使用 RxJS 一步步实现搜索功能
总结

常规方式实现搜索

做一个搜索功能在前端开发中其实并不陌生，一般的实现方式是：监听文本框的输入事件，将输入内容发送到后台，最终将后台返回的数据进行处理并展示成搜索结果。

<input id="text"></input>
<script>
    var text = document.querySelector('#text');
    text.addEventListener('keyup', (e) =>{
        var searchText = e.target.value;
        // 发送输入内容到后台
        $.ajax({
            url: `search.qq.com/${searchText}`,
            success: data => {
              // 拿到后台返回数据，并展示搜索结果
              render(data);
            }
        });
    });
</script>

上面代码实现我们要的功能，但存在两个较大的问题：

多余的请求
当想搜索“爱迪生”时，输入框可能会存在三种情况，“爱”、“爱迪”、“爱迪生”。而这三种情况将会发起 3 次请求，存在 2 次多余的请求。
已无用的请求仍然执行
一开始搜了“爱迪生”，然后马上改搜索“达尔文”。结果后台返回了“爱迪生”的搜索结果，执行渲染逻辑后结果框展示了“爱迪生”的结果，而不是当前正在搜索的“达尔文”，这是不正确的。

减少多余请求数，可以用 setTimeout 函数节流的方式来处理，核心代码如下

<input id="text"></input>
<script>
    var text = document.querySelector('#text'),
        timer = null;
    text.addEventListener('keyup', (e) =>{
        // 在 250 毫秒内进行其他输入，则清除上一个定时器
        clearTimeout(timer);
        // 定时器，在 250 毫秒后触发
        timer = setTimeout(() => {
            console.log('发起请求..');
        },250)
    })
</script>

已无用的请求仍然执行的解决方式，可以在发起请求前声明一个当前搜索的状态变量，后台将搜索的内容及结果一起返回，前端判断返回数据与当前搜索是否一致，一致才走到渲染逻辑。最终代码为

<input id="text"></input>
<script>
    var text = document.querySelector('#text'),
        timer = null,
        currentSearch = '';

    text.addEventListener('keyup', (e) =>{
        clearTimeout(timer)
        timer = setTimeout(() => {
            // 声明一个当前所搜的状态变量
            currentSearch ＝ '书'; 

            var searchText = e.target.value;
            $.ajax({
                url: `search.qq.com/${searchText}`,
                success: data => {
                    // 判断后台返回的标志与我们存的当前搜索变量是否一致
                    if (data.search === currentSearch) {
                        // 渲染展示
                        render(data);
                    } else {
                        // ..
                    }
                }           
            });
        },250)
    })
</script>

上面代码基本满足需求，但代码开始显得乱糟糟。我们来使用 RxJS 实现上面代码功能，如下

var text = document.querySelector('#text');
var inputStream = Rx.Observable.fromEvent(text, 'keyup')
                    .debounceTime(250)
                    .pluck('target', 'value')
                    .switchMap(url => Http.get(url))
                    .subscribe(data => render(data));

可以明显看出，基于 RxJS 的实现，代码十分简洁！

RxJS · 流 Stream

RxJS 是 Reactive Extensions for JavaScript 的缩写，起源于 Reactive Extensions，是一个基于可观测数据流在异步编程应用中的库。RxJS 是 Reactive Extensions 在 JavaScript 上的实现，而其他语言也有相应的实现，如 RxJava、RxAndroid、RxSwift 等。学习 RxJS，我们需要从可观测数据流(Streams)说起，它是 Rx 中一个重要的数据类型。

流是在时间流逝的过程中产生的一系列事件。它具有时间与事件响应的概念。

下雨天时，雨滴随时间推移逐渐产生，下落时对水面产生了水波纹的影响，这跟 Rx 中的流是很类似的。而在 Web 中，雨滴可能就是一系列的鼠标点击、键盘点击产生的事件或数据集合等等。

RxJS 基础实现原理简析

对流的概念有一定理解后，我们来讲讲 RxJS 是怎么围绕着流的概念来实现的，讲讲 RxJS 的基础实现原理。RxJS 是基于观察者模式和迭代器模式以函数式编程思维来实现的。

观察者模式

观察者模式在 Web 中最常见的应该是 DOM 事件的监听和触发。

订阅：通过 addEventListener 订阅 document.body 的 click 事件。
发布：当 body 节点被点击时，body 节点便会向订阅者发布这个消息。

document.body.addEventListener('click', function listener(e) {
    console.log(e);
},false);

document.body.click(); // 模拟用户点击

将上述例子抽象模型，并对应通用的观察者模型

迭代器模式

迭代器模式可以用 JavaScript 提供了 Iterable Protocol 可迭代协议来表示。Iterable Protocol 不是具体的变量类型，而是一种可实现协议。JavaScript 中像 Array、Set 等都属于内置的可迭代类型，可以通过 iterator 方法来获取一个迭代对象，调用迭代对象的 next 方法将获取一个元素对象，如下示例。

var iterable = [1, 2];

var iterator = iterable[Symbol.iterator]();

iterator.next(); // => { value: "1", done: false}
iterator.next(); // => { value: "2", done: false}

iterator.next(); // => { value: undefined, done: true}

元素对象中：value 表示返回值，done 表示是否已经到达最后。

遍历迭代器可以使用下面做法。

var iterable = [1, 2];
var iterator = iterable[Symbol.iterator]();

while(true) {
    let result;
    try {
        result = iterator.next();  // <= 获取下一个值
    } catch (err) {
        handleError(err);  // <= 错误处理
    }
    if (result.done) {
        handleCompleted();  // <= 无更多值（已完成）
        break;
    }
    doSomething(result.value);
}

主要对应三种情况：

获取下一个值
调用 next 可以将元素一个个地返回，这样就支持了返回多次值。
无更多值(已完成)
当无更多值时，next 返回元素中 done 为 true。
错误处理
当 next 方法执行时报错，则会抛出 error 事件，所以可以用 try catch 包裹 next 方法处理可能出现的错误。

RxJS 的观察者 + 迭代器模式

RxJS 中含有两个基本概念：Observables 与 Observer。Observables 作为被观察者，是一个值或事件的流集合；而 Observer 则作为观察者，根据 Observables 进行处理。
Observables 与 Observer 之间的订阅发布关系(观察者模式) 如下：

订阅：Observer 通过 Observable 提供的 subscribe() 方法订阅 Observable。
发布：Observable 通过回调 next 方法向 Observer 发布事件。

下面为 Observable 与 Observer 的伪代码

// Observer
var Observer = {
    next(value) {
        alert(`收到${value}`);
    }
};

// Observable
function Observable (Observer) {
    setTimeout(()=>{
        Observer.next('A');
    },1000)
}

// subscribe
Observable(Observer);

上面实际也是观察者模式的表现，那么迭代器模式在 RxJS 中如何体现呢？

在 RxJS 中，Observer 除了有 next 方法来接收 Observable 的事件外，还可以提供了另外的两个方法：error() 和 complete()，与迭代器模式一一对应。

var Observer = {
    next(value) { /* 处理值*/ },
    error(error) { /* 处理异常 */ },
    complete() { /* 处理已完成态 */ }
};

结合迭代器 Iterator 进行理解：

next()
Observer 提供一个 next 方法来接收 Observable 流，是一种 push 形式；而 Iterator 是通过调用 iterator.next() 来拿到值，是一种 pull 的形式。
complete()
当不再有新的值发出时，将触发 Observer 的 complete 方法；而在 Iterator 中，则需要在 next 的返回结果中，当返回元素 done 为 true 时，则表示 complete。
error()
当在处理事件中出现异常报错时，Observer 提供 error 方法来接收错误进行统一处理；Iterator 则需要进行 try catch 包裹来处理可能出现的错误。

下面是 Observable 与 Observer 实现观察者 + 迭代器模式的伪代码，数据的逐渐传递传递与影响其实就是流的表现。

// Observer
var Observer = {
    next(value) {
        alert(`收到${value}`);
    },
    error(error) {
        alert(`收到${error}`);
    },
    complete() {
        alert("complete");
    },
};

// Observable
function Observable (Observer) {
    [1,2,3].map(item=>{
        Observer.next(item);
    });

    Observer.complete();
    // Observer.error("error message");
}

// subscribe
Observable(Observer);

RxJS 基础实现

有了上面的概念及伪代码，那么在 RxJS 中是怎么创建 Observable 与 Observer 的呢?

创建 Observable

RxJS 提供 create 的方法来自定义创建一个 Observable，可以使用 next 来发出流。

var Observable = Rx.Observable.create(observer => {
    observer.next(2);
    observer.complete();
    return  () => console.log('disposed');
});

创建 Observer

Observer 可以声明 next、err、complete 方法来处理流的不同状态。

var Observer = Rx.Observer.create(
    x => console.log('Next:', x),
    err => console.log('Error:', err),
    () => console.log('Completed')
);

最后将 Observable 与 Observer 通过 subscribe 订阅结合起来。

var subscription = Observable.subscribe(Observer);

RxJS 中流是可以被取消的，调用 subscribe 将返回一个 subscription，可以通过调用 subscription.unsubscribe() 将流进行取消，让流不再产生。

看了起来挺复杂的？换一个实现形式：

// @Observables 创建一个 Observables
var streamA = Rx.Observable.of(2);

// @Observer streamA$.subscribe(Observer)
streamA.subscribe(v => console.log(v));

将上面代码改用链式写法，代码变得十分简洁：

Rx.Observable.of(2).subscribe(v => console.log(v));

RxJS · Operators 操作

Operators 操作·入门

Rx.Observable.of(2).subscribe(v => console.log(v));

上面代码相当于创建了一个流(2)，最终打印出2。那么如果想将打印结果翻倍，变成4，应该怎么处理呢？

方案一?：改变事件源，让 Observable 值 X 2

Rx.Observable.of(2 * 2 /* <= */).subscribe(v => console.log(v));

方案二?：改变响应方式，让 Observer 处理 X 2

Rx.Observable.of(2).subscribe(v => console.log(v * 2 /* <= */));

优雅方案： RxJS 提供了优雅的处理方式，可以在事件源(Observable)与响应者(Observer)之间增加操作流的方法。

Rx.Observable.of(2)
             .map(v => v * 2) /* <= */
             .subscribe(v => console.log(v));

map 操作跟数组操作的作用是一致的，不同的这里是将流进行改变，然后将新的流传出去。在 RxJS 中，把这类操作流的方式称之为 Operators(操作)。RxJS提供了一系列 Operators，像map、reduce、filter 等等。操作流将产生新流，从而保持流的不可变性，这也是 RxJS 中函数式编程的一点体现。关于函数式编程，这里暂不多讲，可以看看另外一篇文章《谈谈函数式编程》

到这里，我们知道了，流从产生到最终处理，可能经过的一些操作。即 RxJS 中 Observable 将经过一系列 Operators 操作后，到达 Observer。

          Operator1   Operator2
Observable ----|-----------|-------> Observer

一系列的 Operators 操作

RxJS 提供了非常多的操作，像下面这些。

Aggregate,All,Amb,ambArray,ambWith,AssertEqual,averageFloat,averageInteger,averageLong,blocking,blockingFirst,blockingForEach,blockingSubscribe,Buffer,bufferWithCount,bufferWithTime,bufferWithTimeOrCount,byLine,cache,cacheWithInitialCapacity,case,Cast,Catch,catchError,catchException,collect,concatWith,Connect,connect_forever,cons,Contains,doAction,doAfterTerminate,doOnComplete,doOnCompleted,doOnDispose,doOnEach,doOnError,doOnLifecycle,doOnNext,doOnRequest,dropUntil,dropWhile,ElementAt,ElementAtOrDefault,emptyObservable,fromNodeCallback,fromPromise,fromPublisher,fromRunnable,Generate,generateWithAbsoluteTime,generateWithRelativeTime,Interval,intervalRange,into,latest (Rx.rb version of Switch),length,mapTo,mapWithIndex,Materialize,Max,MaxBy,mergeArray,mergeArrayDelayError,mergeWith,Min,MinBy,multicastWithSelector,nest,Never,Next,Next (BlockingObservable version),partition,product,retryWhen,Return,returnElement,returnValue,runAsync,safeSubscribe,take_with_time,takeFirst,TakeLast,takeLastBuffer,takeLastBufferWithTime,windowed,withFilter,withLatestFrom,zipIterable,zipWith,zipWithIndex

关于每一个操作的含义，可以查看官网进行了解。operators 具有静态（static）方法和实例（ instance）方法，下面使用 Rx.Observable.xx 和 Rx.Observable.prototype.xx 来简单区分，举几个例子。

Rx.Observable.of
of 可以将普通数据转换成流式数据 Observable。如上面的 Rx.Observable.of(2)。

Rx.Observable.fromEvent
除了数值外，RxJS 还提供了关于事件的操作，fromEvent 可以用来监听事件。当事件触发时，将事件 event 转成可流动的 Observable 进行传输。下面示例表示：监听文本框的 keyup 事件，触发 keyup 可以产生一系列的 event Observable。

var text = document.querySelector('#text');
Rx.Observable.fromEvent(text, 'keyup')
             .subscribe(e => console.log(e));

Rx.Observable.prototype.map
map 方法跟我们平常使用的方式是一样的，不同的只是这里是将流进行改变，然后将新的流传出去。上面示例已有涉及，这里不再多讲。

Rx.Observable.of(2)
             .map(v => 10 * v)
             .subscribe(v => console.log(v));

Rx 提供了许多的操作，为了更好的理解各个操作的作用，我们可以通过一个可视化的工具 marbles 图来辅助理解。如 map 方法对应的 marbles 图如下

箭头可以理解为时间轴，上面的数据经过中间的操作，转变成下面的模样。

Rx.Observable.prototype.mergeMap
mergeMap 也是 RxJS 中常用的接口，我们来结合 marbles 图(flatMap(alias))来理解它

上面的数据流中，产生了新的分支流(流中流)，mergeMap 的作用则是将分支流调整回主干上，最终分支上的数据流都经过主干的其他操作，其实也是将流中流进行扁平化。

Rx.Observable.prototype.switchMap
switchMap 与 mergeMap 都是将分支流疏通到主干上，而不同的地方在于 switchMap 只会保留最后的流，而取消抛弃之前的流。

除了上面提到的 marbles，也可以 ASCII 字符的方式来绘制可视化图表，下面将结合 Map、mergeMap 和 switchMap 进行对比来理解。

@Map             @mergeMap            @switchMap
                         ↗  ↗                 ↗  ↗
-A------B-->           a2 b2                a2 b2  
-2A-----2B->          /  /                 /  /  
                    /  /                 /  /
                  a1 b1                a1 b1
                 /  /                 /  /
                -A-B----------->     -A-B---------->
                --a1-b1-a2-b2-->     --a1-b1---b2-->

mergeMap 和 switchMap 中，A 和 B 是主干上产生的流，a1、a2 为 A 在分支上产生，b1、b2 为 B 在分支上产生，可看到，最终将归并到主干上。switchMap 只保留最后的流，所以将 A 的 a2 抛弃掉。

Rx.Observable.prototype.debounceTime
debounceTime 操作可以操作一个时间戳 TIMES，表示经过 TIMES 毫秒后，没有流入新值，那么才将值转入下一个操作。

RxJS 中的操作符是满足我们以前的开发思维的，像 map、reduce 这些。另外，无论是 marbles 图还是用 ASCII 字符图这些可视化的方式，都对 RxJS 的学习和理解有非常大的帮助。

使用 RxJS 一步步实现搜索示例

RxJS 提供许多创建流或操作流的接口，应用这些接口，我们来一步步将搜索的示例进行 Rx 化。

使用 RxJS 提供的 fromEvent 接口来监听我们输入框的 keyup 事件，触发 keyup 将产生 Observable。

var text = document.querySelector('#text');
Rx.Observable.fromEvent(text, 'keyup')
             .subscribe(e => console.log(e));

这里我们并不想输出事件，而想拿到文本输入值，请求搜索，最终渲染出结果。涉及到两个新的 Operators 操作，简单理解一下：

Rx.Observable.prototype.pluck('target', 'value')
将输入的 event，输出成 event.target.value。
Rx.Observable.prototype.mergeMap()
将请求搜索结果输出回给 Observer 上进行渲染。

var text = document.querySelector('#text');
Rx.Observable.fromEvent(text, 'keyup')
             .pluck('target', 'value') // <--
             .mergeMap(url => Http.get(url)) // <--
             .subscribe(data => render(data))

上面代码实现了简单搜索呈现，但同样存在一开始提及的两个问题。那么如何减少请求数，以及取消已无用的请求呢？我们来了解 RxJS 提供的其他 Operators 操作，来解决上述问题。

Rx.Observable.prototype.debounceTime(TIMES)
表示经过 TIMES 毫秒后，没有流入新值，那么才将值转入下一个环节。这个与前面使用 setTimeout 来实现函数节流的方式有一致效果。
Rx.Observable.prototype.switchMap()
使用 switchMap 替换 mergeMap，将能取消上一个已无用的请求，只保留最后的请求结果流，这样就确保处理展示的是最后的搜索的结果。

最终实现如下，与一开始的实现进行对比，可以明显看出 RxJS 让代码变得十分简洁。

var text = document.querySelector('#text');
Rx.Observable.fromEvent(text, 'keyup')
             .debounceTime(250) // <- throttling behaviour
             .pluck('target', 'value')
             .switchMap(url => Http.get(url)) // <- Kill the previous requests
             .subscribe(data => render(data))

总结

本篇作为 RxJS 入门篇到这里就结束，关于 RxJS 中的其他方面内容，后续再拎出来进一步分析学习。
RxJS 作为一个库，可以与众多框架结合使用，但并不是每一种场合都需要使用到 RxJS。复杂的数据来源，异步多的情况下才能更好凸显 RxJS 作用，这一块可以看看民工叔写的《流动的数据——使用 RxJS 构造复杂单页应用的数据逻辑》相信会有更好的理解。

附:
RxJS(JavaScript) https://github.com/Reactive-Extensions/RxJS
RxJS(TypeScript ) https://github.com/ReactiveX/rxjs

查看更多文章 >>
https://github.com/joeyguo/blog

freeozyl80 commented 8 years ago

Great

wangtun commented 8 years ago

filter吧

xuesea commented 7 years ago

mark

ovaldi commented 7 years ago

mark

Answer 1 · 2016-11-02T02:07:30.000Z

最近刚开始看RxJS。也觉得有点懵逼。

博主的文章很清晰。谢谢分享，期待后续之作。

Answer 2 · 2016-11-02T02:44:38.000Z

nice article

Answer 3 · 2016-11-03T01:15:46.000Z

小鹏前来报到

Answer 4 · 2016-11-03T06:22:58.000Z

看完文档再来看这个，透彻多了

Answer 5 · 2016-11-15T03:29:28.000Z

写得很好，感谢博主！

Answer 6 · 2016-11-15T07:30:27.000Z

话说迭代器那边有点小错误

var iterable = [1, 2];
var iterator = iterable[Symbol.iterator]();

while(true) {
    let result;
    try {
        result = iterator.next();  // <= 获取下一个值
    } catch (err) {
        handleError(err);  // <= 错误处理
    }
    if (result.done) {
        handleCompleted();  // <= 无更多值（已完成）
        break;
    }
    doSomething(result.value);
}

感谢博主的讲解！

Answer 7 · 2016-11-16T00:42:17.000Z

nice article!

Answer 8 · 2016-12-05T02:15:18.000Z

一点补充
加上distinctUntilChanged操作符，过滤掉用户按左右键或者选择编辑但没有改变输入框中值的情况

var text = document.querySelector('#text');
Rx.Observable.fromEvent(text, 'keyup')            
             .pluck('target', 'value')
             .distinctUntilChanged() //filter left arrow , right arrow key or selection editing
             .debounceTime(250) // <- throttling behaviour
             .switchMap(url =>  Http.get(url)) // <- Kill the previous requests
             .subscribe(data => render(data))

Answer 9 · 2016-12-08T07:29:28.000Z

最近在看 RxJS，非常荣幸看到这篇文章，以及感谢作者的分享。
我按照作者的做法，发觉了一个小问题
RxJS 基础实现=》创建 Observer 那里的代码，我在使用的时候，老是报 Error: TypeError: observer.complete is not a function(…)
然后看在 chrome 看了一下 Observer 对象，发现他有个方法是 completed ，原文少了个d。

Answer 10 · 2016-12-15T04:00:02.000Z

网上大多数文章只贴出了直接使用rxjs是多么的方便，但是没有贴出来不适用rxjs的代码是怎样的。作者把两者都抛出来进行类比，一下就清楚了！感谢！

Answer 11 · 2017-01-11T09:05:18.000Z

官网的入门例子能看懂，大致**也能明白。
文中内容也还OK，看来重要的是要掌握众多的operators。
但突然跳出来的分支和主干，没能明白跟map的关系，map不是arr.map么？
我再读一遍吧。。。

Answer 12 · 2017-01-29T09:09:00.000Z

感谢大神分享~学习了很多！请问哪里能继续学习一下，文章中提到的PPT《使用RxJS构建流式前端应用》呢？谢谢！

Answer 13 · 2017-01-30T13:06:34.000Z

@Penggggg 这里哈《使用RxJS构建流式前端应用》

Answer 14 · 2017-02-18T13:36:47.000Z

那个流中流还是看的不是很明白，什么时候会出现分支流？是不是在点击发异步请求的时候

Answer 15 · 2017-02-22T05:15:04.000Z

@flyinhigh 是，异步请求时相当于发出一个新的流

Answer 16 · 2017-02-23T10:06:59.000Z

你好，请问一下.switchMap(url => Http.get(url)) // <- Kill the previous requests为什么可以取消上一个请求呢？用您写的ppt里面的例子来说：
---“成人”------------------------------------------------------------------------------
-------------------------data1(后台返回的搜索“成人”的结果)------------------
---------------“读书”------------------------------------------------------------------
----------------------------------------data2(后台返回的搜索“读书”的结果)---

那么结果应该是：?
------------------------data1--------data2-----------------------------------------

所以不太理解上面所写的能够“ Kill the previous requests”？

Answer 17 · 2017-02-26T08:31:58.000Z

@Archsx 这里其实是 Rx 中 switchMap 里所处理的：当有新的流进入时，则会抛弃前一个流。

Answer 18 · 2017-06-03T06:48:01.000Z

感谢前辈的积累，谢谢，写的非常明了。

Answer 19 · 2017-06-23T09:52:21.000Z

如何过滤掉无用的键盘事件呢？比如enter和方向键

Answer 20 · 2020-03-14T15:11:19.000Z

不错不错

目录