Java SE -聚合操作[Aggregate Operaitions]

爱被打了一巴掌 2024-02-18 15:09 26阅读 0赞

[Aggregate Operations][]

## 1. 聚合操作 ##

使用collection的操作，不是存完数据就不管了，而是要用其中的元素。（聚合操作是对collection进行的操作）

#### 1.1.1管道和流 ####

管道是一系列聚合操作的集合。例如在collection中通过聚合操作filter和forEach，打印符合特定目标的成员：

roster
        .stream()
        .filter(e -> e.getGender() == Person.Sex.MALE)
        .forEach(e -> System.out.println(e.getName()));

管道包含成员：  
\- 一个源。可以是collection，可以是array，一个generator函数，IO channel。  
\-0个或多个中间操作。**中间操作产生一个新的流**，例如`filter`  
\-**一个终端操作（terminal operation）。产生一个非流结果，可能是一个基本类型的值，collection，或者不返回值（`forEach`）。**

double average = roster
      .stream()
      .filter(p -> p.getGender() == Person.Sex.MALE)
      .mapToInt(Person::getAge)
      .average()
      .getAsDouble();

`mapToInt`返回一个新的类型为`IntStream`（只包含Integer类型）的流，`average`返回`OptionalDouble`，如果roster没有元素，`average`返回空的`OptionalDouble`，引起`getAsDouble`抛出`NoSuchElementException`。JDK包含很多**终端操作，average合并流的内容并且返回一个结果，这种类型的操作叫做reduction operation.**

#### 1.1.2 聚合操作和迭代器的区别 ####

聚合操作，像forEach，看起来像迭代器。本质区别：  
\- 使用内部迭代。聚合操作没有next方法来告诉它们如何处理collection的下一个元素。内部的意思是，用户只提供collection；外部的意思是，用户不仅提供collection，还要提供迭代的方式。外部迭代只能顺序迭代，内部迭代没有这个限制。因此内部迭代更适合并行计算，即将一个问题分成若干子问题，同时解决，并将子问题的结果合并。  
\-处理流的元素。聚合操作处理流的元素，也可以叫做**流操作**。  
\-支持参数。对于大部分聚合操作，可以指定lambda表达式作为参数（这在java中也叫做回调方法）。

### 1.2. 化简（Reduction） ###

JDK包含很多终端操作，例如average，sum，min，max和count，它们**通过合并流的内容，返回一个值，这些操作叫做化简操作。**除了一个值，jdk也提供返回collection的操作。许多化简操作完成特定任务，例如求平均值、分类聚合。特定任务外，JDK也提供通用化简操作，`reduce`（一个值），`collect`（集合）。

#### 1.2.1 Stream.reduce ####

roster.stream()
      .mapToInt(Person::getAge)
      .sum();
    
    roster.stream()
      .map(Person::getAge)
      .reduce(0,(a,b) -> a + b);

reduce操作的两个参数：  
\-identity元素：reduction的初始值和默认结果，如果流没有元素。这个例子中是0，它是ages和的初始值，也是如果roster集合没有成员时的默认值。  
\-accumulator函数：accumulator函数需要两个参数：reduction目前的部分计算结果，以及流中的下一个元素，返回值是新的部分计算结果。这个例子中，accumulator函数是lambda表达式，计算两个Integer的和并返回一个Integer值。  
`(a, b) -> a + b`  
`reduce`操作总是返回一个新的值。然而，accumulator函数每次处理一个流元素后返回一个新值。假设你想精简流元素得到一个更复杂的对象，例如collection。这个可能阻碍应用的表现：如果你的reduce操作包含增加元素到collection中，每次accumulator函数处理一个元素，它都创建一个新的collection包含该元素，效率差。这时，更新一个已经存在的collection更有效率。这就是Stream.collect方法。

#### 1.2.2 Stream.collect ####

不像reduce方法，总是创建一个新值，`collect`方法改变或者mutate，一个已经存在的值。

求流数据的平均值，需要两块数据：总数和这些值的和。就像包括`reduce`在内的所有化简方法，`collect`方法只返回一个值。可以创建一个新的数据类型，它包含成员变量，即总的数据量以及它们的和：

