java - 我应该如何对流进行汇总？

我已经看过并尝试过如何在流中求和的各种不同实现。这是我的代码:

List<Person> persons = new ArrayList<Person>();

for(int i=0; i < 10000000; i++){
    persons.add(new Person("random", 26));
}

Long start = System.currentTimeMillis();
int test = persons.stream().collect(Collectors.summingInt(p -> p.getAge()));
Long end = System.currentTimeMillis();
System.out.println("Sum of ages = " + test + " and it took : " + (end - start) + " ms with collectors");

Long start3 = System.currentTimeMillis();
int test3 = persons.parallelStream().collect(Collectors.summingInt(p -> p.getAge()));
Long end3 = System.currentTimeMillis();
System.out.println("Sum of ages = " + test3 + " and it took : " + (end3 - start3) + " ms with collectors and parallel stream");


Long start2 = System.currentTimeMillis();
int test2 = persons.stream().mapToInt(p -> p.getAge()).sum();
Long end2 = System.currentTimeMillis();
System.out.println("Sum of ages = " + test2 + " and it took : " + (end2 - start2) + " ms with map and sum");

Long start4 = System.currentTimeMillis();
int test4 = persons.parallelStream().mapToInt(p -> p.getAge()).sum();
Long end4 = System.currentTimeMillis();
System.out.println("Sum of ages = " + test4 + " and it took : " + (end4 - start4) + " ms with map and sum and parallel stream");

这给了我以下结果:

Sum of ages = 220000000 and it took : 110 ms with collectors
Sum of ages = 220000000 and it took : 272 ms with collectors and parallel stream
Sum of ages = 220000000 and it took : 137 ms with map and sum
Sum of ages = 220000000 and it took : 134 ms with map and sum and parallel stream

我尝试了几次，每次都给我不同的结果(大多数情况下，最后一个解决方案是最好的)，所以我想知道:

1)正确的方法是什么？

2)为什么呢？ (与其他解决方案有什么区别？)

最佳答案

在我们获得实际答案之前，您应该了解以下几点:

测试结果的差异可能很大，具体取决于许多因素(例如，您正在运行的计算机)。这是在8核计算机上一次运行的结果:

Sum of ages = 260000000 and it took : 94 ms with collectors
Sum of ages = 260000000 and it took : 61 ms with collectors and parallel stream
Sum of ages = 260000000 and it took : 70 ms with map and sum
Sum of ages = 260000000 and it took : 94 ms with map and sum and parallel stream

然后在以后的运行中:

Sum of ages = 260000000 and it took : 68 ms with collectors
Sum of ages = 260000000 and it took : 67 ms with collectors and parallel stream
Sum of ages = 260000000 and it took : 66 ms with map and sum
Sum of ages = 260000000 and it took : 109 ms with map and sum and parallel stream

Micro基准测试不是一个容易的话题。有一些方法可以做到这一点(稍后再介绍)，但是在大多数情况下，仅尝试使用System.currentTimeMillies()将无法可靠地工作。

仅仅因为Java 8使并行操作变得容易，并不意味着应该在所有地方都使用它们。并行操作在某些情况下有意义，而在其他情况下则没有意义。

好的，现在让我们看看您使用的各种方法。

顺序收集器:您使用的summingInt收集器具有以下实现:

public static <T> Collector<T, ?, Integer> summingInt(ToIntFunction<? super T> mapper) {
    return new CollectorImpl<>(
            () -> new int[1],
            (a, t) -> { a[0] += mapper.applyAsInt(t); },
            (a, b) -> { a[0] += b[0]; return a; },
            a -> a[0], Collections.emptySet());
}

因此，首先将创建一个具有一个元素的新数组。然后，对于流中的每个Person元素，collect函数将使用Person#getAge()函数将年龄作为Integer(而不是int!)检索，并将该年龄添加到先前的年龄(在1D数组中)。最后，在处理完所有流之后，它将从该数组中提取值并将其返回。因此，这里有很多自动装箱和-unboxing操作。

并行收集器:这使用ReferencePipeline#forEach(Consumer)函数来累积从映射函数获得的年龄。再次有很多自动装箱和-unboxing。

顺序图和总和:在这里，您将Stream<Person>映射到IntStream。这意味着一件事是，不再需要自动装箱或-unboxing；在某些情况下，这可以节省大量时间。然后使用以下实现对结果流求和:

@Override
public final int sum() {
    return reduce(0, Integer::sum);
}

此处的reduce函数将调用ReduceOps#ReduceOp#evaluateSequential(PipelineHelper<T> helper, Spliterator<P_IN> spliterator)。
本质上，这将对所有数字使用Integer::sum函数，从0和第一个数字开始，然后从第二个数字开始，以此类推。

并行映射和求和:这里的事情变得很有趣。它使用相同的sum()函数，但是在这种情况下，reduce将调用ReduceOps#ReduceOp#evaluateParallel(PipelineHelper<T> helper, Spliterator<P_IN> spliterator)而不是顺序选项。基本上，这将使用分而治之的方法将这些值相加。现在，分而治之的最大优势当然是可以轻松并行完成。但是，它确实需要多次拆分和重新加入流，这会浪费时间。因此，取决于与元素相关的实际任务的复杂性，它的运行速度可能会有很大差异。在添加的情况下，在大多数情况下可能不值得。从我的结果可以看出，它始终是较慢的方法之一。

现在，要真正了解需要花多长时间，让我们做一个适当的微型基准测试。我将使用带有以下基准代码的JMH:

