java并发编程之计划任务ScheduledExecutorService的使用

63 阅读 0 评论 42 点赞

我是靠谱客的博主健康长颈鹿，最近开发中收集的这篇文章主要介绍java并发编程之计划任务ScheduledExecutorService的使用，觉得挺不错的，现在分享给大家，希望可以做个参考。

概述

java中的计划任务Timer工具类提供了以计时器或计划任务的功能来实现按指定时间或时间间隔
执行任务，但由于Timer工具类并不是以池pool,而是以队列的方法来管理线程的，所以在高并发的情况下运行效率较低，在新版JDK中提供了ScheduledExecutorService对象来解决效率与定时任务的功能。

8.1 ScheduledExecutorService的使用

类ScheduledExecutorService的主要作用就是可以将定时任务与线程池功能结合使用。
ScheduledThreadPoolExecutor的父接口还是Executor
类ScheduledThreadPoolExecutor的父类是ThreadPoolExecutor

8.2 ScheduledThreadPoolExecutor使用Callable延迟运行

本示例使用Callable接口进行任务延迟运行的实验，具有返回值的功能

public class MyCallableA implements Callable<String> {
@Override
public String call() throws Exception {
System.out.println("callA begin"+Thread.currentThread().getName()+" "+System.currentTimeMillis());
Thread.sleep(3000);
System.out.println("callA end"+Thread.currentThread().getName()+" "+System.currentTimeMillis());
return "A线程";
}
}
................................
public class MyCallableB 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 "B线程";
}
}
.............................
public class Test {
public static void main(String[] args) throws InterruptedException, ExecutionException {
List<Callable> callableList = new ArrayList();
callableList.add(new MyCallableA());
callableList.add(new MyCallableB());
//调用方法newSingleThreadScheduledExecutor()
//取得一个单任务的计划任务池
ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor();
ScheduledFuture<String> futureA = executor.schedule(callableList.get(0), 4L,TimeUnit.SECONDS);//任务4秒后执行
ScheduledFuture<String> futureB = executor.schedule(callableList.get(1),4L,TimeUnit.SECONDS );//任务4秒后执行
System.out.println("开始"+System.currentTimeMillis());
System.out.println("返回值A:"+futureA.get());
System.out.println("返回值B:"+futureB.get());
System.out.println("结束"+System.currentTimeMillis());
}
}
运行结果：
开始1556335280921
callA beginpool-1-thread-1 1556335284921
callA endpool-1-thread-1 1556335287921
返回值A:A线程
callB beginpool-1-thread-1 1556335287921
callB endpool-1-thread-1 1556335287922
返回值B:B线程
结束1556335287922

从开始到结束的运行时间为7秒，阻塞点是get()方法。
从运行结果上看：
public <V>ScheduleFuture<V>schedule(<V>long delay,TimeUnit unit)方法中的第二个参数在多个任务中同时消耗时间，并不是一个任务执行完毕后在等待4秒继续执行的效果。由于第1个任务从计划任务到运行结束需要用时7秒，那么第二个任务其实是想在第4秒后执行，由于是单任务的计划任务池，所以第二个任务的执行时间被延后3秒。

在此实验中使用工厂类Executors的newSingleThreadScheduledExecutor()方法来创建ScheduleExecutorService对象，但返回的真正对象确实ScheduledThreadPoolExecutor,因为ScheduledThreadPoolExecutor实现了ScheduledExecutorService接口。

方法newSingleThreadScheduleExecutor()在JDK中的源代码如下：

public static ScheduledExecutorService newSingleThreadScheduledExecutor() {
return new DelegatedScheduledExecutorService
(new ScheduledThreadPoolExecutor(1));
}

从源代码中还可以发现，调用newSingleThreadScheduledExecutor()方法时，在源代码中实例化new ScheduledThreadPoolExecutor(1)对象时传入参数为1，是单任务执行的计划任务池。

