从 RxJS 到 Flink：如何处理数据流？

简介： 前端开发的本质是什么？响应式编程相对于 MVVM 或者 Redux 有什么优势？响应式编程的思想是否能够应用到后端开发中？本文以一个新闻网站为例，阐述在前端开发中如何使用响应式编程思想；再以计算电商平台双11每小时成交额为例，分享一样的思想在实时计算中的相同与不一样之处。

一 、前端开发在开发什么

你们在前端开发的过程当中，可能会想过这样一个问题：前端开发到底是在开发什么？在我看来，前端开发的本质是让网页视图可以正确地响应相关事件。在这句话中有三个关键字："网页视图"，"正确地响应"和"相关事件"。

"相关事件"可能包括页面点击，鼠标滑动，定时器，服务端请求等等，"正确地响应"意味着咱们要根据相关的事件来修改一些状态，而"网页视图"就是咱们前端开发中最熟悉的部分了。

按照这样的观点咱们能够给出这样 视图 = 响应函数(事件) 的公式：

View = reactionFn(Event)

在前端开发中，须要被处理事件能够归类为如下三种：

• 用户执行页面动做，例如 click, mousemove 等事件。
• 远程服务端与本地的数据交互，例如 fetch, websocket。
• 本地的异步事件，例如 setTimeout, setInterval async_event。

这样咱们的公式就能够进一步推导为：

View = reactionFn(UserEvent | Timer | Remote API)

二 、应用中的逻辑处理

为了可以更进一步理解这个公式与前端开发的关系，咱们以新闻网站举例，该网站有如下三个要求：

• 单击刷新：单击 Button 刷新数据。
• 勾选刷新：勾选 Checkbox 时自动刷新，不然中止自动刷新。
• 下拉刷新：当用户从屏幕顶端下拉时刷新数据。

若是从前端的角度分析，这三种需求分别对应着：

• 单击刷新：click -> fetch
• 勾选刷新：change -> (setInterval + clearInterval) -> fetch
• 下拉刷新：(touchstart + touchmove + touchend) -> fetch news_app

一、 MVVM

在 MVVM 的模式下，对应上文的响应函数（reactionFn）会在 Model 与 ViewModel 或者 View 与 ViewModel 之间进行被执行，而事件 (Event) 会在 View 与 ViewModel 之间进行处理。

MVVM 能够很好的抽象视图层与数据层，可是响应函数（reactionFn）会散落在不一样的转换过程当中，这会致使数据的赋值与收集过程难以进行精确追踪。另外由于事件 (Event) 的处理在该模型中与视图部分紧密相关，致使 View 与 ViewModel 之间对事件处理的逻辑复用困难。

2 、Redux

在 Redux 最简单的模型下，若干个事件 (Event) 的组合会对应到一个 Action 上，而 reducer 函数能够被直接认为与上文提到的响应函数 (reactionFn) 对应。

可是在 Redux 中：

• State 只能用于描述中间状态，而不能描述中间过程。
• Action 与 Event 的关系并不是一一对应致使 State 难以追踪实际变化来源。

3 、响应式编程与 RxJS

维基百科中是这样定义响应式编程：

在计算中，响应式编程或反应式编程（英语：Reactive programming）是一种面向数据流和变化传播的声明式编程范式。这意味着能够在编程语言中很方便地表达静态或动态的数据流，而相关的计算模型会自动将变化的值经过数据流进行传播。

以数据流维度从新考虑用户使用该应用的流程：

• 点击按钮 -> 触发刷新事件 -> 发送请求 -> 更新视图
• 勾选自动刷新
• 手指触摸屏幕
• 自动刷新间隔 -> 触发刷新事件 -> 发送请求 -> 更新视图
• 手指在屏幕上下滑
• 自动刷新间隔 -> 触发刷新事件 -> 发送请求 -> 更新视图
• 手指在屏幕上中止滑动 -> 触发下拉刷新事件 -> 发送请求 -> 更新视图
• 自动刷新间隔 -> 触发刷新事件 -> 发送请求 -> 更新视图
• 关闭自动刷新