class Averager implements IntConsumer
    {
      private int total = 0;
      private int count = 0;
      public double average() {
        return count > 0 ? ((double) total)/count : 0;
      }
    
      public void accept(int i) {total += i; count++; }
    
      public void combine(Averager other) {
        total += other.total;
        count += other.count;
      }
    }

下面的管道用上面的类，以及`collect`方法来计算年龄平均值：

Averager avergeCollect = roster.stream()
      .filter(p->p.getGender() == Person.Sex.MALE)
      .map(Person::getAge)
      .collect(Averager::new, Averager::accept, AVerager::combine);
    
    avergeCollect.average(); // average value.

`collect`在该例中有三个参数：  
\-`supplier`：supplier是一个工厂函数；构造新的实例。对于collect操作，它创建结果容器的实例。这个例子中，它是Averager类的新的实例。  
\-`accumulator`：accumulator函数将流元素整合到结果容器中。这个例子中，它改变Averager结果容器，通过增加count变量，并将当前的流元素的值加到total成员变量中。  
\-`combiner`：combiner函数接受两个结果容器做参数，并融合它们的内容。这个例子中，它改变`Averager`结果容器，通过合并`count`，合并`total`。  
注意：  
\-`supplier`是lambda表达式（或方法引用）而不是`reduce`中的`identity`元素。  
\-`accumulator`和`combiner`函数不返回值。  
\-`collect`操作可以用于并行流（如果运行在并行流中用`collect`方法，不管`combiner`什么时候创建一个新的对象比如这个例子中的`Averager`对象，JDK都会创建一个新的线程，其结果是你不需要担心同步）  
尽管JDK提供`average`操作用来计算流元素的平均值，可以用`collect`操作和一个自定义的类实现。  
`collect`最适合集合操作。

##### exp 1 #####

List<String> names = roster
      .stream()
      .filter(p -> p.getGender() == Person.Sex.MALE)
      .map(p -> p.getName())
      .collect(Collectors.toList());

上面例子用到的collect版本，接受一个类型为`Collector`的参数。该类封装了`collect`需要的三个参数。  
`Collectors`类包含许多有用的化简操作，例如积累元素到集合中，或者归类。这些化简操作返回`Collector`类的实例，可以用来作为collect操作的参数。  
这个例子用`Collectors.toList`，积累流元素到一个新的List实例。和大部分`Collectors`类的操作一样，`toList`方法返回`Collector`实例，而不是collection。

##### exp2 #####

Map<Person.Sex, List<Person>> byGender = roster
      .stream()
      .collect(Collectors.groupingBy(Person::getGender));

`groupingBy`操作返回`map`，key是lambda表达式（这里叫分类函数，classification function）的返回值，这个例子中，返回的`map`包含两个key，`Person.Sex.MALE`和`Person.Sex.FEMALE`。key的相应的value，是List实例，该List实例的元素由分类函数处理的流元素组成。

public static <T, K> Collector<T, ?, Map<K, List<T>>>
        groupingBy(Function<? super T, ? extends K> classifier)  {...}

##### exp3 #####

Map<Person.Sex, List<Person>> namesByGender = roster
      .stream()
      .collect(
        Collectors.groupingBy(
          Person::getGender,
          Collectors.mapping(
            Person::getName,
            Collectors.toList())));

`groupingBy`操作接受两个参数，一个分类函数和一个Collector的实例。Collector参数又被称为downstream collector。这是java运行时应用于另一个collector结果的collector。因此，此`groupingBy`操作使您能够将`collect`方法应用于由`groupingBy`运算符创建的列表值。 此示例应用collector `mapping`，它将映射函数`Person::getName`应用于流的每个元素。 因此，结果流仅由成员的名称组成。包含一个或更多个downstream collector的管道，像这个例子，被称为多级化简（multilevel reduction）。返回namesByGender key为Person.Sex的列表的值，value为该性别的人的名字的列表。

Collector<T, ?, Map<K, D>> groupingBy(Function<? super T, ? extends K> classifier,
                                              Collector<? super T, A, D> downstream) {...}

##### exp4 #####

Map<Person.Sex, Integer> totalAgeByGender =
        roster
            .stream()
            .collect(
                Collectors.groupingBy(
                    Person::getGender,                      
                    Collectors.reducing(
                        0,
                        Person::getAge,
                        Integer::sum)));

