多线程 ForkJoinPool

亦凉 2022-05-15 05:37 249阅读 0赞

背景：ForkJoinPool的优势在于，可以充分利用多cpu，多核cpu的优势，把一个任务拆分成多个“小任务”，把多个“小任务”放到多个处理器核心上并行执行；当多个“小任务”执行完成之后，再将这些执行结果合并起来即可。这种思想值得学习。

# 介绍 #

Java7 提供了ForkJoinPool来支持将一个任务拆分成多个“小任务”并行计算，再把多个“小任务”的结果合并成总的计算结果。

ForkJoinPool是ExecutorService的实现类，因此是一种特殊的线程池。

使用方法：创建了ForkJoinPool实例之后，就可以调用ForkJoinPool的submit(ForkJoinTask<T> task) 或invoke(ForkJoinTask<T> task)方法来执行指定任务了。

其中ForkJoinTask代表一个可以并行、合并的任务。ForkJoinTask是一个抽象类，它还有两个抽象子类：RecusiveAction和RecusiveTask。其中RecusiveTask代表有返回值的任务，而RecusiveAction代表没有返回值的任务。

下面的UML类图显示了ForkJoinPool、ForkJoinTask之间的关系：

# ![70][]案列一：通过多线程分多个小任务进行打印数据  无返回值的 #

/**
 * 
 */
 package forkJoinPool;
 
 import java.util.concurrent.ForkJoinPool;
 import java.util.concurrent.RecursiveAction;
 import java.util.concurrent.TimeUnit;
 
 /**
 * @author liuchaojun
 * @date 2018-9-10 下午02:57:29
 * 
 * RecursiveAction 无返回值的
 * 
 */
 public class PrintTask extends RecursiveAction {
 
 	private static final long serialVersionUID = 1L;
 	private static final int INDEX = 50;
 	private int start;
 	private int end;
 
 	/**
 	 * 
 	 */
 	public PrintTask(int start, int end) {
 		// TODO Auto-generated constructor stub
 		this.start = start;
 		this.end = end;
 	}
 
 	/*
 	 * (non-Javadoc)
 	 * 
 	 * @see java.util.concurrent.RecursiveAction#compute()
 	 */
 	@Override
 	protected void compute() {
 		if (end - start < INDEX) {
 			for (int i = start; i < end; i++) {
 				System.out.println(Thread.currentThread().getName() + "----"
 						+ i);
 			}
 		} else {
 			int middle = (end + start) / 2;
 			PrintTask taskLeft = new PrintTask(start, middle);
 			PrintTask taskRight = new PrintTask(middle, end);
 			//taskLeft.invoke();执行给定的任务，在完成后返回其结果。
 			//并行执行两个“小任务”
 /*			taskLeft.fork();
 			taskRight.fork();*/
 			invokeAll(taskLeft, taskRight);//执行给定的任务
 		}
 	}
 
 	public static void main(String[] args) throws InterruptedException {
 		PrintTask task = new PrintTask(0, 300);
 		ForkJoinPool pool = new ForkJoinPool();
 		pool.submit(task);
 		pool.awaitTermination(2, TimeUnit.SECONDS);//阻塞2秒
 		pool.shutdown();
 	}
 }

## **运行结果** ##

![70 1][]

![70 2][]

### 本机电脑cpu4核，通过上面观察线程名称，可以看出4个cpu都在进行 ###

# 案列二：通过多线程分多个小任务进行数据累加  返回结果集 #

