14-案例实现

案例实现

按照如下步骤来实现本案例。

1．完成一个继承 RecursiveAction 的 Task 类：

public class Task extends RecursiveAction{

2．在该类中添加一个存放增加元素的 int 型数组：

private final int array[];

3．声明两个 int 类型的属性 start 和 end ，它们记录当前任务需要处理的数据块的起点和终点：

private final int start;
private final int end;

4．在类的构造函数中完成上述属性的初始化工作：

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

5．用任务的主逻辑实现 compute() 方法。如果当前任务需要处理的元素超过100个，那么首先把该任务分成两部分，并分别创建两个子任务完成这两部分工作，再用 fork() 方法开始执行任务并用 join() 方法等待它的终止：

protected void compute() {
  if (end-start>100) {
    int mid=(start+end)/2;
    Task task1=new Task(array,start,mid);
    Task task2=new Task(array,mid,end);
    task1.fork();
    task2.fork();
    task1.join();
    task2.join();

6．如果任务必须处理100个或者更少的元素，则在每次操作后把数组里的每个元素自增1并休眠5ms：

  } else {
      for (int i=start; i<end; i++) {
        array[i]++;
        try {
          Thread.sleep(5);
        } catch (InterruptedException e) {
          e.printStackTrace();
        }
      }
    }
  }
}

7．实现本案例的主类，新建 Main 类和 main() 方法：

public class Main {
  public static void main(String[] args) throws Exception {

8．创建 ForkJoinPool 类型的 pool 对象：

ForkJoinPool pool=new ForkJoinPool();

9．创建一个可以容纳10000个元素的整数数组 array ：

int array[]=new int[10000];

10．创建一个新的 Task 对象来处理整个数组：

Task task1=new Task(array,0,array.length);

11．调用 execute() 方法，把任务提交到线程池中：

pool.execute(task1);

12．如果任务没有完成执行，则通过 showLog() 方法写入 ForkJoinPool 对象的状态信息并让线程休眠1s：

while (!task1.isDone()) {
  showLog(pool);
  TimeUnit.SECONDS.sleep(1);
}

13．通过 shutdown() 方法关闭线程池：

pool.shutdown();

14．用 awaitTermination() 等待线程池中的任务全部执行完成：

pool.awaitTermination(1, TimeUnit.DAYS);

15．用 showLog() 方法获取 ForkJoinPool 类的状态信息，并在程序执行结束时打印结束信息：

showLog(pool);
System.out.printf("Main: End of the program.\n");

16．实现 showLog() 方法。该方法接收一个 ForkJoinPool 类型的参数，并输出正在执行的线程和任务的状态信息：

private static void showLog(ForkJoinPool pool) {
  System.out.printf("**********************\n");
  System.out.printf("Main: Fork/Join Pool log\n");
  System.out.printf("Main: Fork/Join Pool: Parallelism: %d\n",
                    pool.getParallelism());
  System.out.printf("Main: Fork/Join Pool: Pool Size: %d\n",
                    pool.getPoolSize());
  System.out.printf("Main: Fork/Join Pool: Active Thread Count:
                    %d\n", pool.getActiveThreadCount());
  System.out.printf("Main: Fork/Join Pool: Running Thread Count:
                    %d\n", pool.getRunningThreadCount());
  System.out.printf("Main: Fork/Join Pool: Queued Submission:
                    %d\n", pool.getQueuedSubmissionCount());
  System.out.printf("Main: Fork/Join Pool: Queued Tasks: %d\n",
                    pool.getQueuedTaskCount());
  System.out.printf("Main: Fork/Join Pool: Queued Submissions:
                    %s\n", pool.hasQueuedSubmissions());
  System.out.printf("Main: Fork/Join Pool: Steal Count: %d\n",
                    pool.getStealCount());
  System.out.printf("Main: Fork/Join Pool: Terminated : %s\n",
                    pool.isTerminated());
  System.out.printf("**********************\n");
}