Java并发——ScheduledExecutorService

桃扇骨 2021-08-30 15:55 341阅读 0赞

### ScheduledExecutorService ###

类 Scheduledexecutor Service的主要作用就是可以将定时任务与线程池功能结合使用。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70]

#### ScheduledThreadPoolExecutor使用Callable延迟运行 ####

public class MyCallableAA implements Callable<String> { 
        @Override
        public String call() throws Exception { 
            try { 
                System.out.println("callA begin " + Thread.currentThread().getName() + " " + System.currentTimeMillis());
                Thread.sleep(3000);
                System.out.println("callA end " + Thread.currentThread().getName() + " " + System.currentTimeMillis());
            } catch (Exception e) { 
                e.printStackTrace();
            }
            return "returnA";
        }
    }
    
    
    
    public class MyCallableBB implements Callable<String> { 
        @Override
        public String call() throws Exception { 
            System.out.println("callB begin " + Thread.currentThread().getName() + " " + System.currentTimeMillis());
            System.out.println("callB end " + Thread.currentThread().getName() + " " + System.currentTimeMillis());
            return "returnB";
        }
    }
    
    
    public class MyRun1 { 
        public static void main(String[] args) { 
            try { 
                List<Callable> callables = new ArrayList<>();
                callables.add(new MyCallableAA());
                callables.add(new MyCallableBB());
                //调用方法newSingleThreadScheduledExecutor()，取得一个单任务的计划任务执行池
                ScheduledExecutorService executorService = Executors.newSingleThreadScheduledExecutor();
                ScheduledFuture<String> futureA = executorService.schedule(callables.get(0), 4L, TimeUnit.SECONDS);
                ScheduledFuture<String> futureB = executorService.schedule(callables.get(1), 4L, TimeUnit.SECONDS);
    
                System.out.println(" X=" + System.currentTimeMillis());
                System.out.println("返回值A：" + futureA.get());
                System.out.println("返回值B：" + futureB.get());
                System.out.println(" Y=" + System.currentTimeMillis());
            } catch (InterruptedException e) { 
                e.printStackTrace();
            } catch (ExecutionException e) { 
                e.printStackTrace();
            }
        }
    }

从X到Y的运行时间为7秒，阻塞点是get()方法。  
从运行结果中可以发现：`public <V> Scheduled Future<V> schedule(Callable<v> callable, long delay, Time Unit unit)`  
方法中的第2个参数在多个任务中同时消耗时间,并不是一个任务执行完毕后再等待4秒继续执行的效果。由于第1个任务从计划任务到运行结束需要用时7秒,那么第2个任务其实是想在第4秒被执行,由于是单任务的计划任务池,所以第2个任务的执行时间被延后3秒。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 1]

在此实验中使用工厂类 Executors的 `newSingleThreadScheduledExecutor`方法来创建  
ScheduledExecutorService对象,但返回的真正对象却是 `Scheduledthreadpoolexecutor`,因为  
Scheduled Thread PoolExecutor实现了ScheduledexecutorService接口。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 2]

#### ScheduledThreadPoolExecutor使用Runnable延迟运行 ####

public class MyRunnableAA implements Runnable { 
        @Override
        public void run() { 
            try { 
                System.out.println("runnableA begin " + Thread.currentThread().getName() + " " + System.currentTimeMillis());
                Thread.sleep(3000);
                System.out.println("runnableA end " + Thread.currentThread().getName() + " " + System.currentTimeMillis());
            } catch (Exception e) { 
                e.printStackTrace();
            }
        }
    }
    
    public class MyRunnableBB implements Runnable { 
        @Override
        public void run() { 
            System.out.println("runnableB begin " + Thread.currentThread().getName() + " " + System.currentTimeMillis());
            System.out.println("runnableB end " + Thread.currentThread().getName() + " " + System.currentTimeMillis());
        }
    }
    
    public class MyRun2 { 
        public static void main(String[] args) { 
            List<Runnable> list = new ArrayList<>();
            list.add(new MyRunnableAA());
            list.add(new MyRunnableBB());
            ScheduledExecutorService executorService = Executors.newSingleThreadScheduledExecutor();
    
            System.out.println(" X=" + System.currentTimeMillis());
            executorService.schedule(list.get(0), 0L, TimeUnit.SECONDS);
            executorService.schedule(list.get(1), 1L, TimeUnit.SECONDS);
            System.out.println(" Y=" + System.currentTimeMillis());
        }
    }

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

#### 延迟运行并取得返回值 ####

