for循环优化Java并行计算多线程，有没有解决方案?

Question

for (int i = 0; i < data.length; i++) {
    data[i] = Math.sqrt(Math.pow(uData[i], 2) + Math.pow(vData[i], 2));
}

是需要计算风速
一次length 大概有6500个，但是要计算 1500个层次。 相当于要计算 1500 * 6500
目前是用for循环，大概全部完成需要250~300秒。

也试过手动创建多线程跑，计算时间并没有改善，不知道为什么。

private double[] parallelHandelWindData(double[] dataU, double[] dataV) throws InterruptedException {
        Integer length = dataU.length;
        double[] data = new double[length];
        //创建一个线程池
        ExecutorService executorService = new ThreadPoolExecutor(
                10,
                10,
                20,
                TimeUnit.SECONDS,
                new LinkedBlockingDeque<>(5),
                Executors.defaultThreadFactory(),
                new ThreadPoolExecutor.DiscardOldestPolicy());

        List<Runnable> tasks = new ArrayList<>();

        for (Integer i = 0; i < 8; i++) {
            int startIndex = length / 8 * i;
            int endIndex = length / 8 * (i + 1);
            double[] uDataBlock = ArrayUtil.sub(dataU, startIndex, endIndex);
            double[] vDataBlock = ArrayUtil.sub(dataV, startIndex, endIndex);
            tasks.addAll(Arrays.asList(() -> sqrtAndPowWindData(data, uDataBlock, vDataBlock, startIndex, endIndex)));
        }

        //逐个提交任务
        tasks.forEach(executorService::submit);

        executorService.shutdown();
        // 线程池最大有效执行时间
        executorService.awaitTermination(10, TimeUnit.MINUTES);

        return data;
    }

    private double[] sqrtAndPowWindData(double[] data, double[] uDataBlock, double[] vDataBlock, int startIndex, int endIndex) {
        for (int i = startIndex; i < endIndex; i++) {
            data[i] = Math.sqrt(Math.pow(uDataBlock[i-startIndex], 2) + Math.pow(vDataBlock[i-startIndex], 2));
        }
        return data;
    }

Answer 1

先确定问题在哪，再提问，你的代码中的 Math 这点计算量，完全不够当代计算机看的。

这是我模拟的代码，运行 1500 次，每次生成两组 6500 长度的 double 数组，两个数组执行你的方法，平均 350 毫秒执行完成，我的机器是 mac m1 pro，即使是再差的机器，也不会比这个数值高到哪里去。

    private static void sqrtAndPowWindData(Double[] uData, Double[] vData) {
        for (int i = 0; i < uData.length; i++) {
            Math.sqrt(Math.pow(uData[i], 2) + Math.pow(vData[i], 2));
        }
    }


    public static void main(String[] args) {
        long start = System.currentTimeMillis();

        for (int j = 0; j < 1500; j++) {
            List<Double> d1List = new ArrayList<>();
            List<Double> d2List = new ArrayList<>();
            for (int i = 0; i < 6500; i++) {
                d1List.add(RandomUtil.randomDouble());
                d2List.add(RandomUtil.randomDouble());
            }
            sqrtAndPowWindData(ArrayUtil.toArray(d1List, Double.class), ArrayUtil.toArray(d2List, Double.class));
        }
        System.out.println(System.currentTimeMillis() - start);

因此，只可能消耗大量时间的地方是你获取数据的部分。

Answer 2

看看数据结果能缓存不? 计算平方的. 计算过的值, 不要再次计算了, 直接保存起来, 用空间换时间.

Answer 3

同意@zxdposter 的回答，你先看看耗时究竟在哪
但从代码看的话，只是普通数学计算，量级也不大，不应该这么久

量级大的话可以考虑：Flink之类的框架来做

Answer 4

这个类应该可以用，代码包含了使用示例：

import java.util.Arrays;
import java.util.concurrent.*;
import java.util.function.BiFunction;
import java.util.function.Consumer;

public class ParallelArrayComputing {

  private final ExecutorService threadPool;

  public ParallelArrayComputing(int threadPoolSize) {
    threadPool = new ThreadPoolExecutor(
      threadPoolSize, threadPoolSize, 20, TimeUnit.SECONDS,
      new LinkedBlockingDeque<>(5), Executors.defaultThreadFactory(),
      new ThreadPoolExecutor.DiscardOldestPolicy()
    );
  }

  public void shutdownNow() {
    this.threadPool.shutdownNow();
  }

  public void shutdownAndWait() throws InterruptedException {
    this.threadPool.shutdown();
    this.threadPool.awaitTermination(1, TimeUnit.HOURS);
  }

  /**
   * 投喂数据
   *
   * @param u            数据1
   * @param v            数据2
   * @param calculation  计算方法
   * @param whenComplete 当完成计算时要做的事
   */
  public void feed(
    double[] u, double[] v,
    BiFunction<double[], double[], double[]> calculation,
    Consumer<double[]> whenComplete
  ) {
    if (u == null || v == null || u.length != v.length) {
      throw new IllegalArgumentException();
    }

    if (u.length == 0) {
      whenComplete.accept(new double[0]);
      return;
    }

    threadPool.execute(() -> {
      double[] result = calculation.apply(u, v);
      whenComplete.accept(result);
    });
  }

  //////////////////////////// 使用方法

  public static void main(String[] args) throws InterruptedException {

    // 定义计算方法
    BiFunction<double[], double[], double[]> calculation = (u, v) -> {
      double[] result = new double[u.length];
      for (int i = 0; i < result.length; i++) {
        result[i] = Math.sqrt(Math.pow(u[i], 2) + Math.pow(v[i], 2));
      }
      return result;
    };

    // 定义对结算结果的处理
    Consumer<double[]> outputResult =
      result -> System.out.println("Finished: " + Arrays.toString(result));

    // 初始化 ParallelArrayComputing 对象
    ParallelArrayComputing pac = new ParallelArrayComputing(3);

    // 投喂一组数据
    double[] u = new double[]{1, 2, 3};
    double[] v = new double[]{4, 5, 6};
    pac.feed(u, v, calculation, outputResult);

    // 等待所有计算完成后关闭
    pac.shutdownAndWait();
  }
}