reducing接受三个参数：  
\-identity，像Stream.reduce操作，identity元素既是初始值也是流没有元素时的默认结果。  
\-mapper：reducing操作将映射函数作用域所有流元素。  
\-operation：operation函数用于化简映射得到的值。

Map<Person.Sex, Double> averageAgeByGender = roster
      .stream()
      .collect(
        Collectors.groupingBy(
          Person::getGender,
          Collectors.averagingInt(Person::getAge)));

##### exp5 #####

此外，Collectors还有partitioningBy方法。

Map<Boolean, List<Foo>> man = Stream.of(t, j, a)
                .collect(Collectors.partitioningBy(foo -> foo.getGender().equals("man")));

### 1.3. 并行（Parallelism） ###

并行计算就是将问题分解成若干同时计算的子问题（每个子问题在一个单独的线程中运行），将每个子问题的解决结果合并成最终结果。Java SE提供fork/join框架，辅助实现并行计算。使用该框架，必须指定问题如何被划分。通过聚合操作，java运行时将为你划分和组合计算结果。

使用collection实现并行的问题是collection并不是线程安全的，多线程操作collection将出现线程干扰（thread interference）或内存一致性差错（memory consistency errors）。collection框架提供同步外壳(synchronization wrapper)，用来给任意的collection添加自动同步，使其线程安全。然而，synchronization将导致线程竞争（thread contention）。线程竞争阻碍线程的并行运行。聚合操作和并行流，实现了非线程安全的集合的并行执行，前提是在操作时不修改集合。

并行执行不一定比顺序执行速度更快，这取决于数据和处理器内核。虽然聚合操作使实现并行更容易，但是否需要用并行解决问题取决于程序员。

[例子][Link 1]

#### 1.3.1 并行执行流 ####

当流并行执行时，java运行时会把流分成若干个子流，聚合操作并行地遍历并处理所有子流并合并结果。  
总是创建顺序流，除非用Collection.parallelStream创建并行的流。BaseStream.parallel也可以创建并行流。下面的例子并行的计算所有男性成员的平均年龄：

double average = roster
        .parallelStream()
        .filter(p -> p.getGender() == Person.Sex.MALE)
        .mapToInt(Person::getAge)
        .average()
        .getAsDouble();

#### 1.3.2 Concurrent Reduction ####

Map<Person.Sex, List<Person>> byGender = 
        roster
            .stream()
            .collect( Collectors.groupingBy(Person::getGender));

等效并行：

ConcurrentMap<Person.Sex, List<Person>> byGender = 
        roster
            .parallelStream()
            .collect( Collectors.groupingByConcurrent(Person::getGender));

对于包含collect操作的管道（pipeline），如果下列同时成立，java运行时表现出concurrent reduction：  
1. 流是并行的（parallel stream）  
2. collector具有`Collector.Characteristics.CONCURRENT`, 这可以通过`Collector.characteristics`方法获得。  
3. 无序。要么流是无序的，要么collector是无序的（`Collector.Characteristics.UNORDERED`）。这可以通过`BaseStream.unordered`来获得。

这个例子返回`ConcurrentMap`实例，不像`groupingByConcurrent`，`groupingBy`对于并行流表现很糟，这是因为后者用key融合两个map，开销昂贵。类似地，`Collectors.toConcurrentMap`对于并行流表现比`Collectors.toMap`好。

#### 1.3.3 顺序 ####

管道处理流元素的顺序，取决于流是串行的还是并行的、流的源、中间操作。

Integer[] intArray = {
       1,2,3};
    List<Integer> listOfIntegers = new ArrayList<>(Arrays.asList(intArray));
    listOfIntegers.stream().forEach(e -> System.out.print(e + " ")); // 1 2 3 按照加入到队列中的顺序打印
    listOfIntegers
        .parallelStream()
        .forEach(e -> System.out.print(e + " ")); // 3 1 2 随机顺序
    listOfIntegers
        .parallelStream()
        .forEachOrdered(e -> System.out.print(e + " ")); // 1 2 3，

按照数据源的顺序打印，失去并行流的作用  
流操作使用内部遍历，java编译器和运行时，决定处理流元素的顺序，来最大化有利于并行计算，除非由流操作指定顺序。

#### 1.3.4 副作用 ####

