成果被他人窃取_工作窃取模式

class Fibonacci extends RecursiveTask<Integer>

import java.util.concurrent.*;
import java.util.stream.LongStream;
/** * 求和计算 * 1.最low的：循环求和 * 2.一般的：ForkJoin分支求和 * 3.最快的：Stream并行流求和 */
public class ForkJoinDemo { 

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

new ForkJoinDemo().test1();
new ForkJoinDemo().test2();
new ForkJoinDemo().test3();
/** * test1===>和为500000000500000000,耗时9937 * test2===>和为500000000500000000,耗时8622 * test3===>和为500000000500000000,耗时276 */
}
//1.最low的：循环求和
public void test1() { 

long start = System.currentTimeMillis();
Long sum = 0L;
for (Long i = 0L; i <= 10_0000_0000L; i++) { 

sum += i;
}
long end = System.currentTimeMillis();
System.out.println("test1===>和为" + sum + ",耗时" + (end - start));
}
//2.一般的：ForkJoin分支求和
public void test2() throws ExecutionException, InterruptedException { 

long start = System.currentTimeMillis();
// 也可以使用公用的线程池 ForkJoinPool.commonPool()：
// pool = ForkJoinPool.commonPool()
ForkJoinPool pool = new ForkJoinPool();
ForkJoinTask<Long> task = pool.submit(new MySumForkJoin(0L, 10_0000_0000L));//提交任务
Long sum = task.get();
long end = System.currentTimeMillis();
System.out.println("test2===>和为" + sum + ",耗时" + (end - start));
}
//3.最快的：Stream并行流求和
public void test3() { 
//没有Long的拆箱装箱操作
long start = System.currentTimeMillis();
long sum = LongStream.rangeClosed(0L, 10_0000_0000L).parallel().reduce(0, Long::sum);//reduce第一个参数为0，为了保证，在进行流条件计算下，保证每次累计的值不受多线程影响
long end = System.currentTimeMillis();
System.out.println("test3===>和为" + sum + ",耗时" + (end - start));
}
}
/** * 如何让使用ForkJoin * 1.ForkJoinPool，通过ForkJoinPool来执行 * 2.计算任务forkJoinPool.execute(ForkJoinTask task) * 3.计算类要继承ForkJoinTask(执行任务RecursiveTask：有返回值 RecursiveAction：无返回值) */
class MySumForkJoin extends RecursiveTask<Long> { 

private Long start;//开始值
private Long end;//结束值
private Long criticalValue = 10000L;//临界值，超过这个值开始分任务执行
public MySumForkJoin(Long start, Long end) { 

this.start = start;
this.end = end;
}
//计算方法:递归调用的
@Override
protected Long compute() { 

if (end - start <= criticalValue) { 

Long sum = 0L;
for (Long i = start; i <= end; i++) { 

sum += i;
}
return sum;
} else { 
//ForkJoin执行分支合并计算，
// System.out.println("开始ForkJoin执行分支合并计算");
//1.先求中间值
Long middle = (start + end) / 2;//不直接(start+end)/2，防止溢出
/*把任务一分为二，递归拆分(注意此处有递归)到底拆分成多少分 需要根据具体情况而定*/
MySumForkJoin task1 = new MySumForkJoin(start, middle);
MySumForkJoin task2 = new MySumForkJoin(middle + 1, end);
// task1.fork();//拆分任务，把任务压入双端队列，这里手动只拆分成两个任务，可以拆分多个
// task2.fork();//拆分任务，把任务压入双端队列
invokeAll(task1, task2);//拆分任务，把任务压入双端队列
/** * compute分出多个task后：以task1，task2为例 * 错误做法： * 1）依次执行task1.fork()，task2.fork() * 2）依次执行task1.join()，task2.join() * 正确做法： * 1）直接调用invokeAll(task1，task2) * * fork(): * 把task置入当前ForkJoinWorkerThread的queue中，等待被消费 * join()： * 当前ForkJoinWorkerThread执行等待该task执行结束 * * 错误做法过程： * 将task1与task2置入queue中，当前ForkJoinWorkerThread停下来等待task1，task2的执行结束。 * * 分析： * 当前的ForkJoinWorkerThread可以说join之后什么事情都没有做，只是等待ing。而task1和task2会在新的线程中执行。 * 会浪费当前ForkJoinWorkerThread的宝贵线程资源。而且最糟糕的是，会创建一个新的ForkJoinWorkerThread去执行新的task，可想而知，如果任务量特别多，那会同时开启特别多的线程。 */
return task1.join() + task2.join();
}
}
}