Fork/Join——ForkJoinPool核心方法（一）

迷南。 2021-08-30 18:02 333阅读 0赞

### ForkJoinPool核心方法 ###

#### 1. ####

方法
    public vod execute(ForkJoinTask<?> task)
    的使用

在ForkJoinPool.java类中的execute()方法是以异步的方式执行任务。

public class MyRun1 { 
        public static void main(String[] args) { 
            try { 
                ForkJoinPool pool = new ForkJoinPool();
                pool.execute(new ActionA());
                Thread.sleep(5000);
            } catch (InterruptedException e) { 
                e.printStackTrace();
            }
        }
    }
    
    class ActionA extends RecursiveAction { 
    
        @Override
        protected void compute() { 
            System.out.println("ThreadName=" + Thread.currentThread().getName());
        }
    }

![在这里插入图片描述][20200625125743735.png]

#### 2. ####

方法
    public void execute(Runnable task)
    的使用

public class MyRun2 { 
        public static void main(String[] args) throws InterruptedException { 
            ForkJoinPool pool = new ForkJoinPool();
            pool.execute(() -> { 
                System.out.println("ThreadName=" + Thread.currentThread().getName());
            });
            Thread.sleep(5000);
        }
    }

![在这里插入图片描述][20200625125945536.png]

#### 3. ####

方法
    public void execute(ForkJoinTask<?> task)
    如何处理返回值

public class MyRun3 { 
        public static void main(String[] args) { 
            try { 
                ActionB task = new ActionB();
                ForkJoinPool pool = new ForkJoinPool();
                pool.execute(task);
                //execute方法无返回值，想取得返回值得通过RecursiveTask对象
                System.out.println(System.currentTimeMillis());
                System.out.println(task.get());
                System.out.println(System.currentTimeMillis());
                Thread.sleep(5000);
            } catch (InterruptedException e) { 
                e.printStackTrace();
            } catch (ExecutionException e) { 
                e.printStackTrace();
            }
        }
    }
    
    class ActionB extends RecursiveTask<String> { 
    
        @Override
        protected String compute() { 
            try { 
                Thread.sleep(5000);
            } catch (InterruptedException e) { 
                e.printStackTrace();
            }
            return "我是返回值";
        }
    }

虽然public void execute(ForkJoinTask<?> task)方法无返回值，但还是可以通过RecursiveTask对象处理返回值。调用get()方法即阻塞。  
![在这里插入图片描述][20200625130607198.png]

#### 4. ####

方法
    public<T> ForkJoinTask<T> submit(ForkJoinTask<T> task)
    的使用

方法execute()无返回值，submit()有返回值。

public class MyRun4 { 
        public static void main(String[] args) { 
            try { 
                ActionC task = new ActionC();
                ForkJoinPool pool = new ForkJoinPool();
                ForkJoinTask<String> returnTask = pool.submit(task);
                System.out.println("begin " + System.currentTimeMillis());
                System.out.println(returnTask.get());
                System.out.println("end " + System.currentTimeMillis());
            } catch (InterruptedException e) { 
                e.printStackTrace();
            } catch (ExecutionException e) { 
                e.printStackTrace();
            }
        }
    }
    
    class ActionC extends RecursiveTask<String> { 
    
        @Override
        protected String compute() { 
            try { 
                Thread.sleep(5000);
            } catch (InterruptedException e) { 
                e.printStackTrace();
            }
            return "我是返回值";
        }
    }

![在这里插入图片描述][20200625131340298.png]

#### 5. ####

方法
    public ForkJoinTask<?> submit(Runnable task)
    的使用

public class MyRun5 { 
        public static void main(String[] args) { 
            try { 
                ForkJoinPool pool = new ForkJoinPool();
                System.out.println("begin " + System.currentTimeMillis());
                ForkJoinTask<?> submit = pool.submit(() -> { 
                    try { 
                        Thread.sleep(5000);
                        System.out.println("ThreadName=" + Thread.currentThread().getName());
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    }
                });
                System.out.println(submit.get());
                System.out.println("end " + System.currentTimeMillis());
                Thread.sleep(5000);
            } catch (InterruptedException e) { 
                e.printStackTrace();
            } catch (ExecutionException e) { 
                e.printStackTrace();
            }
        }
    }

任务成功被运行，传入Runnable接口虽然没有返回值，但调用get()方法呈阻塞状态。  
![在这里插入图片描述][20200625131711775.png]

#### 6. ####

方法
    public <T> ForkJoinTask<T> submit(Callable<T> task)
    的使用

