Java线程池及其底层源码实现分析

public class TestFuture{
2
    public static void main(String[] args){
3
        //FutureTask实现类Runnable和Future接口
4
        FutureTask<Integer> task = new FutureTask<>(
5
            Thread.sleep(500);//阻塞等待500毫秒
6
            return 1000;
7
        );
8
        
9
        //new的方式启动线程任务
10
        new Thread(task).start();
11
        System.out.println(task.get());//阻塞等待500毫秒后得到返回值
12
        ///////////////////////////////////////////////////////////////////
13
        ExecutorService service = Executors.newFixedThreadPool(5);//创建5个线程的线程池
14
        Future<Integer> future = service.submit(()->{//相当于运行类callable接口的call方法，返回1
15
            Thread.sleep(500);
16
            return 1;
17
        });
18
        System.out.println(future.get());//阻塞等待500毫秒后得到返回值
19
    }
20
}

public class WorkStealingPoolTest {
2
    public static void main(String[] args) throws IOException {
3
        //根据CPU核数启动相应个数的线程（4核cpu---4个线程）
4
        ExecutorService service = Executors.newWorkStealingPool();
5
        System.out.println(Runtime.getRuntime().availableProcessors());
6
7
        service.execute(new R(1000));//线程1执行任务1----1秒
8
        service.execute(new R(2000));//线程2执行任务2----2秒
9
        service.execute(new R(2000));//线程3执行任务3----2秒
10
        service.execute(new R(2000));//线程4执行任务4----2秒
11
        service.execute(new R(2000));//任务5阻塞，当线程1执行完后把任务5偷过来执行
12
        
13
        //由于产生的是守护线程，主线程不阻塞的话，看不到输出
14
        System.in.read();//将主线程阻塞 
15
    }
16
17
    static class R implements Runnable {
18
19
        int time;
20
21
        R(int t) {
22
            this.time = t;
23
        }
24
25
        @Override
26
        public void run() {
27
            
28
            try {
29
                TimeUnit.MILLISECONDS.sleep(time);
30
            } catch (InterruptedException e) {
31
                e.printStackTrace();
32
            }
33
            //打印线程名---ForkJoinPool
34
            System.out.println(time  + " " + Thread.currentThread().getName());
35
            
36
        }
37
38
    }
39
}

public class ForkJoinPoolTest{
2
        
3
        public static void main(String[] args) throws Exception {
4
        ForkJoinPool pool = new ForkJoinPool();
5
        MyTask task = new MyTask(inits, 0, inits.;ength-1);
6
        ForkJoinTask<int[]> taskResult = pool.submit(task);
7
        try {
8
            taskResult.get();//阻塞等待所有线程结果计算完成
9
        } catch (InterruptedException | ExecutionException e) {
10
            e.printStackTrace(System.out);
11
        }
12
    }
13
14
    /**
15
     * 单个排序的子任务
16
     */
17
    static class MyTask extends RecursiveTask<int[]> {
18
19
        private int[] source;
20
        private int start;
21
        private int end;
22
23
        public MyTask(int[] source,int start, int end ) {
24
            this.source = source;
25
            this.start = start;
26
            this.end = end;
27
        }
28
29
30
        @Override
31
        protected int[] compute() {
32
            //长度小于50，进行计算
33
            if(source.length <= 50) {
34
                long sum = 0L;
35
                for(int i=start; i<end; i++) sum += nums[i];
36
                return sum;
37
            } 
38
            //长度大于50，继续划分子任务
39
            int middle = start + (end-start)/2;
40
            
41
            AddTask subTask1 = new MyTask(source,start,middle);
42
            AddTask subTask2 = new MyTask(source,middle,end);
43
            subTask1.fork();//递归创建子任务线程
44
            subTask2.fork();
45
            
46
            //计算完成后将两个子任务的结果合并
47
            return subTask1.join() + subTask2.join();
48
        }
49
    }
50
}