從RxJS到Flink該如何處理數(shù)據(jù)流，針對這個問題，這篇文章詳細(xì)介紹了相對應(yīng)的分析和解答，希望可以幫助更多想解決這個問題的小伙伴找到更簡單易行的方法。

為平湖等地區(qū)用戶提供了全套網(wǎng)頁設(shè)計制作服務(wù)，及平湖網(wǎng)站建設(shè)行業(yè)解決方案。主營業(yè)務(wù)為成都網(wǎng)站建設(shè)、網(wǎng)站設(shè)計、平湖網(wǎng)站設(shè)計，以傳統(tǒng)方式定制建設(shè)網(wǎng)站，并提供域名空間備案等一條龍服務(wù)，秉承以專業(yè)、用心的態(tài)度為用戶提供真誠的服務(wù)。我們深信只要達(dá)到每一位用戶的要求，就會得到認(rèn)可，從而選擇與我們長期合作。這樣，我們也可以走得更遠(yuǎn)！

一 前端開發(fā)在開發(fā)什么

大家在前端開發(fā)的過程中，可能會想過這樣一個問題：前端開發(fā)究竟是在開發(fā)什么？在我看來，前端開發(fā)的本質(zhì)是讓網(wǎng)頁視圖能夠正確地響應(yīng)相關(guān)事件。在這句話中有三個關(guān)鍵字："網(wǎng)頁視圖"，"正確地響應(yīng)"和"相關(guān)事件"。

"相關(guān)事件"可能包括頁面點擊，鼠標(biāo)滑動，定時器，服務(wù)端請求等等，"正確地響應(yīng)"意味著我們要根據(jù)相關(guān)的事件來修改一些狀態(tài)，而"網(wǎng)頁視圖"就是我們前端開發(fā)中最熟悉的部分了。

按照這樣的觀點我們可以給出這樣 視圖 = 響應(yīng)函數(shù)(事件) 的公式：

View = reactionFn(Event)

在前端開發(fā)中，需要被處理事件可以歸類為以下三種：

• 用戶執(zhí)行頁面動作，例如 click, mousemove 等事件。

• 遠(yuǎn)程服務(wù)端與本地的數(shù)據(jù)交互，例如 fetch, websocket。

• 本地的異步事件，例如 setTimeout, setInterval async_event。

這樣我們的公式就可以進(jìn)一步推導(dǎo)為：

View = reactionFn(UserEvent | Timer | Remote API)

二 應(yīng)用中的邏輯處理

為了能夠更進(jìn)一步理解這個公式與前端開發(fā)的關(guān)系，我們以新聞網(wǎng)站舉例，該網(wǎng)站有以下三個要求：

• 單擊刷新：單擊 Button 刷新數(shù)據(jù)。

• 勾選刷新：勾選 Checkbox 時自動刷新，否則停止自動刷新。

• 下拉刷新：當(dāng)用戶從屏幕頂端下拉時刷新數(shù)據(jù)。

如果從前端的角度分析，這三種需求分別對應(yīng)著：

• 單擊刷新：click -> fetch

• 勾選刷新：change -> (setInterval + clearInterval) -> fetch

• 下拉刷新：(touchstart + touchmove + touchend) -> fetch news_app

1 MVVM

在 MVVM 的模式下，對應(yīng)上文的響應(yīng)函數(shù)（reactionFn）會在 Model 與 ViewModel 或者 View 與 ViewModel 之間進(jìn)行被執(zhí)行，而事件 (Event) 會在 View 與 ViewModel 之間進(jìn)行處理。

MVVM 可以很好的抽象視圖層與數(shù)據(jù)層，但是響應(yīng)函數(shù)（reactionFn）會散落在不同的轉(zhuǎn)換過程中，這會導(dǎo)致數(shù)據(jù)的賦值與收集過程難以進(jìn)行精確追蹤。另外因為事件 (Event) 的處理在該模型中與視圖部分緊密相關(guān)，導(dǎo)致 View 與 ViewModel 之間對事件處理的邏輯復(fù)用困難。

2 Redux

在 Redux 最簡單的模型下，若干個事件 (Event) 的組合會對應(yīng)到一個 Action 上，而 reducer 函數(shù)可以被直接認(rèn)為與上文提到的響應(yīng)函數(shù) (reactionFn) 對應(yīng)。

