3

我正在玩 fork/join 并想到以下示例：

App1：2个for循环将一些随机数生成到anArrayList中并将其传递给a fork MyTask（Fork）：遍历ArrayLists并添加所有数字，然后返回值

import java.util.ArrayList;
import java.util.concurrent.ForkJoinPool;

public class App1 {

    ArrayList list = new ArrayList();

    static final ForkJoinPool mainPool = new ForkJoinPool();

    public App1() {
        for (int i = 0; i < 10; i++) {
            list.clear();
            for (int j = 1000; j <= 100000; j++) {
                int random = 1 + (int)(Math.random() * ((100 - 1) + 1));
                list.add(random);
            }
            mainPool.invoke(new MyTask(list));
        }
        // At the end showing all results
        // System.out.println (result1 + result2 + result3...);
    }

    public static void main(String[] args) {
        App1 app = new App1();
    }
}


import java.util.ArrayList;
import java.util.concurrent.RecursiveTask;

public class MyTask extends RecursiveTask<Integer> {

    ArrayList list = new ArrayList();
    int result;

    public MyTask(ArrayList list) {
        this.list = list;
    }

    @Override
    protected Integer compute() {
        for(int i=0; i<=list.size(); i++){
            result += (int)list.get(i); // adding up all numbers
        }
        return result;
    }
}

我不确定我是否走在正确的轨道上。我也不知道如何从分叉中收集所有结果。
有人可以看看我的代码吗？

4

4 回答 4

1

虽然您的代码看起来不错，但您以错误的方式使用 ForkJoinPool。这是一个用于任务的工具，可以拆分为独立的子任务，并且可以通过多线程来提高速度。

您的任务可能不够大，无法真正从多线程中受益，但抛开这一点，因为它是一个学习练习，您仍然需要将主要任务拆分为子任务，但您所做的只是计算整个数组一次。

您可以在代码中执行的操作：

  • 使用更适合的不同多线程模式。

  • 分叉任务并交出数组内的开始和结束位置，然后在这些子任务完成后总结结果。

由于您可能对后一种感兴趣，因此这里有一个有关如何执行此操作的示例：

public class MyTask extends RecursiveTask<Integer> {

  final ArrayList<Integer> list;
  final int start, end;

  public MyTask(ArrayList<Integer> list, int start, int end) {
    this.list = list;
    this.start = start;
    this.end = end;
  }

  @Override
  protected Integer compute() {
    if (end - start > 10) { // is this task big enough to justify more threading?
      final int half = (end + start) / 2;
      final MyTask firstHalf = new MyTask(list, start, half);
      final MyTask secondHalf = new MyTask(list, half+1, end);
      invokeAll(firstHalf, secondHalf);
      return firstHalf.get() + secondHalf.get();
    } else {
      int result = 0;
      for(int i=start; i<=end; i++){
        result += list.get(i); 
      }
      return result;
    }
  }
}
于 2013-11-12T10:10:15.840 回答
0

您使用的RecursiveTask方式不正确。你必须调用任务递归，看看下面的一个可能的解决方案。

所以最后所有的工作都被分割成< THRESHOLD递归的部分。

更新：

你可能想看看Java Concurrent Animated，只需下载 jar 并执行它，它有一个很好的视觉解释事情是如何工作的。

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.RecursiveTask;

public class RecursiveListSum {

  private static class RecursiveSum extends RecursiveTask<Long> {
    private static final long serialVersionUID = 1L;
    private static final int THRESHOLD = 1000;
    private final List<Integer> list;
    private final int begin;
    private final int end;

    public RecursiveSum(List<Integer> list, int begin, int end) {
      super();
      this.list = list;
      this.begin = begin;
      this.end = end;
    }