public class Test {
public static void main(String[] args) throws InterruptedException, ExecutionException {
List<Callable> callableList = new ArrayList();
callableList.add(new MyCallableA());
callableList.add(new MyCallableB());
//调用方法newScheduledThreadPool(corePoolSize)
//取得一个同时运行corePoolSize任务个数的计划任务执行池ScheduledExecutorService
ScheduledExecutorService executor = Executors.newScheduledThreadPool(2);
ScheduledFuture<String> futureA = executor.schedule(callableList.get(0), 4L,TimeUnit.SECONDS);
ScheduledFuture<String> futureB = executor.schedule(callableList.get(1),4L,TimeUnit.SECONDS );
System.out.println("开始"+System.currentTimeMillis());
System.out.println("返回值A:"+futureA.get());
System.out.println("返回值B:"+futureB.get());
System.out.println("结束"+System.currentTimeMillis());
}
}
运行结果：
开始1556336913929
callA beginpool-1-thread-1 1556336917931
callB beginpool-1-thread-2 1556336917931
callB endpool-1-thread-2 1556336917931
callA endpool-1-thread-1 1556336920931
返回值A:A线程
返回值B:B线程
结束1556336920931

从结果看出，任务A,B都是4秒后执行，A花了3秒执行完毕。

newScheduledThreadPool源码：
public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
return new ScheduledThreadPoolExecutor(corePoolSize);
}

8.3 ScheduledThreadPoolExecutor使用Runnable延迟运行

除了没返回值，其他效果同Callable一样

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

此示例测试的是执行任务的时间大于>period预定的周期时间，也就是产生了超时效果。


方法scheduleAtFixedRate()
public ScheduledFuture<?> scheduleAtFixedRate(Runnable command,
long initialDelay,
long period,
TimeUnit unit);
public class Test {
public static void main(String[] args) throws InterruptedException, ExecutionException {
List<Callable> callableList = new ArrayList();
//	callableList.add(new MyCallableA());
//	callableList.add(new MyCallableB());
//	//调用方法newScheduledThreadPool(corePoolSize)
//	//取得一个同时运行corePoolSize任务个数的计划任务执行池ScheduledExecutorService
//	ScheduledExecutorService executor = Executors.newScheduledThreadPool(2);
//	ScheduledFuture<String> futureA = executor.schedule(callableList.get(0), 4L,TimeUnit.SECONDS);
//	ScheduledFuture<String> futureB = executor.schedule(callableList.get(1),4L,TimeUnit.SECONDS );
//	System.out.println("开始"+System.currentTimeMillis());
//	System.out.println("返回值A:"+futureA.get());
//	System.out.println("返回值B:"+futureB.get());
//	System.out.println("结束"+System.currentTimeMillis());
ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor();
System.out.println("X="+System.currentTimeMillis());
executor.scheduleAtFixedRate(new MyRunnable(), 4, 2, TimeUnit.SECONDS);//四秒后开始执行，每两秒开始执行
System.out.println("Y="+System.currentTimeMillis());
}
}
运行结果：
X=1556338461067
Y=1556338461069
begin=1556338465071ThreadName=pool-1-thread-1
end = 1556338469071ThreadName=pool-1-thread-1
begin=1556338469071ThreadName=pool-1-thread-1
end = 1556338473072ThreadName=pool-1-thread-1
begin=1556338473072ThreadName=pool-1-thread-1
end = 1556338477072ThreadName=pool-1-thread-1
begin=1556338477072ThreadName=pool-1-thread-1
end = 1556338481072ThreadName=pool-1-thread-1
begin=1556338481072ThreadName=pool-1-thread-1
end = 1556338485072ThreadName=pool-1-thread-1
begin=1556338485072ThreadName=pool-1-thread-1
。。。。。。。。。。。。。。。。。

上面运行结果，虽然让每两秒执行一次，但任务时间是4秒，所以，所有任务都延迟了。

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

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

方法scheduleWithFixedDelay()的主要作用是设置多个任务之间固定的运行时间间隔。
第一个实验执行任务时间大于 period预定的时间