package forkJoinPool;
 import java.util.Random;
 import java.util.concurrent.ForkJoinPool;
 import java.util.concurrent.Future;
 import java.util.concurrent.RecursiveTask;
 import java.util.concurrent.TimeUnit;
 
 /**
 * @author liuchaojun
 * @date 2018-9-10 下午03:50:24
 */
 public class PrintTask2 {
 	public static void main(String[] args) throws Exception {
 		int[] arr = new int[200];
 		Random r = new Random();
 		int tempSum = 0;// 普通总数
 		for (int i = 0; i < arr.length; i++) {
 			tempSum += (arr[i] = r.nextInt(10));
 		}
 		System.out.println("普通总数结果为:" + tempSum);
 
 		ForkJoinPool pool = new ForkJoinPool();
 		MyTask task = new MyTask(0, arr.length, arr);
 		Future<Integer> sum = pool.submit(task);
 		System.out.println("多线程的执行结果：" + sum.get());// get 如果需要，等待计算完成，然后检索其结果。
 		pool.awaitTermination(2, TimeUnit.SECONDS);
 		pool.shutdown(); // 关闭线程池
 	}
 }
 
 class MyTask extends RecursiveTask<Integer> {
 	private static final long serialVersionUID = 1L;
 	private static final int INDEX = 50;// 每个小任务执行50个
 	private int start;
 	private int end;
 	private int arr[];
 
 	/**
 	 * @param start
 	 * @param end
 	 * @param arr
 	 */
 	public MyTask(int start, int end, int[] arr) {
 
 		this.start = start;
 		this.end = end;
 		this.arr = arr;
 	}
 
 	/*
 	 * (non-Javadoc)
 	 * 
 	 * @see java.util.concurrent.RecursiveTask#compute()
 	 */
 	@Override
 	protected Integer compute() {
 		int sum = 0;
 		if (end - start < INDEX) {
 			for (int i = start; i < end; i++) {
 				sum += arr[i];
 			}
 			return sum;
 		} else {
 			int middle = (end + start) / 2;
 			MyTask taskLeft2 = new MyTask(start, middle, arr);
 			MyTask taskRight2 = new MyTask(middle, end, arr);
 			/*invokeAll(taskLeft2, taskRight2);*/
 			taskLeft2.fork();
 			taskRight2.fork();
 			int leftValue = taskLeft2.join();// 当计算完成时返回计算结果。
 			int rightValue = taskRight2.join();
 			return leftValue + rightValue;
 		}
 	}
 }

运行结果：

![70 3][]

# 提供一些API 的方法，可以参考下 #

## **ForkJoinPool类** ##

