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