package com.stackoverflow.user2352924;

import org.openjdk.jmh.annotations.*;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

@BenchmarkMode(Mode.Throughput)
@OutputTimeUnit(TimeUnit.MINUTES)
@Warmup(iterations = 5, time = 5, timeUnit = TimeUnit.SECONDS)
@Measurement(iterations = 10, time = 10, timeUnit = TimeUnit.SECONDS)
@State(Scope.Benchmark)
@Fork(1)
@Threads(2)
public class MicroBenchmark {

    private static List<Person> persons = new ArrayList<>();

    private int test;

    static {
        for(int i=0; i < 10000000; i++){
            persons.add(new Person("random", 26));
        }
    }

    @Benchmark
    public void sequentialCollectors() {
        test = 0;
        test += persons.stream().collect(Collectors.summingInt(p -> p.getAge()));
    }

    @Benchmark
    public void parallelCollectors() {
        test = 0;
        test += persons.parallelStream().collect(Collectors.summingInt(p -> p.getAge()));
    }

    @Benchmark
    public void sequentialMapSum() {
        test = 0;
        test += persons.stream().mapToInt(p -> p.getAge()).sum();
    }

    @Benchmark
    public void parallelMapSum() {
        test = 0;
        test += persons.parallelStream().mapToInt(p -> p.getAge()).sum();
    }

}

这个Maven项目的pom.xml看起来像这样:

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>com.stackoverflow.user2352924</groupId>
    <artifactId>StackOverflow</artifactId>
    <version>1.0</version>
    <packaging>jar</packaging>

    <name>Auto-generated JMH benchmark</name>

    <prerequisites>
        <maven>3.0</maven>
    </prerequisites>

    <dependencies>
        <dependency>
            <groupId>org.openjdk.jmh</groupId>
            <artifactId>jmh-core</artifactId>
            <version>${jmh.version}</version>
        </dependency>
        <dependency>
            <groupId>org.openjdk.jmh</groupId>
            <artifactId>jmh-generator-annprocess</artifactId>
            <version>${jmh.version}</version>
            <scope>provided</scope>
        </dependency>
    </dependencies>

    <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <jmh.version>0.9.5</jmh.version>
        <javac.target>1.8</javac.target>
        <uberjar.name>benchmarks</uberjar.name>
    </properties>

    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>3.1</version>
                <configuration>
                    <compilerVersion>${javac.target}</compilerVersion>
                    <source>${javac.target}</source>
                    <target>${javac.target}</target>
                </configuration>
            </plugin>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-shade-plugin</artifactId>
                <version>2.2</version>
                <executions>
                    <execution>
                        <phase>package</phase>
                        <goals>
                            <goal>shade</goal>
                        </goals>
                        <configuration>
                            <finalName>microbenchmarks</finalName>
                            <transformers>
                                <transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
                                    <mainClass>org.openjdk.jmh.Main</mainClass>
                                </transformer>
                            </transformers>
                        </configuration>
                    </execution>
                </executions>
            </plugin>
        </plugins>
        <pluginManagement>
            <plugins>
                <plugin>
                    <artifactId>maven-clean-plugin</artifactId>
                    <version>2.5</version>
                </plugin>
                <plugin>
                    <artifactId>maven-deploy-plugin</artifactId>
                    <version>2.8.1</version>
                </plugin>
                <plugin>
                    <artifactId>maven-install-plugin</artifactId>
                    <version>2.5.1</version>
                </plugin>
                <plugin>
                    <artifactId>maven-jar-plugin</artifactId>
                    <version>2.4</version>
                </plugin>
                <plugin>
                    <artifactId>maven-javadoc-plugin</artifactId>
                    <version>2.9.1</version>
                </plugin>
                <plugin>
                    <artifactId>maven-resources-plugin</artifactId>
                    <version>2.6</version>
                </plugin>
                <plugin>
                    <artifactId>maven-site-plugin</artifactId>
                    <version>3.3</version>
                </plugin>
                <plugin>
                    <artifactId>maven-source-plugin</artifactId>
                    <version>2.2.1</version>
                </plugin>
                <plugin>
                    <artifactId>maven-surefire-plugin</artifactId>
                    <version>2.17</version>
                </plugin>
            </plugins>
        </pluginManagement>
    </build>

</project>

确保Maven也与Java 8一起运行，否则您将收到难看的错误。

在这里我不会详细介绍如何使用JMH(在其他地方也可以使用)，但这是我得到的结果:

# Run complete. Total time: 00:08:48

Benchmark                                     Mode  Samples     Score  Score error    Units
c.s.u.MicroBenchmark.parallelCollectors      thrpt       10  3658,949      775,115  ops/min
c.s.u.MicroBenchmark.parallelMapSum          thrpt       10  2616,905      221,109  ops/min
c.s.u.MicroBenchmark.sequentialCollectors    thrpt       10  5502,160      439,024  ops/min
c.s.u.MicroBenchmark.sequentialMapSum        thrpt       10  6120,162      609,232  ops/min

因此，在我运行这些测试时，在我的系统上，顺序映射总和要快得多，在并行映射总和(使用分治法)仅可以完成2600多次的同时，可以完成6100多次操作实际上，顺序方法都比并行方法快得多。

现在，在可以更轻松地并行运行的情况下-例如Person#getAge()函数比 getter 要复杂得多-并行方法可能是更好的解决方案。最后，这完全取决于被测试案例中并行运行的效率。

要记住的另一件事:如果有疑问，请执行适当的微观基准测试。 ;-)