Java7 Fork/Join线程窃取学习笔记

阳光穿透心脏的1/2处 2023-07-03 10:47 24阅读 0赞

### 文章目录 ###

*  前言
 *  show the code
 *   *  show the result
 *  笔记
 *  官网介绍
 *  参考博客

# 前言 #

线程窃取算法是一种思想，就是将大的任务瓜分为小的任务，然后当线程空闲时，会去争抢其他空闲的任务。  
个人劝告：如果不了解这个技术，大家要谨慎使用，对于其他新技术也是一样。

# show the code #

import com.alibaba.fastjson.JSON;
    
    import java.util.Random;
    import java.util.concurrent.ForkJoinPool;
    import java.util.concurrent.ForkJoinTask;
    import java.util.concurrent.RecursiveTask;
    
    class SumTask extends RecursiveTask<Long> {
    
        public static void main(String[] args) throws Exception {
            // 创建随机数组成的数组:
            long[] array = new long[400];
            Random random = new Random(100);
            for (int i = 0; i < 400; i++) {
                array[i] = random.nextInt(100);
            }
            System.out.println(JSON.toJSONString(array));
            // fork/join task:
            ForkJoinPool fjp = new ForkJoinPool(4); // 最大并发数4
            ForkJoinTask<Long> task = new SumTask(array, 0, array.length);
            long startTime = System.currentTimeMillis();
            Long result = fjp.invoke(task);
            long endTime = System.currentTimeMillis();
            System.out.println("Fork/join sum: " + result + " in " + (endTime - startTime) + " ms.");
        }
    
        static final int THRESHOLD = 100;
        long[] array;
        int start;
        int end;
    
        SumTask(long[] array, int start, int end) {
            this.array = array;
            this.start = start;
            this.end = end;
        }
    
        @Override
        protected Long compute() {
            if (end - start <= THRESHOLD) {
                // 如果任务足够小,直接计算:
                long sum = 0;
                for (int i = start; i < end; i++) {
                    sum += array[i];
                }
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                }
                System.out.println(String.format("compute %d~%d = %d", start, end, sum));
                return sum;
            }
            // 任务太大,一分为二:
            int middle = (end + start) / 2;
            System.out.println(String.format("split %d~%d ==> %d~%d, %d~%d", start, end, start, middle, middle, end));
            SumTask subtask1 = new SumTask(this.array, start, middle);
            SumTask subtask2 = new SumTask(this.array, middle, end);
            invokeAll(subtask1, subtask2);
            /*subtask1.invoke();
            subtask2.invoke();*/
            Long subresult1 = subtask1.join();
            Long subresult2 = subtask2.join();
            Long result = subresult1 + subresult2;
            System.out.println("result = " + subresult1 + " + " + subresult2 + " ==> " + result);
            return result;
        }
    }

## show the result ##

> \[15,50,74,88,91,66,36,88,23,13,22,17,56,57,52,59,80,78,73,19,53,28,65,72,67,31,48,92,0,28,74,95,16,73,44,94,87,68,6,29,55,0,39,71,31,2,85,15,62,0,58,36,19,8,59,45,25,52,45,48,46,57,22,54,88,34,53,77,11,71,30,56,0,51,24,63,92,32,87,83,46,26,98,93,34,71,94,12,33,38,26,28,90,53,79,39,59,26,55,52,10,91,21,59,57,62,68,16,84,24,4,78,10,0,8,12,34,97,86,84,39,57,45,88,52,38,69,43,30,0,8,16,49,16,35,5,41,55,83,51,38,66,98,38,99,11,97,97,92,62,59,21,55,78,9,55,53,67,90,76,43,48,79,0,11,1,95,71,30,9,9,88,27,94,99,82,26,38,15,74,92,25,89,20,54,89,79,7,89,47,68,28,88,71,97,1,53,36,75,83,7,77,11,2,38,24,79,35,29,14,93,93,17,48,47,39,10,88,13,33,47,60,65,66,63,4,20,67,35,39,93,6,16,1,15,26,33,96,36,3,94,30,15,78,47,13,26,76,28,26,49,36,88,68,86,42,70,12,94,9,36,36,98,33,57,69,20,84,88,78,44,20,79,30,2,75,76,61,66,13,61,32,51,54,83,24,34,88,11,50,26,52,76,97,26,31,42,11,51,71,64,73,60,93,54,19,53,94,65,49,87,24,15,80,77,19,80,76,72,4,92,67,53,20,72,67,9,36,99,90,94,88,49,79,96,9,41,96,12,63,49,55,32,60,96,74,37,53,81,16,19,59,91,81,62,17,76,14,34,48,28,27,68,51,46,70,8,2,14,66,47,75,49,60,46,82,90,5,38,22,36,64,20,15,62,23,77,26,59,54,70,43,5,32,73,64,19,68,91,80\]  
> split 0~400 ==> 0~200, 200~400  
> split 0~200 ==> 0~100, 100~200  
> split 200~400 ==> 200~300, 300~400  
> compute 200~300 = 4611  
> compute 300~400 = 5319  
> compute 0~100 = 4965  
> compute 100~200 = 5116  
> result = 4611 + 5319 ==> 9930  
> result = 4965 + 5116 ==> 10081  
> result = 10081 + 9930 ==> 20011  
> Fork/join sum: 20011 in 1013 ms.

# 笔记 #

上面这个例子可以看到，会将大的任务去瓜分。从结果来看就是，大家先分好任务，分完各自执行，join等待两个任务完成相加，然后返回值。

这里为啥是先分任务？跟invokeAll有关系。  
如果我们代码变成：

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODMzNjY1OA_size_16_color_FFFFFF_t_70]  
拆一个之后，执行，执行之后再拆，自然就慢了。

所以下面廖大就说要正确使用fork/join

# 官网介绍 #

*  [类ForkJoinTask ][ForkJoinTask]

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODMzNjY1OA_size_16_color_FFFFFF_t_70 1]  
解读一下：

1.  ForkJoinPool蛮熟悉的吧，Future异步甚至一些框架也使用它来作为线程池。
2.  fork=invoke,join等待结果，跟future.get有点相似
3.  invokeAll是会开启所有任务的，跟之前讲的一致。
4.  RecursiveAction用于没有返回值， RecursiveTask用于有返回值，大家开发的时候继承一下

# 参考博客 #

这里搬出雪峰老师的教程，可以说学java都会认识吧  
[Java的Fork/Join任务，你写对了吗？][Java_Fork_Join]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODMzNjY1OA_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20200205190844592.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODMzNjY1OA==,size_16,color_FFFFFF,t_70
[ForkJoinTask]: https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/ForkJoinTask.html
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODMzNjY1OA_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20200205191305166.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODMzNjY1OA==,size_16,color_FFFFFF,t_70
[Java_Fork_Join]: https://www.liaoxuefeng.com/article/1146802219354112