高并发设计模式之ForkJoin模式

分而治之是一种思想,所谓分而治之就是把一个复杂的算法问题按一定的分解方法分为规模较小的若干部分,然后逐个解决,分别找出各部分的解,最后把各部分的解在整合成整个问题的解.ForkJoin模式就是分而治之思想的另一种应用.

ForkJoin模式的原理

ForkJoin模式先把一个大任务分解成许多独立的子任务,然后开启多个线程并行去处理这些子任务.执行的过程大致如下.

ForkJoin框架

JUC包提供了一套ForkJoin框架的实现,具体以ForkJoinPool线程池的形式提供,并且该线程池在java8的Lambda并行流框架中充当着底层框架的角色.包含如下组件.

1:ForkJoinPool 执行任务的线程池,继承了 AbstractExecutorService类.

2:ForkJoinWorkerThread 执行任务的工作线程.每个线程都维护这一个内部队列,用于存放内部任务.该类继承了Thread类.

3:ForkJoinTask: 用于ForkJoinPool的任务抽象类,实现了Future接口.

4: RecursiveAction 不返回结果的递归执行任务,是ForkJoinTask的子类,在子任务不带返回结果时使用.

ForkJoin框架的使用实战

假设需要计算0-100的累加求和,可以使用ForkJoin框架完成.首先需要设计一个可以递归执行的异步子任务类.

1:可递归执行的异步任务类AccumulateTask

参考代码如下.