但是在 Redux 中：

• State 只能用于描述中間狀態(tài)，而不能描述中間過程。

• Action 與 Event 的關(guān)系并非一一對應(yīng)導(dǎo)致 State 難以追蹤實際變化來源。

3 響應(yīng)式編程與 RxJS

維基百科中是這樣定義響應(yīng)式編程：

在計算中，響應(yīng)式編程或反應(yīng)式編程（英語：Reactive programming）是一種面向數(shù)據(jù)流和變化傳播的聲明式編程范式。這意味著可以在編程語言中很方便地表達(dá)靜態(tài)或動態(tài)的數(shù)據(jù)流，而相關(guān)的計算模型會自動將變化的值通過數(shù)據(jù)流進(jìn)行傳播。

以數(shù)據(jù)流維度重新考慮用戶使用該應(yīng)用的流程：

• 點擊按鈕 -> 觸發(fā)刷新事件 -> 發(fā)送請求 -> 更新視圖

• 勾選自動刷新

• 手指觸摸屏幕

• 自動刷新間隔 -> 觸發(fā)刷新事件 -> 發(fā)送請求 -> 更新視圖

• 手指在屏幕上下滑

• 自動刷新間隔 -> 觸發(fā)刷新事件 -> 發(fā)送請求 -> 更新視圖

• 手指在屏幕上停止滑動 -> 觸發(fā)下拉刷新事件 -> 發(fā)送請求 -> 更新視圖

• 自動刷新間隔 -> 觸發(fā)刷新事件 -> 發(fā)送請求 -> 更新視圖

• 關(guān)閉自動刷新

以 Marbles 圖表示：

拆分上圖邏輯，就會得到使用響應(yīng)式編程開發(fā)當(dāng)前新聞應(yīng)用時的三個步驟：

• 定義源數(shù)據(jù)流

• 組合/轉(zhuǎn)換數(shù)據(jù)流

• 消費數(shù)據(jù)流并更新視圖

我們分別來進(jìn)行詳細(xì)描述。

定義源數(shù)據(jù)流

使用 RxJS，我們可以很方便的定義出各種 Event 數(shù)據(jù)流。

1）單擊操作

涉及 click 數(shù)據(jù)流。

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

2）勾選操作

涉及 change 數(shù)據(jù)流。

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

3）下拉操作

涉及 touchstart, touchmove 與 touchend 三個數(shù)據(jù)流。

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

4）定時刷新

interval$ = interval(5000);

5）服務(wù)端請求

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

組合/轉(zhuǎn)換數(shù)據(jù)流

1）點擊刷新事件流

在點擊刷新時，我們希望短時間內(nèi)多次點擊只觸發(fā)最后一次，這通過 RxJS 的 debounceTime operator 就可以實現(xiàn)。

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

2）自動刷新流

使用 RxJS 的 switchMap 與之前定義好的 interval$ 數(shù)據(jù)流配合。

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

3）下拉刷新流

結(jié)合之前定義好的 touchstart$touchmove$ 與 touchend$ 數(shù)據(jù)流。

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 函數(shù)將定義好的 clickRefresh$autoRefresh$ 與 pullRefresh$ 合并，就得到了刷新數(shù)據(jù)流。

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

消費數(shù)據(jù)流并更新視圖

將刷新數(shù)據(jù)流直接通過 switchMap 打平到在第一步到定義好的 fetch$，我們就獲得了視圖數(shù)據(jù)流。

可以通過在 Angular 框架中可以直接 async pipe 將視圖流直接映射為視圖：

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

在其他框架中可以通過 subscribe 獲得數(shù)據(jù)流中的真實數(shù)據(jù)，再更新視圖。

至此，我們就使用響應(yīng)式編程完整的開發(fā)完成了當(dāng)前新聞應(yīng)用，示例代碼[1]由 Angular 開發(fā)，行數(shù)不超過 160 行。

我們總結(jié)一下，使用響應(yīng)式編程思想開發(fā)前端應(yīng)用時經(jīng)歷的三個過程與第一節(jié)中公式的對應(yīng)關(guān)系：

View = reactionFn(UserEvent | Timer | Remote API)

1）描述源數(shù)據(jù)流