* * ### 构造方法摘要 ###
 
 <table> Constructors&nbsp; 
 <tbody> 
 <tr> 
 <th>构造方法与描述</th> 
 </tr> 
 <tr> 
 <td><code><a rel="nofollow">ForkJoinPool</a>()</code> Creates a <code>ForkJoinPool</code> with parallelism equal to <a rel="nofollow"><code>Runtime.availableProcessors()</code></a>, using the <a rel="nofollow">default thread factory</a>, no UncaughtExceptionHandler, and non-async LIFO processing mode. </td> 
 </tr> 
 <tr> 
 <td><code><a rel="nofollow">ForkJoinPool</a>(int&nbsp;parallelism)</code> Creates a <code>ForkJoinPool</code> with the indicated parallelism level, the <a rel="nofollow">default thread factory</a>, no UncaughtExceptionHandler, and non-async LIFO processing mode. </td> 
 </tr> 
 <tr> 
 <td><code><a rel="nofollow">ForkJoinPool</a>(int&nbsp;parallelism, <a rel="nofollow">ForkJoinPool.ForkJoinWorkerThreadFactory</a>&nbsp;factory, <a rel="nofollow">Thread.UncaughtExceptionHandler</a>&nbsp;handler, boolean&nbsp;asyncMode)</code> Creates a <code>ForkJoinPool</code> with the given parameters. </td> 
 </tr> 
 </tbody> 
 </table>
 
 * ### 方法摘要 ###
 
 <table> Methods&nbsp; 
 <tbody> 
 <tr> 
 <th>修饰符与类型</th> 
 <th>方法与描述</th> 
 </tr> 
 <tr> 
 <td><code>boolean</code></td> 
 <td><code><a rel="nofollow">awaitTermination</a>(long&nbsp;timeout, <a rel="nofollow">TimeUnit</a>&nbsp;unit)</code> Blocks until all tasks have completed execution after a shutdown request, or the timeout occurs, or the current thread is interrupted, whichever happens first. </td> 
 </tr> 
 <tr> 
 <td><code>protected int</code></td> 
 <td><code><a rel="nofollow">drainTasksTo</a>(<a rel="nofollow">Collection</a>&lt;? super <a rel="nofollow">ForkJoinTask</a>&lt;?&gt;&gt;&nbsp;c)</code> Removes all available unexecuted submitted and forked tasks from scheduling queues and adds them to the given collection, without altering their execution status. </td> 
 </tr> 
 <tr> 
 <td><code>void</code></td> 
 <td><code><a rel="nofollow">execute</a>(<a rel="nofollow">ForkJoinTask</a>&lt;?&gt;&nbsp;task)</code> Arranges for (asynchronous) execution of the given task. </td> 
 </tr> 
 <tr> 
 <td><code>void</code></td> 
 <td><code><a rel="nofollow">execute</a>(<a rel="nofollow">Runnable</a>&nbsp;task)</code> Executes the given command at some time in the future. </td> 
 </tr> 
 <tr> 
 <td><code>int</code></td> 
 <td><code><a rel="nofollow">getActiveThreadCount</a>()</code> Returns an estimate of the number of threads that are currently stealing or executing tasks. </td> 
 </tr> 
 <tr> 
 <td><code>boolean</code></td> 
 <td><code><a rel="nofollow">getAsyncMode</a>()</code> Returns <code>true</code> if this pool uses local first-in-first-out scheduling mode for forked tasks that are never joined. </td> 
 </tr> 
 <tr> 
 <td><code><a rel="nofollow">ForkJoinPool.ForkJoinWorkerThreadFactory</a></code></td> 
 <td><code><a rel="nofollow">getFactory</a>()</code> Returns the factory used for constructing new workers. </td> 
 </tr> 
 <tr> 
 <td><code>int</code></td> 
 <td><code><a rel="nofollow">getParallelism</a>()</code> Returns the targeted parallelism level of this pool. </td> 
 </tr> 
 <tr> 
 <td><code>int</code></td> 
 <td><code><a rel="nofollow">getPoolSize</a>()</code> Returns the number of worker threads that have started but not yet terminated. </td> 
 </tr> 
 <tr> 
 <td><code>int</code></td> 
 <td><code><a rel="nofollow">getQueuedSubmissionCount</a>()</code> Returns an estimate of the number of tasks submitted to this pool that have not yet begun executing. </td> 
 </tr> 
 <tr> 
 <td><code>long</code></td> 
 <td><code><a rel="nofollow">getQueuedTaskCount</a>()</code> Returns an estimate of the total number of tasks currently held in queues by worker threads (but not including tasks submitted to the pool that have not begun executing). </td> 
 </tr> 
 <tr> 
 <td><code>int</code></td> 
 <td><code><a rel="nofollow">getRunningThreadCount</a>()</code> Returns an estimate of the number of worker threads that are not blocked waiting to join tasks or for other managed synchronization. </td> 
 </tr> 
 <tr> 
 <td><code>long</code></td> 
 <td><code><a rel="nofollow">getStealCount</a>()</code> Returns an estimate of the total number of tasks stolen from one thread's work queue by another. </td> 
 </tr> 
 <tr> 
 <td><code><a rel="nofollow">Thread.UncaughtExceptionHandler</a></code></td> 
 <td><code><a rel="nofollow">getUncaughtExceptionHandler</a>()</code> Returns the handler for internal worker threads that terminate due to unrecoverable errors encountered while executing tasks. </td> 
 </tr> 
 <tr> 
 <td><code>boolean</code></td> 
 <td><code><a rel="nofollow">hasQueuedSubmissions</a>()</code> Returns <code>true</code> if there are any tasks submitted to this pool that have not yet begun executing. </td> 
 </tr> 
 <tr> 
 <td><code>&lt;T&gt;&nbsp;T</code></td> 
 <td><code><a rel="nofollow">invoke</a>(<a rel="nofollow">ForkJoinTask</a>&lt;T&gt;&nbsp;task)</code> Performs the given task, returning its result upon completion. </td> 
 </tr> 
 <tr> 
 <td><code>&lt;T&gt;&nbsp;<a rel="nofollow">List</a>&lt;<a rel="nofollow">Future</a>&lt;T&gt;&gt;</code></td> 
 <td><code><a rel="nofollow">invokeAll</a>(<a rel="nofollow">Collection</a>&lt;? extends <a rel="nofollow">Callable</a>&lt;T&gt;&gt;&nbsp;tasks)</code> Executes the given tasks, returning a list of Futures holding their status and results when all complete. </td> 
 </tr> 
 <tr> 
 <td><code>boolean</code></td> 
 <td><code><a rel="nofollow">isQuiescent</a>()</code> Returns <code>true</code> if all worker threads are currently idle. </td> 
 </tr> 
 <tr> 
 <td><code>boolean</code></td> 
 <td><code><a rel="nofollow">isShutdown</a>()</code> Returns <code>true</code> if this pool has been shut down. </td> 
 </tr> 
 <tr> 
 <td><code>boolean</code></td> 
 <td><code><a rel="nofollow">isTerminated</a>()</code> Returns <code>true</code> if all tasks have completed following shut down. </td> 
 </tr> 
 <tr> 
 <td><code>boolean</code></td> 
 <td><code><a rel="nofollow">isTerminating</a>()</code> Returns <code>true</code> if the process of termination has commenced but not yet completed. </td> 
 </tr> 
 <tr> 
 <td><code>static void</code></td> 
 <td><code><a rel="nofollow">managedBlock</a>(<a rel="nofollow">ForkJoinPool.ManagedBlocker</a>&nbsp;blocker)</code> Blocks in accord with the given blocker. </td> 
 </tr> 
 <tr> 
 <td><code>protected &lt;T&gt;&nbsp;<a rel="nofollow">RunnableFuture</a>&lt;T&gt;</code></td> 
 <td><code><a rel="nofollow">newTaskFor</a>(<a rel="nofollow">Callable</a>&lt;T&gt;&nbsp;callable)</code> Returns a RunnableFuture for the given callable task. </td> 
 </tr> 
 <tr> 
 <td><code>protected &lt;T&gt;&nbsp;<a rel="nofollow">RunnableFuture</a>&lt;T&gt;</code></td> 
 <td><code><a rel="nofollow">newTaskFor</a>(<a rel="nofollow">Runnable</a>&nbsp;runnable, T&nbsp;value)</code> Returns a RunnableFuture for the given runnable and default value. </td> 
 </tr> 
 <tr> 
 <td><code>protected <a rel="nofollow">ForkJoinTask</a>&lt;?&gt;</code></td> 
 <td><code><a rel="nofollow">pollSubmission</a>()</code> Removes and returns the next unexecuted submission if one is available. </td> 
 </tr> 
 <tr> 
 <td><code>void</code></td> 
 <td><code><a rel="nofollow">shutdown</a>()</code> Initiates an orderly shutdown in which previously submitted tasks are executed, but no new tasks will be accepted. </td> 
 </tr> 
 <tr> 
 <td><code><a rel="nofollow">List</a>&lt;<a rel="nofollow">Runnable</a>&gt;</code></td> 
 <td><code><a rel="nofollow">shutdownNow</a>()</code> Attempts to cancel and/or stop all tasks, and reject all subsequently submitted tasks. </td> 
 </tr> 
 <tr> 
 <td><code>&lt;T&gt;&nbsp;<a rel="nofollow">ForkJoinTask</a>&lt;T&gt;</code></td> 
 <td><code><a rel="nofollow">submit</a>(<a rel="nofollow">Callable</a>&lt;T&gt;&nbsp;task)</code> Submits a value-returning task for execution and returns a Future representing the pending results of the task. </td> 
 </tr> 
 <tr> 
 <td><code>&lt;T&gt;&nbsp;<a rel="nofollow">ForkJoinTask</a>&lt;T&gt;</code></td> 
 <td><code><a rel="nofollow">submit</a>(<a rel="nofollow">ForkJoinTask</a>&lt;T&gt;&nbsp;task)</code> Submits a ForkJoinTask for execution. </td> 
 </tr> 
 <tr> 
 <td><code><a rel="nofollow">ForkJoinTask</a>&lt;?&gt;</code></td> 
 <td><code><a rel="nofollow">submit</a>(<a rel="nofollow">Runnable</a>&nbsp;task)</code> Submits a Runnable task for execution and returns a Future representing that task. </td> 
 </tr> 
 <tr> 
 <td><code>&lt;T&gt;&nbsp;<a rel="nofollow">ForkJoinTask</a>&lt;T&gt;</code></td> 
 <td><code><a rel="nofollow">submit</a>(<a rel="nofollow">Runnable</a>&nbsp;task, T&nbsp;result)</code> Submits a Runnable task for execution and returns a Future representing that task. </td> 
 </tr> 
 <tr> 
 <td><code><a rel="nofollow">String</a></code></td> 
 <td><code><a rel="nofollow">toString</a>()</code> Returns a string identifying this pool, as well as its state, including indications of run state, parallelism level, and worker and task counts. &nbsp; </td> 
 </tr> 
 </tbody> 
 </table>
 * ## ExecutorService**接口** ##
 * * ### 方法摘要 ###
 
 <table> Methods&nbsp; 
 <tbody> 
 <tr> 
 <th>修饰符与类型</th> 
 <th>方法与描述</th> 
 </tr> 
 <tr> 
 <td><code>boolean</code></td> 
 <td><code><a rel="nofollow">awaitTermination</a>(long&nbsp;timeout, <a rel="nofollow">TimeUnit</a>&nbsp;unit)</code> Blocks until all tasks have completed execution after a shutdown request, or the timeout occurs, or the current thread is interrupted, whichever happens first. </td> 
 </tr> 
 <tr> 
 <td><code>&lt;T&gt;&nbsp;<a rel="nofollow">List</a>&lt;<a rel="nofollow">Future</a>&lt;T&gt;&gt;</code></td> 
 <td><code><a rel="nofollow">invokeAll</a>(<a rel="nofollow">Collection</a>&lt;? extends <a rel="nofollow">Callable</a>&lt;T&gt;&gt;&nbsp;tasks)</code> Executes the given tasks, returning a list of Futures holding their status and results when all complete. </td> 
 </tr> 
 <tr> 
 <td><code>&lt;T&gt;&nbsp;<a rel="nofollow">List</a>&lt;<a rel="nofollow">Future</a>&lt;T&gt;&gt;</code></td> 
 <td><code><a rel="nofollow">invokeAll</a>(<a rel="nofollow">Collection</a>&lt;? extends <a rel="nofollow">Callable</a>&lt;T&gt;&gt;&nbsp;tasks, long&nbsp;timeout, <a rel="nofollow">TimeUnit</a>&nbsp;unit)</code> Executes the given tasks, returning a list of Futures holding their status and results when all complete or the timeout expires, whichever happens first. </td> 
 </tr> 
 <tr> 
 <td><code>&lt;T&gt;&nbsp;T</code></td> 
 <td><code><a rel="nofollow">invokeAny</a>(<a rel="nofollow">Collection</a>&lt;? extends <a rel="nofollow">Callable</a>&lt;T&gt;&gt;&nbsp;tasks)</code> Executes the given tasks, returning the result of one that has completed successfully (i.e., without throwing an exception), if any do. </td> 
 </tr> 
 <tr> 
 <td><code>&lt;T&gt;&nbsp;T</code></td> 
 <td><code><a rel="nofollow">invokeAny</a>(<a rel="nofollow">Collection</a>&lt;? extends <a rel="nofollow">Callable</a>&lt;T&gt;&gt;&nbsp;tasks, long&nbsp;timeout, <a rel="nofollow">TimeUnit</a>&nbsp;unit)</code> Executes the given tasks, returning the result of one that has completed successfully (i.e., without throwing an exception), if any do before the given timeout elapses. </td> 
 </tr> 
 <tr> 
 <td><code>boolean</code></td> 
 <td><code><a rel="nofollow">isShutdown</a>()</code> Returns true if this executor has been shut down. </td> 
 </tr> 
 <tr> 
 <td><code>boolean</code></td> 
 <td><code><a rel="nofollow">isTerminated</a>()</code> Returns true if all tasks have completed following shut down. </td> 
 </tr> 
 <tr> 
 <td><code>void</code></td> 
 <td><code><a rel="nofollow">shutdown</a>()</code> Initiates an orderly shutdown in which previously submitted tasks are executed, but no new tasks will be accepted. </td> 
 </tr> 
 <tr> 
 <td><code><a rel="nofollow">List</a>&lt;<a rel="nofollow">Runnable</a>&gt;</code></td> 
 <td><code><a rel="nofollow">shutdownNow</a>()</code> Attempts to stop all actively executing tasks, halts the processing of waiting tasks, and returns a list of the tasks that were awaiting execution. </td> 
 </tr> 
 <tr> 
 <td><code>&lt;T&gt;&nbsp;<a rel="nofollow">Future</a>&lt;T&gt;</code></td> 
 <td><code><a rel="nofollow">submit</a>(<a rel="nofollow">Callable</a>&lt;T&gt;&nbsp;task)</code> Submits a value-returning task for execution and returns a Future representing the pending results of the task. </td> 
 </tr> 
 <tr> 
 <td><code><a rel="nofollow">Future</a>&lt;?&gt;</code></td> 
 <td><code><a rel="nofollow">submit</a>(<a rel="nofollow">Runnable</a>&nbsp;task)</code> Submits a Runnable task for execution and returns a Future representing that task. </td> 
 </tr> 
 <tr> 
 <td><code>&lt;T&gt;&nbsp;<a rel="nofollow">Future</a>&lt;T&gt;</code></td> 
 <td><code><a rel="nofollow">submit</a>(<a rel="nofollow">Runnable</a>&nbsp;task, T&nbsp;result)</code> Submits a Runnable task for execution and returns a Future representing that task. &nbsp; </td> 
 </tr> 
 </tbody> 
 </table>
 * # 总结 #