public class MyRun3 { 
        public static void main(String[] args) { 
            Callable<String> callable = new Callable<String>() { 
                public String call() throws Exception { 
                    System.out.println("a call run =" + System.currentTimeMillis());
                    return "returnA";
                }
            };
    
            try { 
                List<Callable> list = new ArrayList<>();
                list.add(callable);
                ScheduledExecutorService exector = Executors.newSingleThreadScheduledExecutor();
                System.out.println(" X=" + System.currentTimeMillis());
                ScheduledFuture future = exector.schedule(list.get(0), 4L, TimeUnit.SECONDS);
                System.out.println(future.get() + " A=" + System.currentTimeMillis());
            } catch (Exception e) { 
                e.printStackTrace();
            }
        }
    }

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

#### 使用scheduleAtFixedRate()方法实现周期性执行 ####

*  command：执行线程
 *  initialDelay：初始化延时
 *  period：两次开始执行最小间隔时间
 *  unit：计时单位

执行任务时间大于>period预定的周期时间，也就是产生了超时的效果：

public class MyRun4 { 
        public static void main(String[] args) { 
            Runnable runnable = new Runnable() { 
                @Override
                public void run() { 
                    try { 
                        System.out.println(" begin=" + System.currentTimeMillis() +
                                " ThreadName=" + Thread.currentThread().getName());
                        Thread.sleep(4000);
                        System.out.println(" end=" + System.currentTimeMillis() +
                                " ThreadName=" + Thread.currentThread().getName());
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    }
                }
            };
    
            ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor();
            System.out.println(" X=" + System.currentTimeMillis());
            executor.scheduleAtFixedRate(runnable, 1, 2, TimeUnit.SECONDS);
            System.out.println(" Y=" + System.currentTimeMillis());
        }
    }

任务执行时间(4)大于周期时间(2)，所以产生超时效果，每次执行任务不再间隔2秒。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 4]

执行任务时间<period的时间：

public class MyRun5 { 
        public static void main(String[] args) { 
            Runnable runnable = new Runnable() { 
                @Override
                public void run() { 
                    System.out.println(" begin=" + System.currentTimeMillis() +
                            " ThreadName=" + Thread.currentThread().getName());
                    System.out.println(" end=" + System.currentTimeMillis() +
                            " ThreadName=" + Thread.currentThread().getName());
                }
            };
    
            ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor();
            System.out.println(" X=" + System.currentTimeMillis());
            executor.scheduleAtFixedRate(runnable, 1, 2, TimeUnit.SECONDS);
            System.out.println(" Y=" + System.currentTimeMillis());
        }
    }

每次间隔2秒执行。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 5]

`注意:` scheduleAtFixedRatel方法返回的 Scheduled  
Future对象无法获得返回值,也就是`scheduleAtFixedRate()方法不具有获得返回值的功能,而 schedule方法却可以获得返回值。`所以当使用 scheduleAtFixedRate方法实现重复运行任务的效果时,需要结合自定义 Runnable接口的实现类，不要使用FutureTask类，因为FutureTask类并不能实现重复运行的效果。

#### 使用scheduleWithFixedDelay()方法实现周期性执行 ####

方法 schedule With FixedDelay(的主要作用是设置多个任务之间固定的运行时间间隔。

执行任务时间大于> period定的时间

public class MyRun6 { 
        public static void main(String[] args) { 
            Runnable runnable = new Runnable() { 
                @Override
                public void run() { 
                    try { 
                        System.out.println(" begin=" + System.currentTimeMillis() +
                                " ThreadName=" + Thread.currentThread().getName());
                        Thread.sleep(4000);
                        System.out.println(" end=" + System.currentTimeMillis() +
                                " ThreadName=" + Thread.currentThread().getName());
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    }
                }
            };
    
            ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor();
            System.out.println(" X=" + System.currentTimeMillis());
            executor.scheduleWithFixedDelay(runnable, 1, 2, TimeUnit.SECONDS);
            System.out.println(" Y=" + System.currentTimeMillis());
        }
    }

固定每隔2秒执行一次。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 6]

执行任务时间<period：

public class MyRun6 { 
        public static void main(String[] args) { 
            Runnable runnable = new Runnable() { 
                @Override
                public void run() { 
                    System.out.println(" begin=" + System.currentTimeMillis() +
                            " ThreadName=" + Thread.currentThread().getName());
                    System.out.println(" end=" + System.currentTimeMillis() +
                            " ThreadName=" + Thread.currentThread().getName());
                }
            };
    
            ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor();
            System.out.println(" X=" + System.currentTimeMillis());
            executor.scheduleWithFixedDelay(runnable, 1, 2, TimeUnit.SECONDS);
            System.out.println(" Y=" + System.currentTimeMillis());
        }
    }

