ReactiveX

微軟給的定義是，rx是一個函數庫，讓開發者可以利用可觀察序列和linq風格查詢操作符來編寫異步和基于事件的程式，使用rx，開發者可以用observables表示異步資料流，用linq操作符查詢異步資料流， 用schedulers參數化異步資料流的并發處理，rx可以這樣定義：rx = observables + linq + schedulers。 reactivex.io給的定義是，rx是一個使用可觀察資料流進行異步程式設計的程式設計接口，reactivex結合了觀察者模式、疊代器模式和函數式程式設計的精華。 很多公司都在使用reactivex，例如microsoft、netflix、github、trello、soundcloud。 reactivex不僅僅是一個程式設計接口，它是一種程式設計思想的突破，它影響了許多其它的程式庫和架構以及程式設計語言。 建立：rx可以友善的建立事件流和資料流 組合：rx使用查詢式的操作符組合和變換資料流 監聽：rx可以訂閱任何可觀察的資料流并執行操作 函數式風格：對可觀察資料流使用無副作用的輸入輸出函數，避免了程式裡錯綜複雜的狀态 簡化代碼：rx的操作符通通常可以将複雜的難題簡化為很少的幾行代碼 異步錯誤處理：傳統的try/catch沒辦法處理異步計算，rx提供了合适的錯誤處理機制 輕松使用并發：rx的observables和schedulers讓開發者可以擺脫底層的線程同步和各種并發問題 rx擴充了觀察者模式用于支援資料和事件序列，添加了一些操作符，它讓你可以聲明式的組合這些序列，而無需關注底層的實作：如線程、同步、線程安全、并發資料結構和非阻塞io。 observable通過使用最佳的方式通路異步資料序列填補了這個間隙 單個資料 多個資料 同步 <code>t getdata()</code> <code>iterable&lt;t&gt; getdata()</code> 異步 <code>future&lt;t&gt; getdata()</code> <code>observable&lt;t&gt; getdata()</code> rx的observable模型讓你可以像使用集合資料一樣操作異步事件流，對異步事件流使用各種簡單、可組合的操作。 對于單層的異步操作來說，java中future對象的處理方式是非常簡單有效的，但是一旦涉及到嵌套，它們就開始變得異常繁瑣和複雜。使用future很難很好的組合帶條件的異步執行流程（考慮到運作時各種潛在的問題，甚至可以說是不可能的），當然，要想實作還是可以做到的，但是非常困難，或許你可以用<code>future.get()</code>，但這樣做，異步執行的優勢就完全沒有了。從另一方面說，rx的observable一開始就是為組合異步資料流準備的。 rx的observable不僅支援處理單獨的标量值（就像future可以做的），也支援資料序列，甚至是無窮的資料流。<code>observable</code>是一個抽象概念，适用于任何場景。observable擁有它的近親iterable的全部優雅與靈活。 observable是異步的雙向push，iterable是同步的單向pull，對比： 事件 iterable(pull) observable(push) 擷取資料 <code>t next()</code> <code>onnext(t)</code> 異常處理 throws <code>exception</code> <code>onerror(exception)</code> 任務完成 <code>!hasnext()</code> <code>oncompleted()</code> rx對于對于并發性或異步性沒有任何特殊的偏好，observable可以用任何方式實作，線程池、事件循環、非阻塞io、actor模式，任何滿足你的需求的，你擅長或偏好的方式都可以。無論你選擇怎樣實作它，無論底層實作是阻塞的還是非阻塞的，用戶端代碼将所有與observable的互動都當做是異步的。 observable是如何實作的？