1，invokeAll(task)方法，主动执行其它的ForkJoinTask，并等待Task完成。（同步的）

2，fork方法，让一个task执行（异步的）

3，join方法，让一个task执行（同步的，它和fork不同点是同步或者异步的区别）

4，可以使用join来取得ForkJoinTask的返回值。由于RecursiveTask类实现了Future接口，所以也可以使用get()取得返回值。   
get()和join()有两个主要的区别：   
join()方法不能被中断。如果你中断调用join()方法的线程，这个方法将抛出InterruptedException异常。   
如果任务抛出任何未受检异常，get()方法将返回一个ExecutionException异常，而join()方法将返回一个RuntimeException异常。

5，ForkJoinTask在不显示使用ForkJoinPool.execute/invoke/submit()方法进行执行的情况下，也可以使用自己的fork/invoke方法进行执行。   
使用fork/invoke方法执行时，其实原理也是在ForkJoinPool里执行，只不过使用的是一个“在ForkJoinPool内部生成的静态的”ForkJoinPool。

6，ForkJoinTask有两个子类，RecursiveAction和RecursiveTask。他们之间的区别是，RecursiveAction没有返回值，RecursiveTask有返回值。

7，看看ForkjoinTask的Complete方法的使用场景   
这个方法好要是用来使一个任务结束。这个方法被用在结束异步任务上，或者为那些能不正常结束的任务，提供一个选择。