    @Override
    protected Long compute() {
      final int size = end - begin;
      // if the work to be done is below some threshold, just compute directly.
      if (size < THRESHOLD) {
        long sum = 0;
        for (int i = begin; i < end; i++)
          sum += list.get(i);
        return sum;
      } else {
        // split the work to other tasks - recursive (that's why it is called recursive task!)
        final int middle = begin + ((end - begin) / 2);
        RecursiveSum sum1 = new RecursiveSum(list, begin, middle);
        // invoke the first portion -> will be invoked in thread pool
        sum1.fork();
        RecursiveSum sum2 = new RecursiveSum(list, middle, end);
        // now do a blocking! compute on the second task and wait for the result of the first task.
        return sum2.compute() + sum1.join();
      }
    }
  }

  public static void main(String[] args) {
    // First fill the list
    List<Integer> list = new ArrayList<>();
    long expectedSum = 0;
    for (int i = 0; i < 10000; i++) {
      int random = 1 + (int) (Math.random() * ((100 - 1) + 1));
      list.add(random);
      expectedSum += random;
    }
    System.out.println("expected sum: " + expectedSum);

    // now let the RecursiveTask calc the sum again.
    final ForkJoinPool forkJoinPool = new ForkJoinPool(Runtime.getRuntime().availableProcessors());
    final RecursiveSum recursiveSum = new RecursiveSum(list, 0, list.size());
    long recSum = forkJoinPool.invoke(recursiveSum);
    System.out.println("recursive-sum: " + recSum);
  }

}
于 2013-11-12T10:16:13.490 回答
0

只需使用以下结果invoke：

Integer result = mainPool.invoke(new MyTask(list));
System.out.println(i + "\t" + result);
于 2013-11-12T10:09:26.903 回答
0

您需要扩展RecursiveAction或RecursiveTask根据您的需要是否需要返回某些东西。

在这两个类中，您需要覆盖以下函数：

protected abstract void compute();//in RecursiveAction
protected abstract V compute(); //in RecursiveTask

下面是修改后的快速排序。请注意这一点并尝试实现你自己：

public class ForkJoinQuicksortTask extends RecursiveAction {
    static final int SEQUENTIAL_THRESHOLD = 10000;

    private final int[] a;
    private final int left;
    private final int right;

    public ForkJoinQuicksortTask(int[] a) {
        this(a, 0, a.length - 1);
    }

    private ForkJoinQuicksortTask(int[] a, int left, int right) {
        this.a = a;
        this.left = left;
        this.right = right;
    }

    @Override
    protected void compute() {
        if (serialThresholdMet()) {
            Arrays.sort(a, left, right + 1);
        } else {
            int pivotIndex = partition(a, left, right);
            ForkJoinQuicksortTask  t1 = new ForkJoinQuicksortTask(a, left, pivotIndex-1);
            ForkJoinQuicksortTask t2 = new ForkJoinQuicksortTask(a, pivotIndex + 1, right);
            t1.fork();
            t2.compute();
            t1.join();
        }
    }
    int partition(int[] a, int p, int r){
        int i=p-1;
        int x=a[r];
        for(int j=p;j<r;j++){
            if(a[j]<x){
                i++;
                swap(a, i, j);
            }
        }
        i++;
        swap(a, i, r);
        return i;
    }

    void swap(int[] a, int p, int r){
        int t=a[p];
        a[p]=a[r];
        a[r]=t;
    }

    private boolean serialThresholdMet() {
        return right - left < SEQUENTIAL_THRESHOLD;
    }
    public static void main(String[] args){
        ForkJoinPool fjPool = new ForkJoinPool();
        int[] a=new int[3333344];
        for(int i=0;i<a.length;i++){
            int k=(int)(Math.random()*22222);
            a[i]=k;
        }
        ForkJoinQuicksortTask forkJoinQuicksortTask=new ForkJoinQuicksortTask(a, 0, a.length-1);
        long start=System.nanoTime();
        fjPool.invoke(forkJoinQuicksortTask);
        System.out.println("Time: "+ (System.nanoTime()-start));
    }
}
于 2013-11-12T10:09:43.530 回答