ReactiveX -Rx

轉載：https://www.open-open.com/lib/view/open1440166491833.html

ReactiveX是Reactive Extensions的縮寫，通常簡寫爲Rx，最初是LINQ的一個擴展，由微軟的架構師Erik Meijer領導的團隊開發，在2012年11月開源，Rx是一個編程模型，目標是提供一致的編程接口，幫助開發者更方便的處理異步數據流，Rx庫支持.NET、JavaScript和C++，Rx近幾年愈來愈流行了，如今已經支持幾乎所有的流行編程語言了，Rx的大部分語言庫由ReactiveX這個組織負責維護，比較流行的有RxJava/RxJS/Rx.NET，社區網站是 http://reactivex.io/

什麼是ReactiveX

微軟給的定義是，Rx是一個函數庫，讓開發者能夠利用可觀察序列和LINQ風格查詢操做符來編寫異步和基於事件的程序，使用Rx，開發者能夠用Observables表示異步數據流，用LINQ操做符查詢異步數據流， 用Schedulers參數化異步數據流的併發處理，Rx能夠這樣定義：Rx = Observables + LINQ + Schedulers。

ReactiveX.io給的定義是，Rx是一個使用可觀察數據流進行異步編程的編程接口，ReactiveX結合了觀察者模式、迭代器模式和函數式編程的精華。

ReactiveX的應用

不少公司都在使用ReactiveX，例如Microsoft、Netflix、Github、Trello、SoundCloud。

ReactiveX宣言

ReactiveX不只僅是一個編程接口，它是一種編程思想的突破，它影響了許多其它的程序庫和框架以及編程語言。

使用觀察者模式

• 建立：Rx能夠方便的建立事件流和數據流
• 組合：Rx使用查詢式的操做符組合和變換數據流
• 監聽：Rx能夠訂閱任何可觀察的數據流並執行操做

簡化代碼

• 函數式風格：對可觀察數據流使用無反作用的輸入輸出函數，避免了程序裏錯綜複雜的狀態 簡化代碼：Rx的操做符統統常能夠將複雜的難題簡化爲不多的幾行代碼
• 異步錯誤處理：傳統的try/catch沒辦法處理異步計算，Rx提供了合適的錯誤處理機制
• 輕鬆使用併發：Rx的Observables和Schedulers讓開發者能夠擺脫底層的線程同步和各類併發問題

使用Observable的優點

Rx擴展了觀察者模式用於支持數據和事件序列，添加了一些操做符，它讓你能夠聲明式的組合這些序列，而無需關注底層的實現：如線程、同步、線程安全、併發數據結構和非阻塞IO。

 

Observable經過使用最佳的方式訪問異步數據序列填補了這個間隙

單個數據                   　　　　 多個數據
同步  T getData()                 Iterable<T> getData()
異步  Future<T> getData()         Observable<T> getData()

 

Rx的Observable模型讓你能夠像使用集合數據同樣操做異步事件流，對異步事件流使用各類簡單、可組合的操做。

Observable可組合

對於單層的異步操做來講，Java中Future對象的處理方式是很是簡單有效的，可是一旦涉及到嵌套，它們就開始變得異常繁瑣和複雜。使用Future很難很好的組合帶條件的異步執行流程（考慮到運行時各類潛在的問題，甚至能夠說是不可能的），固然，要想實現仍是能夠作到的，可是很是困難，或許你能夠用Future.get()，但這樣作，異步執行的優點就徹底沒有了。從另外一方面說，Rx的Observable一開始就是爲組合異步數據流準備的。

Observable更靈活

Rx的Observable不只支持處理單獨的標量值（就像Future能夠作的），也支持數據序列，甚至是無窮的數據流。Observable是一個抽象概念，適用於任何場景。Observable擁有它的近親Iterable的所有優雅與靈活。

Observable是異步的雙向push，Iterable是同步的單向pull，對比：

事件           Iterable(pull)          Observable(push)
獲取數據        T next()                onNext(T)
異常處理        throws Exception        onError(Exception)
任務完成        !hasNext()              onCompleted()

 