8，Task的completeExceptionally方法是怎么回事。   
这个方法被用来，在异步的Task中产生一个exception，或者强制结束那些“不会结束”的任务   
这个方法是在Task想要“自己结束自己”时，可以被使用。而cancel方法，被设计成被其它TASK调用。   
当你在一个任务中抛出一个未检查异常时，它也影响到它的父任务（把它提交到ForkJoinPool类的任务）和父任务的父任务，以此类推。

9，可以使用ForkJoinPool.execute(异步，不返回结果)、invoke(同步，返回结果)、submit(异步，返回结果)方法，来执行ForkJoinTask。

10，ForkJoinPool有一个方法commonPool()，这个方法返回一个ForkJoinPool内部声明的静态ForkJoinPool实例。 在jdk1.8里面才有。文档上说，这个方法适用于大多数的应用。这个静态实例的初始线程数，为“CPU核数-1 ”（Runtime.getRuntime().availableProcessors() - 1）。   
ForkJoinTask自己启动时，使用的就是这个静态实例。

[70]: /images/20220515/0f1b58dd44794fdeacb7e2f8369160eb.png
[70 1]: /images/20220515/9f293433449448c0b416c6bb1ec6e626.png
[70 2]: /images/20220515/9cb9247916254a559e86bf9b0e424939.png
[70 3]: /images/20220515/0d361ce56c164a49b3a6bbf48b188aad.png