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

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

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

interval$ = interval(5000);

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

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

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

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()
);

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

在其他框架中可以通过 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!

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

所有事件 -- 找到 --> 相关事件 -- 做出 --> 响应

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

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

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

答案是否定的，该思想不仅可以应用于前端开发，在后端开发乃至实时计算中都有着广泛的应用。

三 打破信息之墙

在前后端开发者之间，通常由一面叫 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 进行状态恢复，保证最终结果的正确性。

的模型无论在响应式编程还是实时计算都是通用的，希望这篇文章能够让大家对数据流的思想有更多的思考。

关于从RxJS到Flink该如何处理数据流问题的解答就分享到这里了，希望以上内容可以对大家有一定的帮助，如果你还有很多疑惑没有解开，可以关注创新互联行业资讯频道了解更多相关知识。