Observable無偏見

Rx對於對於併發性或異步性沒有任何特殊的偏好，Observable能夠用任何方式實現，線程池、事件循環、非阻塞IO、Actor模式，任何知足你的需求的，你擅長或偏好的方式均可以。不管你選擇怎樣實現它，不管底層實現是阻塞的仍是非阻塞的，客戶端代碼將全部與Observable的交互都當作是異步的。

Observable是如何實現的？

public Observable<data> getData();

 

• 它能與調用者在同一線程同步執行嗎？
• 它能異步地在單獨的線程執行嗎？
• 它會將工做分發到多個線程，返回數據的順序是任意的嗎？
• 它使用Actor模式而不是線程池嗎？
• 它使用NIO和事件循環執行異步網絡訪問嗎？
• 它使用事件循環將工做線程從回調線程分離出來嗎？

從Observer的視角看，這些都無所謂，重要的是：使用Rx，你能夠改變你的觀念，你能夠在徹底不影響Observable程序庫使用者的狀況下，完全的改變Observable的底層實現。

使用回調存在不少問題

回調在不阻塞任何事情的狀況下，解決了Future.get()過早阻塞的問題。因爲響應結果一旦就緒Callback就會被調用，它們天生就是高效率的。不過，就像使用Future同樣，對於單層的異步執行來講，回調很容易使用，對於嵌套的異步組合，它們顯得很是笨拙。

Rx是一個多語言的實現

Rx在大量的編程語言中都有實現，並尊重實現語言的風格，並且更多的實現正在飛速增長。

響應式編程

Rx提供了一系列的操做符，你可使用它們來過濾(filter)、選擇(select)、變換(transform)、結合(combine)和組合(compose)多個Observable，這些操做符讓執行和複合變得很是高效。

你能夠把Observable當作Iterable的推送方式的等價物，使用Iterable，消費者從生產者那拉取數據，線程阻塞直至數據準備好。使用Observable，在數據準備好時，生產者將數據推送給消費者。數據能夠同步或異步的到達，這種方式更靈活。

下面的例子展現了類似的高階函數在Iterable和Observable上的應用

// Iterable getDataFromLocalMemory()
  .skip(10)
  .take(5)
  .map({ s -> return s + " transformed" })
  .forEach({ println "next => " + it }) // Observable getDataFromNetwork()
  .skip(10)
  .take(5)
  .map({ s -> return s + " transformed" })
  .subscribe({ println "onNext => " + it })

 

Observable類型給GOF的觀察者模式添加了兩種缺乏的語義，這樣就和Iterable類型中可用的操做一致了：

1. 生產者能夠發信號給消費者，通知它沒有更多數據可用了（對於Iterable，一個for循環正常完成表示沒有數據了；對於Observable，就是調用觀察者的onCompleted方法）
2. 生產者能夠發信號給消費者，通知它遇到了一個錯誤（對於Iterable，迭代過程當中發生錯誤會拋出異常；對於Observable，就是調用觀察者(Observer)的onError方法）

有了這兩種功能，Rx就能使Observable與Iterable保持一致了，惟一的不一樣是數據流的方向。任何對Iterable的操做，你均可以對Observable使用。

名詞定義

這裏給出一些名詞的翻譯

• Reactive 直譯爲反應性的，有活性的，根據上下文通常反覺得反應式，響應式
• Iterable 可迭代對象，支持以迭代器的形式遍歷，許多語言中都存在這個概念
• Observable 可觀察對象，在Rx中定義爲更強大的Iterable，在觀察者模式中是被觀察的對象，一旦數據產生或發生變化，會經過某種方式通知觀察者或訂閱者
• Observer 觀察者對象，監聽Observable發射的數據並作出響應，Subscriber是它的一個特殊實現 *emit 直譯爲發射，發佈，發出，含義是Observable在數據產生或變化時發送通知給Observer，調用Observer對應的方法，文章裏一概譯爲發射 *items 直譯爲項目，條目，在Rx裏是指Observable發射的數據項，文章裏一概譯爲數據，數據項