public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command,
long initialDelay,
long delay,
TimeUnit unit);
方法scheduleWithFixedDelay()并没有超时与非超时的情况，
参数 long
delay的主要作用就是下一个任务的开始时间与上一个任务的结束时间的时间间隔

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

public class MyRunnable implements Runnable{
private String username;
public MyRunnable(String username) {
super();
this.username = username;
}
public String getUsername() {
return username;
}
public void setUsername(String username) {
this.username = username;
}
@Override
public void run() {
System.out.println("run!username="+" "+Thread.currentThread().getName()+":"+System.currentTimeMillis());
}
}
..............................
public class Test {
public static void main(String[] args) throws InterruptedException, ExecutionException {
ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(10);
Runnable runnable1 = new MyRunnable("A");
Runnable runnable2 = new MyRunnable("B");
Runnable runnable3 = new MyRunnable("C");
Runnable runnable4 = new MyRunnable("D");
Runnable runnable5 = new MyRunnable("E");
System.out.println(runnable1.hashCode());
System.out.println(runnable2.hashCode());
System.out.println(runnable3.hashCode());
System.out.println(runnable4.hashCode());
System.out.println(runnable5.hashCode());
executor.scheduleAtFixedRate(runnable1, 10, 2, TimeUnit.SECONDS);
executor.scheduleAtFixedRate(runnable2, 10, 2, TimeUnit.SECONDS);
executor.scheduleAtFixedRate(runnable3, 10, 2, TimeUnit.SECONDS);
executor.scheduleAtFixedRate(runnable4, 10, 2, TimeUnit.SECONDS);
executor.scheduleAtFixedRate(runnable5, 10, 2, TimeUnit.SECONDS);
System.out.println();
BlockingQueue<Runnable> queue = executor.getQueue();
Iterator<Runnable> iterator = queue.iterator();
while(iterator.hasNext()) {
Runnable runnable = (Runnable)iterator.next();
System.out.println("队列中的"+runnable);
}
}
}
..............................................
运行结果：
455896770
1323165413
1880587981
511754216
1721931908
队列中的java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask@1f17ae12[Not completed, task = java.util.concurrent.Executors$RunnableAdapter@300ffa5d[Wrapped task = cn.yu.scheduledexecutorservice.MyRunnable@1b2c6ec2]]
队列中的java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask@6193b845[Not completed, task = java.util.concurrent.Executors$RunnableAdapter@4d405ef7[Wrapped task = cn.yu.scheduledexecutorservice.MyRunnable@4edde6e5]]
队列中的java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask@c4437c4[Not completed, task = java.util.concurrent.Executors$RunnableAdapter@2e817b38[Wrapped task = cn.yu.scheduledexecutorservice.MyRunnable@70177ecd]]
队列中的java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask@3f91beef[Not completed, task = java.util.concurrent.Executors$RunnableAdapter@433c675d[Wrapped task = cn.yu.scheduledexecutorservice.MyRunnable@1e80bfe8]]
队列中的java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask@37bba400[Not completed, task = java.util.concurrent.Executors$RunnableAdapter@1a6c5a9e[Wrapped task = cn.yu.scheduledexecutorservice.MyRunnable@66a29884]]
run!username= pool-1-thread-1:1556344657608
run!username= pool-1-thread-3:1556344657608
run!username= pool-1-thread-2:1556344657608
run!username= pool-1-thread-3:1556344657609
run!username= pool-1-thread-2:1556344657609
run!username= pool-1-thread-4:1556344659607
run!username= pool-1-thread-5:1556344659607
run!username= pool-1-thread-1:1556344659607
run!username= pool-1-thread-3:1556344659610

再看remove()方法