以 Marbles 图表示：

拆分上图逻辑，就会获得使用响应式编程开发当前新闻应用时的三个步骤：

• 定义源数据流
• 组合/转换数据流
• 消费数据流并更新视图

咱们分别来进行详细描述。

定义源数据流

使用 RxJS，咱们能够很方便的定义出各类 Event 数据流。

1）单击操做

涉及 click 数据流。

click$ = fromEvent<MouseEvent>(document.querySelector('button'), 'click');

2）勾选操做

涉及 change 数据流。

change$ = fromEvent(document.querySelector('input'), 'change');

3）下拉操做

涉及 touchstart, touchmove 与 touchend 三个数据流。

touchstart$ = fromEvent<TouchEvent>(document, 'touchstart');
touchend$ = fromEvent<TouchEvent>(document, 'touchend');
touchmove$ = fromEvent<TouchEvent>(document, 'touchmove');

4）定时刷新

interval$ = interval(5000);

5）服务端请求

fetch$ = fromFetch('https://randomapi.azurewebsites.net/api/users');

组合/转换数据流

1）点击刷新事件流

在点击刷新时，咱们但愿短期内屡次点击只触发最后一次，这经过 RxJS 的 debounceTime operator 就能够实现。

clickRefresh$ = this.click$.pipe(debounceTime(300));

2）自动刷新流

使用 RxJS 的 switchMap 与以前定义好的 interval$ 数据流配合。

autoRefresh$ = change$.pipe(
  switchMap(enabled => (enabled ? interval$ : EMPTY))
);

3）下拉刷新流

结合以前定义好的 touchstart$touchmove$ 与 touchend$ 数据流。

pullRefresh$ = touchstart$.pipe(
  switchMap(touchStartEvent =>
    touchmove$.pipe(
      map(touchMoveEvent => touchMoveEvent.touches[0].pageY - touchStartEvent.touches[0].pageY),
      takeUntil(touchend$)
    )
  ),
  filter(position => position >= 300),
  take(1),
  repeat()
);

最后，咱们经过 merge 函数将定义好的 clickRefresh$autoRefresh$ 与 pullRefresh$ 合并，就获得了刷新数据流。

refresh$ = merge(clickRefresh$, autoRefresh$, pullRefresh$));

消费数据流并更新视图

将刷新数据流直接经过 switchMap 打平到在第一步到定义好的 fetch$，咱们就得到了视图数据流。

能够经过在 Angular 框架中能够直接 async pipe 将视图流直接映射为视图：

<div *ngFor="let user of view$ | async">
</div>

在其余框架中能够经过 subscribe 得到数据流中的真实数据，再更新视图。

至此，咱们就使用响应式编程完整的开发完成了当前新闻应用，示例代码[1]由 Angular 开发，行数不超过 160 行。

咱们总结一下，使用响应式编程思想开发前端应用时经历的三个过程与第一节中公式的对应关系：

View = reactionFn(UserEvent | Timer | Remote API)

1）描述源数据流

与事件UserEvent | Timer | Remote API 对应，在 RxJS 中对应函数分别是：

• UserEvent: fromEvent
• Timer: interval, timer
• Remote API: fromFetch, webSocket

2）组合转换数据流

与响应函数(reactionFn)对应，在 RxJS 中对应的部分方法是：

• COMBINING: merge, combineLatest, zip
• MAPPING: map
• FILTERING: filter
• REDUCING: reduce, max, count, scan
• TAKING: take, takeWhile
• SKIPPING: skip, skipWhile, takeLast, last
• TIME: delay, debounceTime, throttleTime

3）消费数据流更新视图

与 View 对应，在 RxJS 及 Angular 中可使用：

• subscribe
• async pipe

响应式编程相对于 MVVM 或者 Redux 有什么优势呢？