public class AccumulateTask extends RecursiveTask<Integer> {private static final int THRESHOLD = 2;/*** 累加的起始编号.*/private int start;/*** 累加的结束编号.*/private int end;public AccumulateTask(int start, int end) {this.start = start;this.end = end;}@Overrideprotected Integer compute() {int sum = 0;//判断任务的规模.若规模小则可以直接计算.Boolean canCompute = end - start <= THRESHOLD;if (canCompute) {for (int i = start; i <= end; i++) {sum += i;}System.out.println("执行任务,计算" + start + "到" + end + "的和,结果是" + sum);} else {//任务过大需要进行切割.int middle = (start + end) / 2;//切割成两个任务.AccumulateTask lTask = new AccumulateTask(start, middle);AccumulateTask rTask = new AccumulateTask(middle + 1, end);//依次调用每个子任务的fork方法.lTask.fork();rTask.fork();//等待子任务完成.依次调用每个子任务的join方法合并并执行结果.Integer leftResult = lTask.join();Integer rightResult = rTask.join();//合并子任务的结果.sum = leftResult + rightResult;}return sum;}
}

自定义的异步任务子类AccumulateTask继承自RecursiveTask,每一次执行可以携带返回值.在AccumulateTask通过THRESHOLD常量设置子任务分解的阈值,并在它的compute()方法中会进行阈值的判断.逻辑如下.

1:若当前的计算规模大于THRESHOLD,则当前任需要进一步分解.相反则直接求和.

2:如果子任务可以直接执行,则进行求和操作,并返回结果.如果任务进行了分解,就需要等待所有的任务执行完毕,然后对各个分解结果进行求和.

ForkJoinPool调度AccumulateTask()

参考代码如下.

public class ForkJoinTest {public static void main(String[] args) throws ExecutionException, InterruptedException, TimeoutException {ForkJoinPool forkJoinPool = new ForkJoinPool();AccumulateTask task = new AccumulateTask(1, 100);ForkJoinTask<Integer> future = forkJoinPool.submit(task);Integer integer = future.get(1, TimeUnit.SECONDS);System.out.println("最终计算结果为" + integer);}
}

ForkJoinPool框架的核心API

ForkJoinPool框架的核心是ForkJoinPool线程池.该线程池使用一个无锁的栈来管理空闲线程,如果一个工作线程暂时取不到可用的任务,咋可能被挂起,而挂起的线程将被压入ForkJoinPool维护的栈中,等待新任务到来.

1:ForkJoinPool的构造器

public ForkJoinPool(int parallelism,//并行度 默认为cpu数,最小为1.ForkJoinWorkerThreadFactory factory,//线程创建工厂UncaughtExceptionHandler handler,//异常处理程序boolean asyncMode) {///是否为异步模式.this(checkParallelism(parallelism),checkFactory(factory),handler,asyncMode ? FIFO_QUEUE : LIFO_QUEUE,"ForkJoinPool-" + nextPoolId() + "-worker-");checkPermission();}

parallelism:并行级别

设定的级别决定框架内并行线程的数量.并行的每一个任务都会有一个线程进行处理,但这个属性并不是框架的最大线程数量,该属性也和ThreadPoolExecutor线程池中的核心线程数和最大线程数有区别.因为他们的工作方式不一样.Forkjoin框架中可存在的线程数量和parallelism参数并不是绝对的关联.

Factory:线程创建工厂

当ForkJoin框架创建一个新的线程,同样会用到线程工厂,只不过这个线程不在需要实现ThreadFactory接口,而是需要实现ForkJoinWorkerThreadFactory接口.后者是一个函数接口.只需要实现一个叫newThread()方法,在ForkJoin框架中有一个默认的实现:DefaultForkJoinWorkerThreadFactory.

handler:异常捕获处理程序

当执行任务中出现异常,并从任务中被抛出时,就会被handler捕获.

asyncMode:异步模式

该参数表示任务是否为异步模式,默认值为false.如果参数值为true,表示子任务的执行遵循FIFO(先进先出)顺序,并且子任务不能被合并,如果为false,表示子任务的执行顺序为LIFO(后进先出),并且子任务可以被合并.虽然表面意思为异步,仅仅指任务调度.

ForkJoinPool的common通用池

private static ForkJoinPool makeCommonPool() {final ForkJoinWorkerThreadFactory commonPoolForkJoinWorkerThreadFactory =new CommonPoolForkJoinWorkerThreadFactory();int parallelism = -1;ForkJoinWorkerThreadFactory factory = null;UncaughtExceptionHandler handler = null;try {  // ignore exceptions in accessing/parsing propertiesString pp = System.getProperty("java.util.concurrent.ForkJoinPool.common.parallelism");String fp = System.getProperty("java.util.concurrent.ForkJoinPool.common.threadFactory");String hp = System.getProperty("java.util.concurrent.ForkJoinPool.common.exceptionHandler");if (pp != null)parallelism = Integer.parseInt(pp);if (fp != null)factory = ((ForkJoinWorkerThreadFactory)ClassLoader.getSystemClassLoader().loadClass(fp).newInstance());if (hp != null)handler = ((UncaughtExceptionHandler)ClassLoader.getSystemClassLoader().loadClass(hp).newInstance());} catch (Exception ignore) {}if (factory == null) {if (System.getSecurityManager() == null)factory = commonPoolForkJoinWorkerThreadFactory;else // use security-managed defaultfactory = new InnocuousForkJoinWorkerThreadFactory();}if (parallelism < 0 && // default 1 less than #cores(parallelism = Runtime.getRuntime().availableProcessors() - 1) <= 0)parallelism = 1;if (parallelism > MAX_CAP)parallelism = MAX_CAP;return new ForkJoinPool(parallelism, factory, handler, LIFO_QUEUE,"ForkJoinPool.commonPool-worker-");}

使用common线程池的优点是可以通过指定系统属性的方式定义并行度线程工厂和异常处理类.

向ForkJoinPool线程池提交任务的方式

1:外部任务提交

提交外部任务有三种方式:方式一使用invoke方法,该方法提交任务后线程会等待,等到任务计算完毕并返回结果.方法二使用execute()方法提交一个任务来异步执行,无返回结果.方法三使用submit方法提交一个任务,会返回一个ForkJoinTask实例,适当的时候获取结果.

2:子任务提交.

由任务实例的fork方法完成,当任务被分解以后,内部会调用ForkJoinPool.WorkQueue.push方法直接把任务放到内部队列中等待被执行.