public class MyRun6 { 
        public static void main(String[] args) { 
            try { 
                MyCallable callable = new MyCallable();
                ForkJoinPool pool = new ForkJoinPool();
                System.out.println("begin " + System.currentTimeMillis());
                ForkJoinTask<String> returnTask = pool.submit(callable);
                System.out.println(returnTask.get());
                System.out.println("end " + System.currentTimeMillis());
            } catch (InterruptedException e) { 
                e.printStackTrace();
            } catch (ExecutionException e) { 
                e.printStackTrace();
            }
        }
    }
    
    class MyCallable implements Callable<String> { 
    
        @Override
        public String call() throws Exception { 
            try { 
                Thread.sleep(3000);
            } catch (InterruptedException e) { 
                e.printStackTrace();
            }
            return "我是返回值callableVersion";
        }
    }

![在这里插入图片描述][20200625132213162.png]

#### 7 ####

方法
    public <T> ForkJoiinTask<T> submit(Runnable task, T result)
    的使用

public class MyRun7 { 
        public static void main(String[] args) throws InterruptedException { 
            UserInfo userInfo = new UserInfo();
            RunnableT runnableT = new RunnableT(userInfo);
    
            ForkJoinPool pool = new ForkJoinPool();
            pool.submit(runnableT, userInfo);
            //取不到值
            System.out.println("userinfo username=" + userInfo.getUsername());
            Thread.sleep(2000);
        }
    }
    
    class UserInfo { 
        private String username;
    
        public UserInfo() { 
        }
    
        public UserInfo(String username) { 
            this.username = username;
        }
    
        public String getUsername() { 
            return username;
        }
    
        public void setUsername(String username) { 
            this.username = username;
        }
    }
    
    class RunnableT implements Runnable { 
        private UserInfo userInfo;
    
        public RunnableT(UserInfo userInfo) { 
            this.userInfo = userInfo;
        }
    
        @Override
        public void run() { 
            try { 
                userInfo.setUsername("设置的值");
                System.out.println("已经设置完结！");
                Thread.sleep(3000);
            } catch (InterruptedException e) { 
                e.printStackTrace();
            }
        }
    }

![在这里插入图片描述][20200625132850493.png]  
运行结果是未取到值，因为是异步运行的，所以要加一个延时：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70]  
![在这里插入图片描述][20200625133059888.png]  
成功取到值，但程序还是有一些问题，也就是sleep(的时间并不能真正的确定，因为不知道Task任务执行完需要多少时间，更改代码如下：

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 1]  
![在这里插入图片描述][20200625133424989.png]

#### 8. ####

方法
    public <T> List<Futre<T>> invokeAll(Collection<? extends Calable<T>> tasks)
    的使用

该方法具有阻塞特性。

public class MyRun8 { 
        public static void main(String[] args) { 
            try { 
                List list = new ArrayList();
                list.add(new CallableC(5000));
                list.add(new CallableC(4000));
                list.add(new CallableC(3000));
                list.add(new CallableC(2000));
                list.add(new CallableC(1000));
    
                ForkJoinPool pool = new ForkJoinPool();
                List<Future<String>> listFuture = pool.invokeAll(list);
                for (int i = 0; i < listFuture.size(); i++) { 
                    System.out.println(listFuture.get(i).get() + " nowTime: " + System.currentTimeMillis());
                }
            } catch (InterruptedException e) { 
                e.printStackTrace();
            } catch (ExecutionException e) { 
                e.printStackTrace();
            }
        }
    }
    
    class CallableC implements Callable<String> { 
        private long sleepValue;
    
        public CallableC(long sleepValue) { 
            this.sleepValue = sleepValue;
        }
    
        @Override
        public String call() throws Exception { 
            try { 
                System.out.println(Thread.currentThread().getName() + " sleep" + sleepValue + " nowTime: " +
                        System.currentTimeMillis());
                Thread.sleep(sleepValue);
            } catch (InterruptedException e) { 
                e.printStackTrace();
            }
            return "我是返回值";
        }
    }

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 2]

#### 9. ####

方法
    public void shutdown()
    的使用

public class Test1 { 
        public static void main(String[] args) throws InterruptedException { 
            ForkJoinPool pool = new ForkJoinPool();
            MyRunnable1 r = new MyRunnable1();
            pool.submit(r);
            Thread.sleep(1000);
            pool.shutdown();
            System.out.println("main end!");
            Thread.sleep(Integer.MAX_VALUE);
        }
    }
    
    class MyRunnable1 implements Runnable { 
    
        @Override
        public void run() { 
            try { 
                System.out.println("begin " + Thread.currentThread().getName() + " " + System.currentTimeMillis());
                Thread.sleep(4000);
                System.out.println("end " + Thread.currentThread().getName() + " " + System.currentTimeMillis());
            } catch (InterruptedException e) { 
                e.printStackTrace();
            }
        }
    }