• 描述事件发生的自己，而非计算过程或者中间状态。
• 提供了组合和转换数据流的方法，这也意味着咱们得到了复用持续变化数据的方法。
• 因为全部数据流均由层层组合与转换得到，这也就意味着咱们能够精确追踪事件及数据变化的来源。

若是咱们将 RxJS 的 Marbles 图的时间轴模糊，并在每次视图更新时增长纵切面，咱们就会发现这样两件有趣的事情：

• Action 是 EventStream 的简化。
• State 是 Stream 在某个时刻的对应。

难怪咱们能够在 Redux 官网中有这样一句话：若是你已经使用了 RxJS，极可能你再也不须要 Redux 了。

The question is: do you really need Redux if you already use Rx? Maybe not. It's not hard to re-implement Redux in Rx. Some say it's a two-liner using Rx.scan() method. It may very well be!

写到这里，咱们对网页视图可以正确地响应相关事件这句话是否能够进行进一步的抽象呢？

全部事件 -- 找到 --> 相关事件 -- 作出 --> 响应

而按时间顺序发生的事件，本质上就是数据流，进一步拓展就可变成：

源数据流 -- 转换 --> 中间数据流 -- 订阅 --> 消费数据流

这正是响应式编程在前端可以完美工做的基础思想。可是该思想是否只在前端开发中有所应用呢？

答案是否认的，该思想不只能够应用于前端开发，在后端开发乃至实时计算中都有着普遍的应用。

3、 打破信息之墙

在先后端开发者之间，一般由一面叫 REST API 的信息之墙隔开，REST API 隔离了先后端开发者的职责，提高了开发效率。但它一样让先后端开发者的眼界被这面墙隔开，让咱们试着来推倒这面信息之墙，一窥一样的思想在实时计算中的应用。

1 、实时计算 与 Apache Flink

在开始下一部分以前，让咱们先介绍一下 Flink。Apache Flink 是由 Apache 软件基金会开发的开源流处理框架，用于在无边界和有边界数据流上进行有状态的计算。它的数据流编程模型在有限和无限数据集上提供单次事件（event-at-a-time）处理能力。

在实际的应用中，Flink 一般用于开发如下三种应用：

• 事件驱动型应用 事件驱动型应用从一个或多个事件流提取数据，并根据到来的事件触发计算、状态更新或其余外部动做。场景包括基于规则的报警，异常检测，反欺诈等等。
• 数据分析应用 数据分析任务须要从原始数据中提取有价值的信息和指标。例如双十一成交额计算，网络质量监测等等。
• 数据管道(ETL)应用 提取-转换-加载（ETL）是一种在存储系统之间进行数据转换和迁移的经常使用方法。ETL 做业一般会周期性地触发，将数据从事务型数据库拷贝到分析型数据库或数据仓库。

咱们这里以计算电商平台双十一每小时成交额为例，看下咱们在以前章节获得方案是否仍然能够继续使用。

在这个场景中咱们首先要获取用户购物下单数据，随后计算每小时成交数据，而后将每小时的成交数据转存到数据库并被 Redis 缓存，最终经过接口获取后展现在页面中。

在这个链路中的数据流处理逻辑为：

用户下单数据流 -- 转换 --> 每小时成交数据流 -- 订阅 --> 写入数据库

与以前章节中介绍的：

源数据流 -- 转换 --> 中间数据流 -- 订阅 --> 消费数据流

思想彻底一致。

若是咱们用 Marbles 描述这个过程，就会获得这样的结果，看起来很简单，彷佛使用 RxJS 的 window operator 也能够完成一样的功能，可是事实真的如此吗？

2 、被隐藏的复杂度

真实的实时计算比前端中响应式编程的复杂度要高不少，咱们在这里举几个例子：

事件乱序

在前端开发过程当中，咱们也会碰到事件乱序的状况，最经典的状况先发起的请求后收到响应，能够用以下的 Marbles 图表示。这种状况在前端有不少种办法进行处理，咱们在这里就略过不讲。

