从前慢现在也慢

这个屌丝很懒，什么也没留下！

热门标签

计划任务ScheduledExecutorService的使用总结

作者：从前慢现在也慢 | 2024-02-15 10:42:02

踩

scheduledexecutorservice

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

1.ScheduledExecutorService的使用

类ScheduledExecutorService的主要作用就是可以将定时任务与线程池功能结合使用。实现类ScheduledThreadPoolExecutor的父接口还是Executor，父类是ThreadPoolExecutor，其部分方法是父类ThreadPoolExecutor提供并在子类ScheduledThreadPoolExecutor中重写的，比如submit()重载方法或shutdown()等方法。

2.ScheduledThreadPoolExecutor使用Callable延迟运行

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


package mycallable;
 
import java.util.concurrent.Callable;
 
public class MyCallableA 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";
	}
}


package mycallable;
 
import java.util.concurrent.Callable;
 
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 "returnB";
	}
}


package test.run;
 
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
 
import mycallable.MyCallableA;
import mycallable.MyCallableB;
 
public class Run1 {
	@SuppressWarnings("unchecked")
	public static void main(String[] args) {
		try {
			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);
			ScheduledFuture<String> futureB = executor.schedule(callableList.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();
		}
	}
}


package test.run;
 
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
 
import mycallable.MyCallableA;
import mycallable.MyCallableB;
 
public class Run2 {
	public static void main(String[] args) {
		try {
			List<Callable> callableList = new ArrayList();
			callableList.add(new MyCallableA());
			callableList.add(new MyCallableB());
			// 调用方法newScheduledThreadPool(corePoolSize >1)
			// 取得一个同时运行corePoolSize任务个数的计划任务执行池
			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("          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();
		}
	}
}


package test.run;
 
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
 
import mycallable.MyCallableA;
import mycallable.MyCallableB;
 
public class Run3 {
	public static void main(String[] args) {
		try {
			List<Callable> callableList = new ArrayList();
			callableList.add(new MyCallableA());
			callableList.add(new MyCallableB());
			// 调用方法newScheduledThreadPool(1)
			// 取得一个单任务的计划任务执行池
			ScheduledExecutorService executor = Executors.newScheduledThreadPool(1);
			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("          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> ScheduledFuture<V> schedule(Callable<V> callable,long delay,TimeUnit unit)方法invokeAll中的第2个参数在多个任务中同时消耗时间，并不是一个任务执行完毕后再等待4秒继续执行的效果。由于第1个任务从计划任务到运行结束需要用时7秒，那么第2个任务其实是想在第4秒被执行，由于是单任务的计划任务池，所以第2个任务的执行时间被延后3秒。即使是newScheduledThreadPool(poolSize)的写法创建出来的任务池也是单任务的。

3.ScheduledThreadPoolExecutor使用Runnable延迟运行

使用Runnable接口进行无返回值的计划任务实例


package mycallable;
 
public class MyRunnableA 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();
		}
	}
}


package mycallable;
 
public class MyRunnableB 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());
	}
}


package test.run;
 
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
 
import mycallable.MyRunnableA;
import mycallable.MyRunnableB;
 
public class Run {
	public static void main(String[] args) {
		List<Runnable> runnableList = new ArrayList();
		runnableList.add(new MyRunnableA());
		runnableList.add(new MyRunnableB());
		ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor();
 
		System.out.println("     X=" + System.currentTimeMillis());
		executor.schedule(runnableList.get(0), 0L, TimeUnit.SECONDS);
		executor.schedule(runnableList.get(1), 0L, TimeUnit.SECONDS);
		System.out.println("     Y=" + System.currentTimeMillis());
 
	}
}

4.延迟运行并取得返回值


package mycallable;
 
import java.util.concurrent.Callable;
 
public class MyCallableA implements Callable<String> {
	public String call() throws Exception {
		System.out.println("a call run =" + System.currentTimeMillis());
		return "returnA";
	}
}


package test.run;
 
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
 
import mycallable.MyCallableA;
 
public class Run {
	public static void main(String[] args) {
		try {
			List<Callable> callableList = new ArrayList();
			callableList.add(new MyCallableA());
			ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor();
			System.out.println("X=" + System.currentTimeMillis());
			ScheduledFuture<String> futureA = executor.schedule(callableList.get(0), 4L, TimeUnit.SECONDS);
			System.out.println(futureA.get() + "   A=" + System.currentTimeMillis());
		} catch (InterruptedException e) {
			e.printStackTrace();
		} catch (ExecutionException e) {
			e.printStackTrace();
		}
	}
}

5.使用scheduleAtFixedRate(Runnable command,long initialDelay,long period,TimeUnit unit)方法实现周期性执行(该方法有任务超时的效果)

5.1 接口scheduleAtFixedRate原型定义及参数说明

command：执行线程

initialDelay：初始化延时

period：两次开始执行最小间隔时间

unit：计时单位


package mycallable;
 
public class MyRunnable implements 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();
		}
	}
}


package test.run;
 
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
 
import mycallable.MyRunnable;
 
public class Run {
	public static void main(String[] args) {
		ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor();
		System.out.println("           X=" + System.currentTimeMillis());
		executor.scheduleAtFixedRate(new MyRunnable(), 1, 2, TimeUnit.SECONDS);
		System.out.println("           Y=" + System.currentTimeMillis());
	}
}

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

6.使用scheduleWithFixedDelay(Runnable command,long initialDelay,long delay,TimeUnit unit)方法实现周期性执行(该方法没有超时与非超时的情况)

