java - 执行并行流 Java 8 时出现异常

Question

我正在编写一个函数来计算给定数字的数字平方和。我正在并行调用此函数以获取 1 到 1000 之间的数字。但有时我会遇到异常(exception)。

public static void main(String[] args) throws Exception {
    IntStream stream = IntStream.rangeClosed(1, 1000);
    Map < Integer, Integer > numbers = new TreeMap < > ();
    stream.parallel().forEach(i - > {
        int result = digitSquareSum(i);
        numbers.put(i, result);
    });
    Files.write(Paths.get("result.csv"), () - > numbers.entrySet().stream()
        . < CharSequence > map(e - > e.getKey() + "," + e.getValue())
        .iterator());
}

private static int digitSquareSum(int i) {
    int result = 0;
    int temp = 0;
    while (i != 0) {
        temp = i % 10;
        result = result + temp * temp;
        i = i / 10;
    }
    return result;
}

我遇到以下异常

Exception in thread "main" java.lang.NullPointerException
    at java.base/jdk.internal.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
    at java.base/jdk.internal.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:62)
    at java.base/jdk.internal.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
    at java.base/java.lang.reflect.Constructor.newInstance(Constructor.java:488)
    at java.base/java.util.concurrent.ForkJoinTask.getThrowableException(ForkJoinTask.java:590)
    at java.base/java.util.concurrent.ForkJoinTask.reportException(ForkJoinTask.java:668)
    at java.base/java.util.concurrent.ForkJoinTask.invoke(ForkJoinTask.java:726)
    at java.base/java.util.stream.ForEachOps$ForEachOp.evaluateParallel(ForEachOps.java:160)
    at java.base/java.util.stream.ForEachOps$ForEachOp$OfInt.evaluateParallel(ForEachOps.java:189)
    at java.base/java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:233)
    at java.base/java.util.stream.IntPipeline.forEach(IntPipeline.java:417)
    at java.base/java.util.stream.IntPipeline$Head.forEach(IntPipeline.java:574)

我的目标是编写一种有效的方法来计算数字的平方和并将其以升序方式写入csv文件。

Answer 1

正如评论中提到的，您会收到此异常，因为 TreeMap 不是线程安全的。来自 javadoc :

Note that this implementation is not synchronized. If multiple threads access a map concurrently, and at least one of the threads modifies the map structurally, it must be synchronized externally. (A structural modification is any operation that adds or deletes one or more mappings; merely changing the value associated with an existing key is not a structural modification.)

我相信您正在使用TreeMap来保留初始数字的顺序。但这不是必需的(并且还引入了 side-effects ):您可以使用 Collectors.toList()其中按照遇到的顺序将所有输入元素收集到一个列表中。

这还可以让您避免创建第二个流来将这些数字写入文件:

List<String> lines = IntStream.rangeClosed(1, 1000)
            .parallel()
            .mapToObj(i -> i + " -> " + digitSquareSum(i))
            .collect(Collectors.toList());
Files.write(Paths.get("result.csv", lines); // the lines will come in the right order