咱们今天想介绍的是数据处理时面临的时间乱序状况。在前端开发中，咱们有一个很重要的前提，这个前提大幅度下降了开发前端应用的复杂度，那就是：前端事件的发生时间和处理时间相同。

想象一下，若是用户执行页面动做，例如 click, mousemove 等事件都变成了异步事件，而且响应时间未知，那整个前端的开发复杂度会如何。

可是事件的发生时间与处理时间不一样，在实时计算领域是一个重要的前提。咱们仍以每小时成交额计算为例，当原始数据流通过层层传输以后，在计算节点的数据的前后顺极可能已经乱序了。

若是咱们仍然以数据的到来时间来进行窗口划分，最后的计算结果就会产生错误：

为了让 window2 的窗口的计算结果正确，咱们须要等待 late event 到来以后进行计算，可是这样咱们就面临了一个两难问题：

• 无限等下去：late event 可能在传输过程当中丢失，window2 窗口永远没有数据产出。
• 等待时间过短：late event 尚未到来，计算结果错误。

Flink 引入了 Watermark 机制来解决这个问题，Watermark 定义了何时再也不等待 late event，本质上提供了实时计算的准确性和实时性的折中方案。

关于 Watermark 有个形象的比喻：上学的时候，老师会将班级的门关上，而后说：“从这个点以后来的同窗都算迟到了，通通罚站“。在 Flink 中，Watermark 充当了老师关门的这个动做。

数据反压

在浏览器中使用 RxJS 时，不知道你们有没有考虑这样一种状况：observable 产生的速度快于 operator 或者 observer 消费的速度时，会产生大量的未消费的数据被缓存在内存中。这种状况被称为反压，幸运的是，在前端产生数据反压只会致使浏览器内存被大量占用，除此以外不会有更严重的后果。

可是在实时计算中，当数据产生的速度高于中间节点处理能力，或者超过了下游数据的消费能力时，应当如何处理？

对于许多流应用程序来讲，数据丢失是不可接受的，为了保证这一点，Flink 设计了这样一种机制：

• 在理想状况，在一个持久通道中缓冲数据。
• 当数据产生的速度高于中间节点处理能力，或者超过了下游数据的消费能力时，速度较慢的接收器会在队列的缓冲做用耗尽后当即下降发送器的速度。更形象的比喻是，在数据流流速变慢时，将整个管道从水槽“回压”到水源，并对水源进行节流，以便将速度调整到最慢的部分，从而达到稳定状态。

Checkpoint

实时计算领域，每秒钟处理的数据可能有数十亿条，这些数据的处理不可能由单台机器独立完成。事实上，在 Flink 中，operator 运算逻辑会由不一样的 subtask 在 不一样的 taskmanager 上执行，这时咱们就面临了另一个问题，当某台机器发生问题时，总体的运算逻辑与状态该如何处理才能保证最后运算结果的正确性？

Flink 中引入了 checkpoint 机制用于保证能够对做业的状态和计算位置进行恢复，checkpoint 使 Flink 的状态具备良好的容错性。Flink 使用了 Chandy-Lamport algorithm 算法的一种变体，称为异步 barrier 快照（asynchronous barrier snapshotting）。

当开始 checkpoint 时，它会让全部 sources 记录它们的偏移量，并将编号的 checkpoint barriers 插入到它们的流中。这些 barriers 会通过每一个 operator 时标注每一个 checkpoint 先后的流部分。

当发生错误时，Flink 能够根据 checkpoint 存储的 state 进行状态恢复，保证最终结果的正确性。

冰山一角

因为篇幅的关系，今天介绍的部分只能是冰山一角，不过

源数据流 -- 转换 --> 中间数据流 -- 订阅 --> 消费数据流

的模型不管在响应式编程仍是实时计算都是通用的，但愿这篇文章可以让你们对数据流的思想有更多的思考。

做者：开发者小助手_LS
原文连接本文为阿里云原创内容，未经容许不得转载