固定每隔2秒执行一次。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 7]

#### getQueue()与remove()方法 ####

方法 getQueue()的作用是取得队列中的任务,而这些任务是未来将要运行的,正在运行的任务不在此队列中。使用 scheduleAtFixedRateO和 scheduleWithFixedDelay()两个方法实现周期性执行任务时,未来欲执行的任务都是放入此队列中。

public class MyRun7 { 
        public static void main(String[] args) { 
            ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(10);
    
            Runnable_A r1 = new Runnable_A("A");
            Runnable_A r2 = new Runnable_A("B");
            Runnable_A r3 = new Runnable_A("C");
            Runnable_A r4 = new Runnable_A("D");
            Runnable_A r5 = new Runnable_A("E");
    
            System.out.println(r1.hashCode());
            System.out.println(r2.hashCode());
            System.out.println(r3.hashCode());
            System.out.println(r4.hashCode());
            System.out.println(r5.hashCode());
    
            executor.scheduleAtFixedRate(r1, 10, 2, TimeUnit.SECONDS);
            executor.scheduleAtFixedRate(r2, 10, 2, TimeUnit.SECONDS);
            executor.scheduleAtFixedRate(r3, 10, 2, TimeUnit.SECONDS);
            executor.scheduleAtFixedRate(r4, 10, 2, TimeUnit.SECONDS);
            executor.scheduleAtFixedRate(r5, 10, 2, TimeUnit.SECONDS);
    
            System.out.println("");
    
            BlockingQueue<Runnable> queue = executor.getQueue();
            Iterator<Runnable> it = queue.iterator();
            while (it.hasNext()) { 
                Runnable next = it.next();
                System.out.println("队列中的：" + next);
            }
        }
    }
    
    class Runnable_A implements Runnable { 
        private String username;
    
        public Runnable_A(String username) { 
            this.username = username;
        }
    
        public String getUsername() { 
            return username;
        }
    
        @Override
        public void run() { 
            System.out.println("run! username=" + Thread.currentThread().getName());
        }
    }

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

public class MyRun7 { 
        public static void main(String[] args) throws InterruptedException { 
            ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(1);
    
            Runnable_A r1 = new Runnable_A("A");
            Runnable_A r2 = new Runnable_A("B");
    
            ScheduledFuture<?> future1 = executor.scheduleAtFixedRate(r1, 0, 2, TimeUnit.SECONDS);
            Thread.sleep(1000);
            ScheduledFuture<?> future2 = executor.scheduleAtFixedRate(r2, 10, 2, TimeUnit.SECONDS);
            Thread.sleep(5000);
            System.out.println(executor.remove((Runnable) future2));
            System.out.println("");
    
            BlockingQueue<Runnable> queue = executor.getQueue();
            Iterator<Runnable> it = queue.iterator();
            while (it.hasNext()) { 
                Runnable next = it.next();
                System.out.println("队列中的:" + next);
            }
        }
    }
    
    class Runnable_A implements Runnable { 
        private String username;
    
        public Runnable_A(String username) { 
            this.username = username;
        }
    
        public String getUsername() { 
            return username;
        }
    
        @Override
        public void run() { 
            System.out.println("run! username=" + Thread.currentThread().getName());
        }
    }

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

#### setExecuteExistingDelayedTasksAfterShtudownPolicy() ####

方法 `setExecuteExisting DelayedTasksAfterShutdownPolicy()`的作用是当对 ScheduledThreadPoolExecutor执行了shutdown方法时,任务是否继续运行,默认值是true,也就是  
当调用了 shutdown方法时任务还是继续运行,`当使用 set ExecuteExistingDelayedTasksAfterShutdownPolicy(alse)时任务不再运行。`

public class MyRun8 { 
        public static void main(String[] args) { 
            Runnable runnable = new Runnable() { 
                @Override
                public void run() { 
                    try { 
                        Thread.sleep(2000);
                        System.out.println("任务执行！");
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    }
                }
            };
    
            ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(1);
            executor.setExecuteExistingDelayedTasksAfterShutdownPolicy(false);
            executor.schedule(runnable, 3, TimeUnit.SECONDS);
            executor.shutdown();
            System.out.println("main end");
        }
    }

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