與事件UserEvent | Timer | Remote API 對應(yīng)，在 RxJS 中對應(yīng)函數(shù)分別是：

• UserEvent: fromEvent

• Timer: interval, timer

• Remote API: fromFetch, webSocket

2）組合轉(zhuǎn)換數(shù)據(jù)流

與響應(yīng)函數(shù)(reactionFn)對應(yīng)，在 RxJS 中對應(yīng)的部分方法是：

• 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）消費數(shù)據(jù)流更新視圖

與 View 對應(yīng)，在 RxJS 及 Angular 中可以使用：

• subscribe

• async pipe

響應(yīng)式編程相對于 MVVM 或者 Redux 有什么優(yōu)點呢？

• 描述事件發(fā)生的本身，而非計算過程或者中間狀態(tài)。

• 提供了組合和轉(zhuǎn)換數(shù)據(jù)流的方法，這也意味著我們獲得了復(fù)用持續(xù)變化數(shù)據(jù)的方法。

• 由于所有數(shù)據(jù)流均由層層組合與轉(zhuǎn)換獲得，這也就意味著我們可以精確追蹤事件及數(shù)據(jù)變化的來源。

如果我們將 RxJS 的 Marbles 圖的時間軸模糊，并在每次視圖更新時增加縱切面，我們就會發(fā)現(xiàn)這樣兩件有趣的事情：

• Action 是 EventStream 的簡化。

• State 是 Stream 在某個時刻的對應(yīng)。

難怪我們可以在 Redux 官網(wǎng)中有這樣一句話：如果你已經(jīng)使用了 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!

寫到這里，我們對網(wǎng)頁視圖能夠正確地響應(yīng)相關(guān)事件這句話是否可以進(jìn)行進(jìn)一步的抽象呢？

所有事件 -- 找到 --> 相關(guān)事件 -- 做出 --> 響應(yīng)

而按時間順序發(fā)生的事件，本質(zhì)上就是數(shù)據(jù)流，進(jìn)一步拓展就可變成：

源數(shù)據(jù)流 -- 轉(zhuǎn)換 --> 中間數(shù)據(jù)流 -- 訂閱 --> 消費數(shù)據(jù)流

這正是響應(yīng)式編程在前端能夠完美工作的基礎(chǔ)思想。但是該思想是否只在前端開發(fā)中有所應(yīng)用呢？

答案是否定的，該思想不僅可以應(yīng)用于前端開發(fā)，在后端開發(fā)乃至實時計算中都有著廣泛的應(yīng)用。

三 打破信息之墻

在前后端開發(fā)者之間，通常由一面叫 REST API 的信息之墻隔開，REST API 隔離了前后端開發(fā)者的職責(zé)，提升了開發(fā)效率。但它同樣讓前后端開發(fā)者的眼界被這面墻隔開，讓我們試著來推倒這面信息之墻，一窺同樣的思想在實時計算中的應(yīng)用。

1 實時計算 與 Apache Flink

在開始下一部分之前，讓我們先介紹一下 Flink。Apache Flink 是由 Apache 軟件基金會開發(fā)的開源流處理框架，用于在無邊界和有邊界數(shù)據(jù)流上進(jìn)行有狀態(tài)的計算。它的數(shù)據(jù)流編程模型在有限和無限數(shù)據(jù)集上提供單次事件（event-at-a-time）處理能力。

在實際的應(yīng)用中，F(xiàn)link 通常用于開發(fā)以下三種應(yīng)用：

• 事件驅(qū)動型應(yīng)用 事件驅(qū)動型應(yīng)用從一個或多個事件流提取數(shù)據(jù)，并根據(jù)到來的事件觸發(fā)計算、狀態(tài)更新或其他外部動作。場景包括基于規(guī)則的報警，異常檢測，反欺詐等等。

• 數(shù)據(jù)分析應(yīng)用 數(shù)據(jù)分析任務(wù)需要從原始數(shù)據(jù)中提取有價值的信息和指標(biāo)。例如雙十一成交額計算，網(wǎng)絡(luò)質(zhì)量監(jiān)測等等。

