RxJava 學習筆記<七> Aggregation

咱們已經看到了如何切掉咱們不想要的 sequence 的部分，如何得到單個值，以及如何檢查 sequence 的內容。這些東西能夠看做是關於包含 sequence 的推理。如今，咱們將看到如何使用 sequence 中的數據來導出新的有意義的值。

咱們將在這裏看到的方法相似於所謂的質變（catamorphism）。在咱們的例子中，這意味着方法消耗序列中的值並將它們組合成一個值。可是，它們並不嚴格知足定義，由於它們不返回單個值。相反，它們返回一個 observable 有望發出單一值。

若是你已經閱讀了全部的例子，你應該注意到一些重複。爲了消除這種狀況並將重點放在重要的事情上，咱們如今將引入一個自定義 Subscriber ，咱們將在示例中使用它

這是一個很是基本的實現，它將每一個事件打印到控制檯。

count

咱們的第一種方法是計數。它的用途與大多數Java容器中的長度和大小相同。此方法將返回一個可觀察到的值，該值將等待 sequence 完成併發出所遇到的值的數目。

    輸出：

對於可能超過標準整數容量的序列也有CountLong。

first

First將返回一個 observable，該值僅發出序列中的第一個值。它與Take(1)相似，只是若是沒有找到，它將發出java.util.NoSuchElementException。若是使用接受謂詞的重載，則返回與謂詞匹配的第一個值。

輸出：

爲了規避 java.util.NoSuchElementException 異常，咱們可使用 firstOrDefault 方法來設置默認值。

last

last 和 lastOrDefault 工做方式與 first 相同，只是返回的項是序列完成前的最後一項。當將重載與謂詞一塊兒使用時，返回的項是與謂詞匹配的最後一項。

single

Single 發出序列中惟一的值，或者在給定時惟一知足謂詞的值。它與第一個和最後一個不一樣之處在於，它不會忽略多個匹配。若是找到多個匹配項，它將發出錯誤。它能夠用來斷言序列必須只包含一個這樣的值。

記住，Single 必須檢查整個序列，以確保您的斷言。

輸出：

經過前面的方法，咱們知道能夠設置默認值 singleOrDefault

Custom aggregators

到目前爲止，咱們在本章中看到的方法彷佛與前幾章中的方法沒有什麼不一樣。咱們如今將看到兩種很是強大的方法，它們將極大地擴展咱們能夠利用 observable 來作的事情。到目前爲止，咱們看到的許多方法均可以使用這些方法來實現。

reduce

您可能據說過[MapReduce](https：//en.wikipinea.org/wiki/MapReduce)中的Report。或者，您也能夠將其命名爲「聚合」、「累積」或「摺疊」。通常的想法是，經過一次合併兩個值，從多個值中產生一個值。在其最基本的重載中，您所須要的只是一個將兩個值組合成一個值的函數。

public final Observable<T> reduce(Func2<T,T,T> accumulator)

這最好用一個例子來解釋。這裏咱們將計算一個整數序列的和：0+1+2+3+4+...。咱們還將計算不一樣示例的最小值；

輸出：

Rx中的 reduce 與並行系統中的「reduce」並不相同。在並行系統中，它意味着值對能夠任意選擇，從而使多臺機器能夠獨立工做。在Rx中，累加器函數從左到右依次應用(如圖所示)。每次，累加器函數將上一步的結果與下一個值結合起來。這在另外一個重載中更爲明顯：

public final <R> Observable<R> reduce(R initialValue, Func2<R,? super T,R> accumulator)

累加器返回的類型與可觀察到的類型不一樣。累加器的第一個參數是前一個累加過程的部分結果，第二個參數是下一個值。要開始這個過程，須要提供一個初始值。咱們將經過從新實現Count來證實這一點的有用性。

輸出：

咱們從累加器0開始，由於咱們已經計算了0項。每當新項目到達時，咱們返回一個新的累加器，該累加器增長了一個。最後一個值對應於源序列中的元素數。

reduce 可用於實現發出單個值的大多數運算符的功能。它不能實如今源完成以前發出值的行爲。所以，您可使用 reduce 實現 last，但 all 的實現不會徹底像原來的那樣。

scan

scan 很是相似於 reduce，關鍵的區別是 scan 將發出全部中間結果。

public final Observable<T> scan(Func2<T,T,T> accumulator)

在咱們的求和示例中，使用 Scan 將生成一個運行求和。

輸出：

Scan 比 reduce 更通用，由於 Reduce 能夠用Scan實現：REPLE(Acc)=Scan(Acc).Take Lasts()

scan 在源發出時發出，不須要源來完成。咱們經過實現返回運行最小值的可觀察值來證實：

輸出：

Aggregation to collections

沒有什麼能夠阻止你的累加器成爲一個集合。您可使用r educe 將Observable<T><轉List<T>中。

輸出：

上面的代碼在形式上有一個問題：reduce 是一個函數摺疊，而這樣的摺疊不該該用於可變累加器。若是咱們以「正確」的方式執行此操做，則必須爲每一個新項建立ArrayList<Integer>的新實例，以下所示：

collect

爲每項新項目建立新集合的性能是不可接受的。出於這個緣由，Rx 提供了 collect 操做符，它與 reduce 同樣，只使用可變累加器。經過使用 collect ，您不遵循不可修改的約定，而且您還簡化了代碼：

輸出：

一般，您沒必要手動收集值。RxJava提供了多種操做符，用於將序列收集到容器中。這些聚合器返回一個operators集合，當它準備好時，它將發出相應的集合，就像咱們在這裏所作的那樣。接下來，咱們將看到這樣的聚合器。

toList

輸出：

toSortedList

toSortedList 工做方式相似於 toList，顧名思義，就是結果是排序的。

輸出：

toMap

toMap 把咱們的 sequence 轉換成 Map<TKey,T> 

輸出：

 

groupBy

這是Rx對Multimap的操做方式。對於每一個值，它計算一個鍵，並根據該鍵將值分組爲單獨的可觀測值。

輸出：

nest

當處理nested observables，Nest 操做符變得頗有用。它容許你把一個 non-nested observable  變成一個nested 。

輸出：

 

原文連接：

https://github.com/Froussios/Intro-To-RxJava/blob/master/Part%202%20-%20Sequence%20Basics/4.%20Aggregation.md