Java Concurrency 简明教程

Java Concurrency - Fork-Join framework

fork-join 框架允许在多个工作进程上中断某个任务，然后等待结果以合并它们。它很大程度上利用了多处理器的计算机容量。以下是 fork-join 框架中使用的核心概念和对象。

The fork-join framework allows to break a certain task on several workers and then wait for the result to combine them. It leverages multi-processor machine’s capacity to great extent. Following are the core concepts and objects used in fork-join framework.

Fork

Fork 是任务将自身拆分为多个可并发执行的更小的独立子任务的过程。

Fork is a process in which a task splits itself into smaller and independent sub-tasks which can be executed concurrently.

Syntax

Sum left  = new Sum(array, low, mid);
left.fork();

此处 Sum 为 RecursiveTask 的子类，而 left.fork() 将任务拆分为子任务。

Here Sum is a subclass of RecursiveTask and left.fork() spilts the task into sub-tasks.

Join

Join 是任务在子任务完成执行后，合并所有子任务结果的过程，否则它会保持等待状态。

Join is a process in which a task join all the results of sub-tasks once the subtasks have finished executing, otherwise it keeps waiting.

Syntax

left.join();

此处 left 为 Sum 类的对象。

Here left is an object of Sum class.

ForkJoinPool

它是一个特殊线程池，旨在处理 fork 和 join 任务拆分。

it is a special thread pool designed to work with fork-and-join task splitting.

Syntax

ForkJoinPool forkJoinPool = new ForkJoinPool(4);

此助手程序使用并发级别的 4 CPU 创建了新的 ForkJoinPool。

Here a new ForkJoinPool with a parallelism level of 4 CPUs.

RecursiveAction

RecursiveAction 表示不返回任何值的某个任务。

RecursiveAction represents a task which does not return any value.

Syntax

class Writer extends RecursiveAction {
   @Override
   protected void compute() { }
}

RecursiveTask

RecursiveTask 表示返回某个值的某个任务。

RecursiveTask represents a task which returns a value.

Syntax

class Sum extends RecursiveTask<Long> {
   @Override
   protected Long compute() { return null; }
}

Example

以下 TestThread 程序演示了线程基础环境中 Fork-Join 框架的用法。

The following TestThread program shows usage of Fork-Join framework in thread based environment.

import java.util.concurrent.ExecutionException;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.RecursiveTask;

public class TestThread {

   public static void main(final String[] arguments) throws InterruptedException,
      ExecutionException {

      int nThreads = Runtime.getRuntime().availableProcessors();
      System.out.println(nThreads);

      int[] numbers = new int[1000];

      for(int i = 0; i < numbers.length; i++) {
         numbers[i] = i;
      }

      ForkJoinPool forkJoinPool = new ForkJoinPool(nThreads);
      Long result = forkJoinPool.invoke(new Sum(numbers,0,numbers.length));
      System.out.println(result);
   }

   static class Sum extends RecursiveTask<Long> {
      int low;
      int high;
      int[] array;

      Sum(int[] array, int low, int high) {
         this.array = array;
         this.low   = low;
         this.high  = high;
      }

      protected Long compute() {

         if(high - low <= 10) {
            long sum = 0;

            for(int i = low; i < high; ++i)
               sum += array[i];
               return sum;
         } else {
            int mid = low + (high - low) / 2;
            Sum left  = new Sum(array, low, mid);
            Sum right = new Sum(array, mid, high);
            left.fork();
            long rightResult = right.compute();
            long leftResult  = left.join();
            return leftResult + rightResult;
         }
      }
   }
}

这将产生以下结果。

This will produce the following result.

Output

32
499500