方法或表达式有副作用：除了返回或者产生一个值，它也会改变计算机的状态。包括mutable reduction（使用collect操作）以及调用`System.out.println`方法debug。JDK在管道中处理特定的副作用。特别地，`collect`方法被设计成以并行安全的方式执行具有副作用的最常见的流操作。`forEach`和`peek`被设计成有副作用，lambda返回`void`，例如调用`System.out.println`也有副作用。应该小心使用`forEach`和`peek`操作；在并行流中使用这些方法，java运行时可能在多个线程中同时调用作为它们参数的lambda表达式。额外地，永远不要传递有副作用的lambda表达式给`filter`和`map`方法。下面的章节讨论interference和有状态lambda表达式，它们都产生副作用并且可能返回不一致的或者难以预料的结果，特别是并行刘中。首先讨论laziness，因为它直接作用于interference。

##### 1.3.4.1. Laziness #####

左右中间操作都是懒惰的。表达式、方法或算法只有在被需要时才执行。（算法是eager如果它被理解执行）。**中间操作是懒惰的，因为直到流的终端操作（terminal operation）开始时，它们才开始处理流的内容。**这允许java编译器和运行时优化它们处理流的方式。在一个类似filter-mapToInt-average管道中，`average`操作获得最开始的几个`mapToInt`结果，`average`操作重复这一过程，直到它获得所有需要的元素，然后才开始计算平均值。

##### 1.3.4.2. Interference #####

流操作的lambda表达式不应当interfere，interference发生在流的源在管道处理流时改变的时候，例如下面的例子抛`ConcurrentModificationException`：

List<String> list = new ArrayList<>(Arrays.asList("one", "two"));
      String concatenatedString = list.stream()
        .peek(s -> list.add("three")) // don't do this! Interference occurs here.
        .reduce((a,b) -> a + " " + b)
        .get();

通过`reduce`接续字符串，返回`Optional<String>`，`reduce`是一个终止操作。管道调用中间操作`peek`，后者向list末尾添加新元素。因为所有中间操作都是懒惰的，这意味着当`get`方法调用时，这个例子中的管道才开始执行，并且在`get`方法完成时执行结束。`peek`操作的参数在管道执行时，试图改变stream源，这引起java运行时抛出`ConcurrentModificationException`。

##### 1.3.4.3. 有状态lambda表达式 #####

避免使用有状态的lambda作为流操作参数。**有状态的表达式，它的执行结果依赖于状态，该状态在管道执行的过程中可能变更。**

List<Integer> serialStorage = new ArrayList<>();
      // serial stream.
      listOfIntegers.stream()
        // don't do this! it uses a stateful lambda expression.
        .map(e -> {serialStorage.add(e); return e;})
        .forEachOrdered(e -> System.out.print(e + " ")); // 1 2 3
    
      serialStorage
        .stream()
        .forEachOrdered(e -> System.out.print(e + " ")); // 1 2 3
    
      // parallel stream.
      List<Integer> parallelStorage = Collections.synchronizedList(new ArrayList<>());
      listOfIntegers.parallelStream()
        // don't do this! it uses a stateful lambda expression.
        .map(e -> {parallelStorage.add(e); return e;})
        .forEachOrdered(e -> System.out.print(e + " ")); // 3 1 2
    
      parallelStorage
        .stream()
        .forEachOrdered(e -> System.out.print(e + " ")); // 2 3 1

不管流是顺序的还是并行的，`forEachOrdered`  
`e -> {parallelStorage.add(e); return e;}` 是有状态的操作，要想获得确定的和可预料的结果，需要确保stream操作的lambda表达式参数不是有状态的。  
注意：上面的例子调用`synchronizedList`，所以`parallelStorage`是线程安全的`List`。但是要记住，collections不是线程安全的，这意味着多个线程不应该同时访问一个特定的collection。如果用`new ArrayList()<>()`方法初始化`parallelStorage`，多个线程访问和改变`parallelStorage`，它不具有同步机制来调度特定线程何时可以访问list实例，因此可能有错误的结果：  
`Parallel stream： 3 1 2 null 2 1`

--------------------

[Aggregate Operations]: https://docs.oracle.com/javase/tutorial/collections/streams/index.html
[Link 1]: https://docs.oracle.com/javase/tutorial/collections/streams/examples/ParallelismExamples.java