调用shutdown()方法后任务依然运行，直到结束。  
![在这里插入图片描述][20200625141454991.png]

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 3]  
上面代码运行后进程马上被销毁，说明对ForkJoinPool对象调用shutdown()方法后再执行任务时出现异常，进程也就马.上销毁了，而正在运行的线程任务也被销毁掉了，运行的效果当然就是字符串“end” 没有输出，仅仅打印了“begin"。  
![在这里插入图片描述][20200625141659828.png]

为了防止在关闭pool后再运行任务，可以加入一个判断来解决进程意外销毁的问题：

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 4]  
![在这里插入图片描述][20200625141853753.png]  
`注意：`shutdown()方法不具有中断的效果，也就是shutdown()方法遇到MyRunnable1中的sleep()方法没有发生中断异常。

#### 10. ####

方法
    public List<Runnable> shutdownNow()
    的使用

public class Demo1 { 
        public static void main(String[] args) throws InterruptedException { 
            MyRunnable111 myRunnable111 = new MyRunnable111();
            ForkJoinPool pool = new ForkJoinPool();
            pool.submit(myRunnable111);
            Thread.sleep(2000);
            pool.shutdownNow();
            System.out.println("main end");
            Thread.sleep(Integer.MAX_VALUE);
        }
    }
    
    class MyRunnable111 implements Runnable { 
    
        @Override
        public void run() { 
            try { 
                for (int i = 0; i < Integer.MAX_VALUE; i++) { 
                    String newString = new String();
                    Math.random();
                    Math.random();
                    Math.random();
                    Math.random();
                    Math.random();
                    Math.random();
                    if (Thread.currentThread().isInterrupted() == true) { 
                        System.out.println("任务没有完成，就中断了！");
                        throw new InterruptedException();
                    }
                }
                System.out.println("任务完成了！");
            } catch (InterruptedException e) { 
                System.out.println("进入Catch中断了任务");
                e.printStackTrace();
            }
        }
    }
    
    class MyRunnable222 implements Runnable { 
    
        @Override
        public void run() { 
            for (int i = 0; i < Integer.MAX_VALUE / 100; i++) { 
                new String();
                Math.random();
                Math.random();
                Math.random();
                Math.random();
                Math.random();
                Math.random();
            }
            System.out.println("任务完成！");
        }
    }

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 5]

MyRunnable222 myRunnable222 = new MyRunnable222();
            ForkJoinPool pool = new ForkJoinPool();
            pool.execute(myRunnable222);
            Thread.sleep(1000);
            pool.shutdownNow();
            System.out.println("main end");
            Thread.sleep(Integer.MAX_VALUE);

![在这里插入图片描述][20200625143659789.png]

MyRunnable222 myRunnable222 = new MyRunnable222();
    ForkJoinPool pool = new ForkJoinPool();
    pool.execute(myRunnable222);
    Thread.sleep(1000);
    pool.shutdownNow();//返回一个空的List
    pool.execute(myRunnable222);
    System.out.println("main end");
    Thread.sleep(Integer.MAX_VALUE);

![在这里插入图片描述][20200625143843597.png]

*  `shutdown`:  
    每个任务正常运行直到结束，池关闭后不再允许有新任务被执行并抛出`RejectedExecutionException`异常。
 *  `shutdownNow结合isInterrupted() ==true 判断`:  
    立即停止当前正在执行的任务，池关闭后不再允许有新任务被执行并抛出RejectedExecutionException异常。
 *  `shutdownNow未结合isInterrupted() ==true 判断`:  
    每个任务正常运行直到结束，池关闭后不再允许有新任务被执行并抛出RejectedExecut ionException异常。

#### 11. ####

方法
    isTerminating()和isTerminated()
    使用

public class Demo2 { 
        public static void main(String[] args) throws ExecutionException, InterruptedException { 
            Runnable runnable = new Runnable() { 
                @Override
                public void run() { 
                    for (int i = 0; i < Integer.MAX_VALUE / 100; i++) { 
                        String newString = new String();
                        Math.random();
                        Math.random();
                        Math.random();
                        Math.random();
                        Math.random();
                        Math.random();
                    }
                }
            };
            ForkJoinPool pool = new ForkJoinPool();
            ForkJoinTask<?> submit = pool.submit(runnable);
            Thread.sleep(500);
            System.out.println("A=" + pool.isTerminating());
            pool.shutdown();
            System.out.println("B=" + pool.isTerminating());
            System.out.println(submit.get());
            Thread.sleep(1000);
            System.out.println("C=" + pool.isTerminated());
        }
    }