`注意：`方法setExecuteExistingDelayedTasksAfterShutdownPolicy可以与 schedule()和 shutdown()方法联合使用,但 `setExecuteExistingDelayedTasksAfterShutdownPolicy()方法不能与 scheduleatFixedRate()和 scheduleWithFixedDelay()方法联合使用。`

那么如果想实现 shutdown关闭线程池后,池中的任务还会继续重复运行,则要将 scheduleatFixedRate和 scheduleWithFixedDelayo方法与 setContinueExistingPeriodicTasksAfterShutdownPolicy()方法联合使用。

#### setContinueExistingPeriodicTasksAfterShutdownPolicy() ####

方法 setContinueExistingPeriodicTasksAfterShutdownPolicy()传入tue的作用是当使用 scheduleAtFixedRate()方法或 scheduleWithFixedDelay()方法时,如果调用 ScheduledThreadPoolExecutor对象的 shutdown()方法,任务还会继续运行,传入false时任务不运行,进程销毁。

public class MyRun8 { 
        public static void main(String[] args) { 
            Runnable runnable = new Runnable() { 
                @Override
                public void run() { 
                    try { 
                        Thread.sleep(2000);
                        System.out.println("任务执行！");
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    }
                }
            };
    
            ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(1);
            executor.setContinueExistingPeriodicTasksAfterShutdownPolicy(false);
            executor.scheduleWithFixedDelay(runnable, 0,3, TimeUnit.SECONDS);
            executor.shutdown();
            System.out.println("main end");
        }
    }

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

#### cancel(boolean)与setRemoveOnCancelPolicy() ####

*  方法 cancel(boolean)的作用设定是否取消任务。
 *  方法 setRemove On CancelPolicy (boolean)的作用设定是否将取消后的任务从队列中清除。

public class MyRun9 { 
        public static void main(String[] args) { 
            ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(10);
            Runnable runnable1 = new Runnable_B("A");
            ScheduledFuture<?> future = executor.schedule(runnable1, 1, TimeUnit.SECONDS);
            System.out.println(future.cancel(true));
            System.out.println("");
            BlockingQueue<Runnable> queue = executor.getQueue();
            Iterator<Runnable> it = queue.iterator();
            while (it.hasNext()) { 
                Runnable next = it.next();
                System.out.println("队列中的：" + next);
            }
            System.out.println("main end");
        }
    }
    
    class Runnable_B implements Runnable { 
    
        private String username;
    
        public Runnable_B(String username) { 
            this.username = username;
        }
    
        public String getUsername() { 
            return username;
        }
    
        public void setUsername(String username) { 
            this.username = username;
        }
    
        @Override
        public void run() { 
            try { 
                while (true) { 
                    if (Thread.currentThread().isInterrupted() == true) { 
                        throw new InterruptedException();
                    }
                    System.out.println("run! username=" + username + " " + Thread.currentThread().getName());
                    Thread.sleep(1000);
                }
            } catch (InterruptedException e) { 
                e.printStackTrace();
                System.out.println("中断了！");
            }
        }
    }

删除成功，但是还保留在队列中。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 10]

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 11]  
删除成功，从队列中剔除。  
![在这里插入图片描述][20200624203231870.png]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70]: /images/20210813/3daa17d429b0484eb650dfed5effdd90.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 1]: /images/20210813/e39f94bb516a4593a4d511507f84bd72.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 2]: /images/20210813/6466739190e84575bb1a341b377308e0.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 3]: /images/20210813/ac9632035c1248b488583fca9ed6fb45.png
[20200624182132951.png]: /images/20210813/87a7b9bf98fc4959891d3536feba151d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 4]: /images/20210813/dfdc4633f1ec44cd857b7fb9d6e281ae.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 5]: /images/20210813/dac8d22105fa49c8bf190e8b66949808.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 6]: /images/20210813/336f7c42a68f482c955c887b10fb3788.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 7]: /images/20210813/a6d5e10fd6e24d9dbcc5ddc5a74a9cdf.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 8]: /images/20210813/44670022d431485c8d78550f54d19d5b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 9]: /images/20210813/391e1fa669044ed1a18352e71cc2e67f.png
[20200624200532453.png]: /images/20210813/f4c52482e32f48668ca353445d21a0b3.png
[20200624201909215.png]: /images/20210813/3dcd47321ebf48739e5fb5af9fd6dffb.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 10]: /images/20210813/31ae9cc8942a4a67b8781576e7f24277.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 11]: /images/20210813/950809e082834cb1b92e135348ce03f6.png
[20200624203231870.png]: /images/20210813/ab7cb9412779474492f2f8319b23489a.png