• 數(shù)據(jù)管道(ETL)應(yīng)用 提取-轉(zhuǎn)換-加載（ETL）是一種在存儲系統(tǒng)之間進(jìn)行數(shù)據(jù)轉(zhuǎn)換和遷移的常用方法。ETL 作業(yè)通常會周期性地觸發(fā)，將數(shù)據(jù)從事務(wù)型數(shù)據(jù)庫拷貝到分析型數(shù)據(jù)庫或數(shù)據(jù)倉庫。

我們這里以計算電商平臺雙十一每小時成交額為例，看下我們在之前章節(jié)得到方案是否仍然可以繼續(xù)使用。

在這個場景中我們首先要獲取用戶購物下單數(shù)據(jù)，隨后計算每小時成交數(shù)據(jù)，然后將每小時的成交數(shù)據(jù)轉(zhuǎn)存到數(shù)據(jù)庫并被 redis 緩存，最終通過接口獲取后展示在頁面中。

在這個鏈路中的數(shù)據(jù)流處理邏輯為：

用戶下單數(shù)據(jù)流 -- 轉(zhuǎn)換 --> 每小時成交數(shù)據(jù)流 -- 訂閱 --> 寫入數(shù)據(jù)庫

與之前章節(jié)中介紹的：

源數(shù)據(jù)流 -- 轉(zhuǎn)換 --> 中間數(shù)據(jù)流 -- 訂閱 --> 消費數(shù)據(jù)流

思想完全一致。

如果我們用 Marbles 描述這個過程，就會得到這樣的結(jié)果，看起來很簡單，似乎使用 RxJS 的 window operator 也可以完成同樣的功能，但是事實真的如此嗎？

2 被隱藏的復(fù)雜度

真實的實時計算比前端中響應(yīng)式編程的復(fù)雜度要高很多，我們在這里舉幾個例子：

事件亂序

在前端開發(fā)過程中，我們也會碰到事件亂序的情況，最經(jīng)典的情況先發(fā)起的請求后收到響應(yīng)，可以用如下的 Marbles 圖表示。這種情況在前端有很多種辦法進(jìn)行處理，我們在這里就略過不講。

我們今天想介紹的是數(shù)據(jù)處理時面臨的時間亂序情況。在前端開發(fā)中，我們有一個很重要的前提，這個前提大幅度降低了開發(fā)前端應(yīng)用的復(fù)雜度，那就是：前端事件的發(fā)生時間和處理時間相同。

想象一下，如果用戶執(zhí)行頁面動作，例如 click, mousemove 等事件都變成了異步事件，并且響應(yīng)時間未知，那整個前端的開發(fā)復(fù)雜度會如何。

但是事件的發(fā)生時間與處理時間不同，在實時計算領(lǐng)域是一個重要的前提。我們?nèi)砸悦啃r成交額計算為例，當(dāng)原始數(shù)據(jù)流經(jīng)過層層傳輸之后，在計算節(jié)點的數(shù)據(jù)的先后順很可能已經(jīng)亂序了。

如果我們?nèi)匀灰詳?shù)據(jù)的到來時間來進(jìn)行窗口劃分，最后的計算結(jié)果就會產(chǎn)生錯誤：

為了讓 window2 的窗口的計算結(jié)果正確，我們需要等待 late event 到來之后進(jìn)行計算，但是這樣我們就面臨了一個兩難問題：

• 無限等下去：late event 可能在傳輸過程中丟失，window2 窗口永遠(yuǎn)沒有數(shù)據(jù)產(chǎn)出。

• 等待時間太短：late event 還沒有到來，計算結(jié)果錯誤。

Flink 引入了 Watermark 機(jī)制來解決這個問題，Watermark 定義了什么時候不再等待 late event，本質(zhì)上提供了實時計算的準(zhǔn)確性和實時性的折中方案。

關(guān)于 Watermark 有個形象的比喻：上學(xué)的時候，老師會將班級的門關(guān)上，然后說：“從這個點之后來的同學(xué)都算遲到了，統(tǒng)統(tǒng)罰站“。在 Flink 中，Watermark 充當(dāng)了老師關(guān)門的這個動作。

數(shù)據(jù)反壓

在瀏覽器中使用 RxJS 時，不知道大家有沒有考慮這樣一種情況：observable 產(chǎn)生的速度快于 operator 或者 observer 消費的速度時，會產(chǎn)生大量的未消費的數(shù)據(jù)被緩存在內(nèi)存中。這種情況被稱為反壓，幸運的是，在前端產(chǎn)生數(shù)據(jù)反壓只會導(dǎo)致瀏覽器內(nèi)存被大量占用，除此之外不會有更嚴(yán)重的后果。