shutdown()方法关闭池之前，isTerminating()方法的返回值一直是false。  
![在这里插入图片描述][20200625173810853.png]

public class Demo2 { 
        public static void main(String[] args) throws ExecutionException, InterruptedException { 
            Runnable runnable = new Runnable() { 
                @Override
                public void run() { 
                    for (int i = 0; i < Integer.MAX_VALUE / 100; i++) { 
                        String newString = new String();
                        Math.random();
                        Math.random();
                        Math.random();
                        Math.random();
                        Math.random();
                        Math.random();
                    }
                }
            };
            ForkJoinPool pool = new ForkJoinPool();
            ForkJoinTask<?> submit = pool.submit(runnable);
            Thread.sleep(500);
            System.out.println("A=" + pool.isTerminating());
            pool.shutdownNow();
            System.out.println("B=" + pool.isTerminating());
            System.out.println(submit.get());
            Thread.sleep(1000);
            System.out.println("C=" + pool.isTerminated());
        }
    }

注意:在控制台输出了一个java.util.concurrent.CancellationException异常信息，说明先调用shutdown()，再调用get() 方法不出现异常，而先调用shutdownNow(),再调用get()方法出现异常CancellationException，说明方法shutdown()与shutdownNow()在对get()方法的处理行为上是不一样的。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 6]

public class Demo2 { 
        public static void main(String[] args) throws ExecutionException, InterruptedException { 
            Runnable runnable = new Runnable() { 
                @Override
                public void run() { 
                    for (int i = 0; i < Integer.MAX_VALUE / 100; i++) { 
                        String newString = new String();
                        Math.random();
                        Math.random();
                        Math.random();
                        Math.random();
                        Math.random();
                        Math.random();
                    }
                }
            };
            ForkJoinPool pool = new ForkJoinPool();
            ForkJoinTask<?> submit = pool.submit(runnable);
            Thread.sleep(500);
            System.out.println("A=" + pool.isTerminating());
            pool.shutdownNow();
            System.out.println("B=" + pool.isTerminating());
            Thread.sleep(30000);
            System.out.println("C=" + pool.isTerminated());
        }
    }

![在这里插入图片描述][20200625174358587.png]

[20200625125743735.png]: /images/20210813/a9610d3419354efb918ad82a48c67f76.png
[20200625125945536.png]: /images/20210813/fef3e724360f45d691bee1a914f69cc1.png
[20200625130607198.png]: /images/20210813/42f2593d9de5455b8e12d49d900c444e.png
[20200625131340298.png]: /images/20210813/37d64b6d12274c018b3b8c3cd6ca8da1.png
[20200625131711775.png]: /images/20210813/c8c7504577674d36ba2d54e6bc5d8be4.png
[20200625132213162.png]: /images/20210813/2745393e978e4881833d29ca9dd52683.png
[20200625132850493.png]: /images/20210813/66db0bca18b146c4b2330d5645d5c789.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70]: /images/20210813/d1817cf6a2b54ecda98f941d9a64ea38.png
[20200625133059888.png]: /images/20210813/919916093dc14c4897ca372c6a48d4a0.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 1]: /images/20210813/01a686d62a084a9f81eb4338875853e9.png
[20200625133424989.png]: /images/20210813/69a8e80fc1384c819b4a599ba43da063.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 2]: /images/20210813/4e5b861f4b0f432aa65494d596fed5a7.png
[20200625141454991.png]: /images/20210813/4549009df1994fb5ba87b077fe59329e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 3]: /images/20210813/07320adee1e54f03b6ea60785c74677e.png
[20200625141659828.png]: /images/20210813/5bfd645cf2f94972aaac9d2d84f02cc6.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 4]: /images/20210813/8c627ff5e6294a7b8f8faef7dab5c117.png
[20200625141853753.png]: /images/20210813/8dcd42b5d2af4753873423a362ea4fdc.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 5]: /images/20210813/e9a05a5eb94f438ea696a26def623e87.png
[20200625143659789.png]: /images/20210813/70a14eede51540d8a9cff6679da6206c.png
[20200625143843597.png]: /images/20210813/ffe42973588c40b39b1b91245ce04691.png
[20200625173810853.png]: /images/20210813/a82b38ad60f84df9bfa66fdfecf8eee3.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 6]: /images/20210813/b78dda7b490a4cce890ec5e6bc03fcd2.png
[20200625174358587.png]: /images/20210813/a6347b16cc3948718ac6f78a87faf5c4.png