command：执行线程

initialDelay：初始化延时

period：前一次执行结束到下一次执行开始的间隔时间（间隔执行延迟时间）

unit：计时单位


package mycallable;
 
public class MyRunnable implements 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();
		}
	}
}


package test.run;
 
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
 
import mycallable.MyRunnable;
 
public class Run {
	public static void main(String[] args) {
		ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor();
		System.out.println("           X=" + System.currentTimeMillis());
		executor.scheduleWithFixedDelay(new MyRunnable(), 1, 2, TimeUnit.SECONDS);
		System.out.println("           Y=" + System.currentTimeMillis());
	}
}

7.使用getQueue()与remove()方法

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


package mycallable;
 
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=" + username + " " + Thread.currentThread().getName());
	}
 
}


package test.run;
 
import java.util.Iterator;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
 
import mycallable.MyRunnable;
 
public class Run1 {
	public static void main(String[] args) {
 
		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);
		}
	}
}


package test.run;
 
import java.util.Iterator;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
 
import mycallable.MyRunnable;
 
public class Run2 {
	public static void main(String[] args) throws InterruptedException {
 
		ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(1);
 
		Runnable runnable1 = new MyRunnable("A");
		Runnable runnable2 = new MyRunnable("B");
 
		ScheduledFuture future1 = executor.scheduleAtFixedRate(runnable1, 0, 2, TimeUnit.SECONDS);
		Thread.sleep(1000);
		ScheduledFuture future2 = executor.scheduleAtFixedRate(runnable2, 10, 2, TimeUnit.SECONDS);
		Thread.sleep(5000);
		System.out.println(executor.remove((Runnable) future2));
		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);
		}
	}
}

8.方法setExecuteExistingDelayedTasksAfterShutdownPolicy()的使用

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


package mycallable;
 
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=" + username + " " + Thread.currentThread().getName() + " "
				+ System.currentTimeMillis());
	}
}


package test.run;
 
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
 
import mycallable.MyRunnable;
 
/**
 * 虽然shutdown()但任务还是执行了
 * @author linhaiy
 *
 */
public class Run1 {
	public static void main(String[] args) throws InterruptedException {
		ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(1);
		Runnable runnable1 = new MyRunnable("A");
		Runnable runnable2 = new MyRunnable("B");
		executor.schedule(runnable1, 3, TimeUnit.SECONDS);
		executor.shutdown();
		System.out.println("已经shutdown了");
	}
}


package test.run;
 
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
 
import mycallable.MyRunnable;
 
/**
 * 任务被取消执行，进程销毁了
 * @author linhaiy
 *
 */
public class Run2 {
	public static void main(String[] args) {
		ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(1);
		Runnable runnable1 = new MyRunnable("A");
		Runnable runnable2 = new MyRunnable("B");
		executor.schedule(runnable1, 3, TimeUnit.SECONDS);
		executor.setExecuteExistingDelayedTasksAfterShutdownPolicy(false);
		executor.shutdown();
		System.out.println("已经shutdown了");
	}
}

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

9.方法setContinueExistingPeriodicTasksAfterShutdownPolicy()

方法setContinueExistingPeriodicTasksAfterShutdownPolicy()传入true的作用是当使用scheduleAtFixedRate()方法或scheduleWithFixedDelay()方法时，如果调用ScheduledThreadPoolExecutor对象的shutdown()方法，任务还会继续运行，传入false时任务不运行，进程销毁。如果使用scheduleAtFixedRate()结合shutdown()方法想实现任务继续运行的效果，则必须使用setContinueExistingPeriodicTasksAfterShutdownPolicy(true)代码。


package mycallable;
 
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=" + username + " " + Thread.currentThread().getName() + " "
				+ System.currentTimeMillis());
	}
}


package test.run;
 
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
 
import mycallable.MyRunnable;
 
public class Run1 {
	public static void main(String[] args) {
		ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(10);
		Runnable runnable1 = new MyRunnable("A");
		executor.scheduleAtFixedRate(runnable1, 1, 2, TimeUnit.SECONDS);
		executor.shutdown();
		System.out.println("执行了shutdown!");
	}
}


package test.run;
 
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
 
import mycallable.MyRunnable;
 
public class Run2 {
	public static void main(String[] args) {
		ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(10);
		Runnable runnable1 = new MyRunnable("A");
		executor.scheduleAtFixedRate(runnable1, 1, 2, TimeUnit.SECONDS);
		executor.setContinueExistingPeriodicTasksAfterShutdownPolicy(true);
		executor.shutdown();
		System.out.println("执行了shutdown!");
	}
}

10.使用cancel(boolean)与setRemoveOnCancelPolicy()方法

10.1 方法cancel(boolean)的作用设定是否取消任务。

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

10.3 当执行cancel()方法后任务虽然被成功取消，但还是在队列中存在。

10.4 在队列中的任务被成功取消，任务也不再运行。

10.5 正在运行的任务可以停止，但要结合if(Thread.currentThread().isInterrupted() == true)判断。

10.6 使用setRemoveOnCancelPolicy(true)代码结合cancel(true),则队列中的任务被删除

总结：在此介绍了基于线程池ThreadPoolExecutor的ScheduledThreadPoolExecutor计划任务执行池对象，使用此类可以高效地实现计划任务线程池，不再重复创建Thread对象，提高了运行效率。此类也支持间隔运行的功能。

声明：本文内容由网友自发贡献，转载请注明出处：【wpsshop博客】