介绍:
Fork/Join并行是获得良好的并行性能的最简单高效的设计技术。他是分治(Divide-and-Conquer)算法的的并行实现,它的典型应用形式:
Result solve(Problem problem) { if (problem is small) directly solve problem else { split problem into independent parts fork new subtasks to solve each part join all subtasks compose result from subresults } }
fork操作启动一个新的并行fork/join子任务,join操作则导致当前线程等待直到子任务执行完成。fork/join算法,就像其它分治算法一样,几乎经常是不断的递归,重复划分子任务——直到它们的规模小到能够使用简单、简短的串行方法完成。
思想:
1、分而治之,将大任务分解为小任务,再合并最终结果
2、工作窃取调度策略
(1)每个worker线程利用它自己的调度队列维护可执行任务。
(2)队列是双端的,支持LIFO(last-in-first-out)也支持FIFO(first-in-first-out)。
(3)一个任务fork的子任务,只会push到它所在线程的队列。
(4)工作线程使用LIFO通过pop处理它自己队列中的线程。
(5)当线程自己没有待处理任务时,它尝试去随机读取(窃取)一个worker线程的工作队列任务(使用FIFO)。
(6)当线程进入join操作,它开始处理其它线程的任务(自己的已经处理完了),直到目标任务完成(通过isDone方法)。因此,所有任务都无阻塞的完成了。
(7)当一个工作线程没有任务了,并且尝试从其它线程处窃取也失败了,它让出资源(通过使用yields, sleeps或者其它优先级调整)并且随后会再次激活,直到所有工作线程都空闲了——此时,它们都阻塞在等待另一个顶层线程的调用。
两种接口:
RecursiveTask: 带返回值,底层用的是Callable接口
RecursiveAction:无返回值,Runnable接口
ForkJoinPool源码中重要字段:
eventCount三种含义: 1、线程被挂起次数 2、小于零表示没有被激活状态 3、标明为第几个工作线程 nextWait: 由于fork/join维护着一个挂起线程链,这个链是栈结构,本字段就是遍历挂起栈用的 qlock: 起到锁的作用 top、base : 分别是存储工作线程队列的头和尾 poolIndex: 指的是工作线程在队列中的下标
核心方法scan流程解析:
设计亮点:
1、互帮互助的设计思路
2、多字段打包操作
示例代码:
完成累加运算
import java.util.ArrayList; import java.util.concurrent.ForkJoinPool; import java.util.concurrent.ForkJoinTask; import java.util.concurrent.RecursiveTask; /** * @Author: Allen * @Description: 用ForkJoinPool实现累加操作 * @Date: Created in 12:42 2018/2/27 * @Modify By: */ public class CountTask extends RecursiveTask<Long> { private static final int THRESHOLD = 10000; private long start; private long end; public CountTask(long start,long end){ this.start = start; this.end = end; } @Override protected Long compute() { long sum = 0; boolean canCompute = (end-start)<THRESHOLD; if (canCompute) // 是否可以再分 for (long i = start; i <= end; i++) { sum+=i; } else{ long step = (end+start)/100000; // 分成100个任务 ArrayList<CountTask> tasks = new ArrayList<>(); // 任务数组 long pos = step; // 每个任务的大小 for (int i = 0; i < 100000; i++) { long lastOne = pos+step; // 根据任务大小获得一个任务 if (lastOne > end) lastOne = end; // 上限超出的话就设置上限 CountTask subTask = new CountTask(pos,lastOne); // 新建一个任务 System.out.println("new task inf : "+pos+"\t"+lastOne); pos+=step+1; // 更新下一个任务的初始位置 tasks.add(subTask); // 将任务都加入到任务数组中 subTask.fork(); // 执行任务 } // 获取累加值 for (CountTask t : tasks) { sum+=t.join(); } } return sum; } public static void main(String[] args) { ForkJoinPool forkJoinPool = new ForkJoinPool(); // 默认是用cpu的最大核数去工作 CountTask task = new CountTask(0,200000000L); ForkJoinTask<Long> result = forkJoinPool.submit(task); // 提交一个任务 try { long res = result.get(); System.out.println("sum= "+res); }catch (Exception e){ e.printStackTrace(); } } }