java - 是否可以利用Stream API提供的并行性来调用固定数量的相互独立的方法？

Question

我有一个 Java 类，在其中使用 Stream 提供的并行性API。这很好地满足了我的目的，因为我的大部分数据输入都是流。然而，有一个地方的代码是这样的:

void aMethod() {
    double[] a = methodA();
    double[] b = methodB();
    double[] c = methodC();
    doSomething(a, b, c);
}

doSomething(double[] a, double[] b, double[] c) {
    // concatenates the three arrays, converts to parallel stream, and does stuff
}

对 methodA、methodB 和 methodC 的三个调用可以并发。有没有办法使用 Stream API 来并行运行？

Answer 1

您可以轻松创建方法引用的(并行)流，充当 double[] 数组的 Supplier。然后可以将它们映射到它们的返回值，并收集到一个列表中。大致如下:

import java.util.List;
import java.util.function.Supplier;
import java.util.stream.Collectors;
import java.util.stream.Stream;

public class ParallelStreamMethods {

    public static void main(String[] args) {
        Stream<Supplier<double[]>> s = Stream.of(
            ParallelStreamMethods::methodA, 
            ParallelStreamMethods::methodB, 
            ParallelStreamMethods::methodC);

        List<double[]> result = 
            s.parallel().map(r -> r.get()).collect(Collectors.toList());

        doSomething(result.get(0), result.get(1), result.get(2));
    }

    private static double[] methodA() {
        return getValues("methodA");
    }

    private static double[] methodB() {
        return getValues("methodB");
    }

    private static double[] methodC() {
        return getValues("methodC");
    }

    private static double[] getValues(String name) {
        System.out.println("Enter "+name);
        try {
            int n = 1000 + (int)(Math.random() * 500);
            Thread.sleep(n);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("Exit "+name);
        return new double[1000];
    }

    private static void doSomething(double[] a, double[] b, double[] c) {
        System.out.println("Doing something with "+a+", "+b+", "+c);
    }
}

请注意，收集器仅用于展示基本思想。根据您打算如何处理结果，可以以不同的方式解决这个问题。不过，评论

// concatenates the three arrays, converts to parallel stream, and does stuff

听起来有点可疑。您可能需要考虑使用 DoubleStream 而不是数组，然后使用 flatMap 它们来创建更大的 DoubleStream。