public class Test {
public static void main(String[] args) throws InterruptedException, ExecutionException {
ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(1);
Runnable runnable1 = new MyRunnable("A");
Runnable runnable2 = new MyRunnable("B");
ScheduledFuture future1 = executor.scheduleAtFixedRate(runnable1, 10, 2, TimeUnit.SECONDS);
Thread.sleep(1000);
ScheduledFuture future2 =executor.scheduleAtFixedRate(runnable2, 10, 2, TimeUnit.SECONDS);
Thread.sleep(5000);
//注意：remove()方法的参数是ScheduleFuture数据类型
System.out.println(executor.remove((Runnable)future2));
BlockingQueue<Runnable> queue = executor.getQueue();
Iterator<Runnable> iterator = queue.iterator();
while(iterator.hasNext()) {
Runnable runnable = (Runnable)iterator.next();
System.out.println("队列中的"+runnable);
}
}
}
运行结果：
true
队列中的java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask@49097b5d[Not completed, task = java.util.concurrent.Executors$RunnableAdapter@4783da3f[Wrapped task = cn.yu.scheduledexecutorservice.MyRunnable@378fd1ac]]
run!username=A pool-1-thread-1:1556345263396
run!username=A pool-1-thread-1:1556345265395
run!username=A pool-1-thread-1:1556345267395
run!username=A pool-1-thread-1:1556345269395
run!username=A pool-1-thread-1:1556345271395
run!username=A pool-1-thread-1:1556345273396
run!username=A pool-1-thread-1:1556345275396
run!username=A pool-1-thread-1:1556345277396
run!username=A pool-1-thread-1:1556345279395
run!username=A pool-1-thread-1:1556345281395
run!username=A pool-1-thread-1:1556345283395
run!username=A pool-1-thread-1:1556345285395

8.7 方法setExecuteExistingDelayedTasksAfterShutdownPolicy()

方法setExecuteExistingDelayedTasksAfterShutdownPolicy(）的作用是当对ScheduledThreadPoolExecutor执行了shutdown()方法时，任务是否继续运行，默认值是true,也就是当调用了shutdown()方法时任务还是继续运行，当使用setExecuteExistingDelayedTasksAfterShutdownPolicy(false)时任务不再运行。

方法setExecuteExistingDelayedTasksAfterShutdownPolicy()可以与schedule()与shutdown()方法联合使用，但setExecuteExistingDelayedTasksAfterShutdownPolicy()方法不能与scheduleAtFixedRate(）和scheduleWithFixedDelay(）方法联合使用。那么如果想要实现shutdown关闭线程池后，池中的任务还会继续重复运行，则要将scheduleAtFixedRate()和scheduleWithFixedDelay()方法与setContinueExistingPeriodicTasksAfterShutdownPolicy()

8.8 方法setContinueExistingPeriodicTasksAfterShutdownPolicy()

方法setContinueExistingPeriodicTasksAfterShutdownPolicy(）传入true的作用是当使用scheduleAtFixedRate()方法或scheduleWithFixedDelay()方法时，如果调用ScheduledThreadPoolExecutor对象的shutdown()方法，任务还会继续运行，传入false时任务不运行，进程销毁。
8.9 使用cancel(boolean)与setRemoveOnCancelPolicy()方法
方法cancel(boolean) 的作用设定是是否取消任务
（当执行cancel（）方法后任务虽然被成功取消，但还是在队列中存在。再队列中的任务被取消后，任务也不再运行。）

ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(10);
//将任务放入队列中1s后执行
ScheduledFutrue future = executor.schedule(runnable,1,TimeUnit,SECONDS);
//不将任务放入队列中，直接执行
ScheduledFutrue future = executor.schedule(runnable,0,TimeUnit,SECONDS);
正在执行的任务，调用了cancel()方法结合if(Thread.currentThread().isInterrupted() == true)
判断时，是可以让任务结束
总结：
1.在队列中的任务可以取消，任务也不再执行
2.正在运行的任务可以停在，但要结合if（Thread.currentThread().isInterrupted() == true）判断

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

ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(10);
ScheduledFuture future = executor.schedule(runnable,1,TimeUnit.SECONDS);
executor.setRemoveOnCancelPolicy(true);
executor.cancel(true);
setRemoveOnCancelPolicy()与cancel()方法结合，队列中的任务被删除。