多线程使用Java在循环内并行执行该方法

Question

我有如下代码。在循环中，它正在执行方法“process”。它是按顺序运行的。我想并行运行这个方法，但它应该在循环中完成，这样我就可以在下一行求和了。i、 即使是并行运行，所有函数都应该在第二个for循环执行之前完成

如何在Jdk1中解决这个问题。7不是JDK1。8版本

public static void main(String s[]){
    int arrlen = 10;
    int arr[] = new int[arrlen] ;

    int t =0;
    for(int i=0;i<arrlen;i++){
        arr[i] = i;
        t = process(arr[i]);
        arr[i] = t;
    }

    int sum =0;
    for(int i=0;i<arrlen;i++){
        sum += arr[i];
    }
    System.out.println(sum);

}

public static int process(int arr){
    return arr*2;
}

Answer 1

基本上，您需要使用自Java1.5以来就存在的Executors和Futures（参见Java Documentation）

在下面的示例中，我创建了一个主类，它使用另一个助手类，该类的行为类似于要并行化的处理器

主类分为3个步骤：

1. 创建进程池并并行执行任务
2. 等待所有任务完成其工作
3. 收集任务的结果

出于说教的原因，我放了一些日志，更重要的是，我在每个流程的业务逻辑中放了一个随机等待时间，模拟流程类运行的耗时算法

每个进程的最大等待时间为2秒，这也是步骤2的最高等待时间，即使您增加了并行任务的数量（只需尝试更改以下代码的变量totalTasks即可对其进行测试）

以下是主要课程：

package com.example;

import java.util.ArrayList;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

public class Main
{
    public static void main(String[] args) throws InterruptedException, ExecutionException
    {
        int totalTasks = 100;

        ExecutorService newFixedThreadPool = Executors.newFixedThreadPool(totalTasks);

        System.out.println("Step 1 - Starting parallel tasks");

        ArrayList<Future<Integer>> tasks = new ArrayList<Future<Integer>>();
        for (int i = 0; i < totalTasks; i++) {
            tasks.add(newFixedThreadPool.submit(new Process(i)));
        }

        long ts = System.currentTimeMillis();
        System.out.println("Step 2 - Wait for processes to finish...");

        boolean tasksCompleted;
        do {
            tasksCompleted = true;

            for (Future<Integer> task : tasks) {
                if (!task.isDone()) {
                    tasksCompleted = false;
                    Thread.sleep(10);
                    break;
                }
            }

        } while (!tasksCompleted);

        System.out.println(String.format("Step 2 - End in '%.3f' seconds", (System.currentTimeMillis() - ts) / 1000.0));

        System.out.println("Step 3 - All processes finished to run, let's collect results...");

        Integer sum = 0;

        for (Future<Integer> task : tasks) {
            sum += task.get();
        }

        System.out.println(String.format("Total final sum is: %d", sum));
    }
}

下面是流程类：

package com.example;

import java.util.concurrent.Callable;

public class Process implements Callable<Integer>
{
    private Integer value;

    public Process(Integer value)
    {
        this.value = value;
    }

    public Integer call() throws Exception
    {
        Long sleepTime = (long)(Math.random() * 2000);

        System.out.println(String.format("Starting process with value %d, sleep time %d", this.value, sleepTime));

        Thread.sleep(sleepTime);

        System.out.println(String.format("Stopping process with value %d", this.value));

        return value * 2;
    }
}

希望这有帮助

Answer 2

下面的例子可能会对你有所帮助。我使用了fork/join框架来实现这一点

对于像您的示例这样的小数组大小，传统方法可能会更快，我怀疑fork/join方法会花费稍长的时间。但对于更大的规模或流程，fork/join框架是合适的。甚至Java8并行流也使用fork/join框架作为底层基础

public class ForkMultiplier extends RecursiveAction {
        int[] array;
        int threshold = 3;
        int start;
        int end;

        public ForkMultiplier(int[] array,int start, int end) {
            this.array = array;
            this.start = start;
            this.end = end;
        }

        protected void compute() {
            if (end - start < threshold) {
                computeDirectly();
            } else {
                int middle = (end + start) / 2;
                ForkMultiplier f1= new ForkMultiplier(array, start, middle);
                ForkMultiplier f2= new ForkMultiplier(array, middle, end);
                invokeAll(f1, f2);
            }
        }

        protected void computeDirectly() {
            for (int i = start; i < end; i++) {
                array[i] = array[i] * 2;
            }
        }
    }

你们班的学生会喜欢下面这个

 public static void main(String s[]){

        int arrlen = 10;
        int arr[] = new int[arrlen] ;


        for(int i=0;i<arrlen;i++){
            arr[i] = i;
        }

        ForkJoinPool pool = new ForkJoinPool();
        pool.invoke(new ForkMultiplier(arr, 0, arr.length));

        int sum =0;
        for(int i=0;i<arrlen;i++){
            sum += arr[i];
        }

        System.out.println(sum);

    }