但是在實時計算中，當(dāng)數(shù)據(jù)產(chǎn)生的速度高于中間節(jié)點處理能力，或者超過了下游數(shù)據(jù)的消費能力時，應(yīng)當(dāng)如何處理？

對于許多流應(yīng)用程序來說，數(shù)據(jù)丟失是不可接受的，為了保證這一點，F(xiàn)link 設(shè)計了這樣一種機(jī)制：

• 在理想情況，在一個持久通道中緩沖數(shù)據(jù)。

• 當(dāng)數(shù)據(jù)產(chǎn)生的速度高于中間節(jié)點處理能力，或者超過了下游數(shù)據(jù)的消費能力時，速度較慢的接收器會在隊列的緩沖作用耗盡后立即降低發(fā)送器的速度。更形象的比喻是，在數(shù)據(jù)流流速變慢時，將整個管道從水槽“回壓”到水源，并對水源進(jìn)行節(jié)流，以便將速度調(diào)整到最慢的部分，從而達(dá)到穩(wěn)定狀態(tài)。

Checkpoint

實時計算領(lǐng)域，每秒鐘處理的數(shù)據(jù)可能有數(shù)十億條，這些數(shù)據(jù)的處理不可能由單臺機(jī)器獨立完成。事實上，在 Flink 中，operator 運算邏輯會由不同的 subtask 在 不同的 taskmanager 上執(zhí)行，這時我們就面臨了另外一個問題，當(dāng)某臺機(jī)器發(fā)生問題時，整體的運算邏輯與狀態(tài)該如何處理才能保證最后運算結(jié)果的正確性？

Flink 中引入了 checkpoint 機(jī)制用于保證可以對作業(yè)的狀態(tài)和計算位置進(jìn)行恢復(fù)，checkpoint 使 Flink 的狀態(tài)具有良好的容錯性。Flink 使用了 Chandy-Lamport algorithm 算法的一種變體，稱為異步 barrier 快照（asynchronous barrier snapshotting）。

當(dāng)開始 checkpoint 時，它會讓所有 sources 記錄它們的偏移量，并將編號的 checkpoint barriers 插入到它們的流中。這些 barriers 會經(jīng)過每個 operator 時標(biāo)注每個 checkpoint 前后的流部分。

當(dāng)發(fā)生錯誤時，F(xiàn)link 可以根據(jù) checkpoint 存儲的 state 進(jìn)行狀態(tài)恢復(fù)，保證最終結(jié)果的正確性。

的模型無論在響應(yīng)式編程還是實時計算都是通用的，希望這篇文章能夠讓大家對數(shù)據(jù)流的思想有更多的思考。

關(guān)于從RxJS到Flink該如何處理數(shù)據(jù)流問題的解答就分享到這里了，希望以上內(nèi)容可以對大家有一定的幫助，如果你還有很多疑惑沒有解開，可以關(guān)注創(chuàng)新互聯(lián)行業(yè)資訊頻道了解更多相關(guān)知識。

本文名稱：從RxJS到Flink該如何處理數(shù)據(jù)流
文章地址：http://www.chinadenli.net/article34/peeepe.html

成都網(wǎng)站建設(shè)公司_創(chuàng)新互聯(lián)，為您提供全網(wǎng)營銷推廣、服務(wù)器托管、網(wǎng)站收錄、網(wǎng)頁設(shè)計公司、云服務(wù)器、靜態(tài)網(wǎng)站

廣告

聲明：本網(wǎng)站發(fā)布的內(nèi)容（圖片、視頻和文字）以用戶投稿、用戶轉(zhuǎn)載內(nèi)容為主，如果涉及侵權(quán)請盡快告知，我們將會在第一時間刪除。文章觀點不代表本網(wǎng)站立場，如需處理請聯(lián)系客服。電話：028-86922220；郵箱：631063699@qq.com。內(nèi)容未經(jīng)允許不得轉(zhuǎn)載，或轉(zhuǎn)載時需注明來源： 創(chuàng)新互聯(lián)