ScheduledExecutorService延时线程池的简单使用

Posted 2021-12-29 默辨

文章目录

• • 一、Timer
  • • 1、代码测试
    • 2、总结
  • 二、ScheduledExecutorService
  • • 1、简单使用
    • 2、源码分析

一、Timer

在java.util包下有一个Timer类，用于实现定时任务

1、代码测试

代码实现步骤：

1、创建一个timer对象

2、调用timer对象的schedule多态方法，根据传入参数的不同，选择以何种方式执行任务

• 第一个测试的是只执行一次的任务
• 第二个测试的是周期性执行的任务

public class TimerTest 
	private static Logger logger = LoggerFactory.getLogger(TimerTest.class);
	public static void main(String[] args) 
		Timer timer = new Timer();
		timer.schedule(new TimerTask() 
			@Override
			public void run() 
				logger.info("线程启动2s后执行");
			
		, 2000);

		timer.scheduleAtFixedRate(new TimerTask() 
			@Override
			public void run() 
				logger.info("项目启动1s后执行，执行周期为2s");
			
		, 1000, 2000);
	

测试结果：

2、总结

在我们实例化Timer类的对象的时候，他会去创建一个线程，用于后期任务的执行。这是使用Timer类执行定时任务的一大特性。如果我们的场景出现了一个线程出现异常，但是其他任务还能继续执行其他任务，那么使用Timer必然是不符合要求的。

在JDK1.5之后，Doug Lea在Java并发包下引入了ScheduledExecutorService，其拥有线程池的复用线程的特性，支持多线程，其中一个线程挂掉，也不会影响其他线程运行（出现异常的线程会被线程池的waiters队列丢弃，即变成一个null任务的线程）。具体线程池是如何丢弃的细节请参考：浅谈ThreadPoolExecutor线程池底层源码

---

二、ScheduledExecutorService

根据类图可以得出ScheduledExecutorService是ThreadPoolExecutor的子类，其拥有线程池的特性

1、简单使用

public class ScheduledThreadPoolTest 
	private static Logger logger = LoggerFactory.getLogger(ScheduledThreadPoolTest.class);

	public static void main(String[] args) throws ExecutionException, InterruptedException 

		ScheduledExecutorService scheduled = new ScheduledThreadPoolExecutor(1);
		// schedule方法之callable接口
		ScheduledFuture<String> ret = scheduled.schedule(new Callable<String>() 
			@Override
			public String call() throws Exception 
				return "callable接口返回";
			
		, 1, TimeUnit.SECONDS);
		logger.info(ret.get());


		// schedule方法之runable接口
		scheduled.schedule(new Runnable() 
			@Override
			public void run() 
				logger.info("线程启动后1s再执行");

			
		, 1, TimeUnit.SECONDS);

		
		// 周期性执行，来不及消费的会存在队列中
		scheduled.scheduleAtFixedRate(() -> 
			// 每2s产生一个任务，如果任务来不及消费，会存储在对应的队列中
			logger.info("在线程启动的1s后执行，且随后周期执行的时间为2s");
			try 
				// 线程执行时间为3s：3<2，所以任务执行完了马上进行下一个任务
				TimeUnit.SECONDS.sleep(3);
			 catch (InterruptedException e) 
				e.printStackTrace();
			
		, 1, 2, TimeUnit.SECONDS);

		// 周期性执行，来不及消费的阻塞，等消费完成
		scheduled.scheduleWithFixedDelay(() -> 
			try 
				// 任务间隔时间为5s：业务时间+周期时间
				TimeUnit.SECONDS.sleep(3);
			 catch (InterruptedException e) 
				e.printStackTrace();
			
			logger.info("线程启动后1s再执行，且随后的执行周期为2s。注意：这个2s是在前一个任务执行完成之后的2s，并不会堆积到队列中");
		, 1, 2, TimeUnit.SECONDS);
	

四个测试案例分别对应ScheduledExecutorService接口中的4个API。对应的效果写在了上面代码的注释即log中

2、源码分析

和线程池的分析顺序大致相同

1、以scheduleAtFixedRate方法为入口（其余三个方法的差不多，仅仅是不同的方法设置了不同的参数，用于后期条件判断）。注意这里的sft.outerTask = t，后文会再次出现

2、调用对应的delayedExecute方法

3、将任务添加到队列中的步骤已经完成，接下来就是当符合了延时的条件，延时队列就会弹出对应的线程任务，调用对应的run方法

4、调用run方法

1. 判定是周期性的还是一次性的：这里的判断结果根据我们是调用何种API（接口中的四个方法）决定的
2. 设置下次执行时间：周期性任务的有两个API，一个是scheduleAtFixedRate，一个是scheduleWithFixedDelay，两个方法传入的值都是正数，但是后面一个方法会将传入的值封装为一个负数（前面的方法值不变），这里的值就是根据这个封装的结果来决定的
3. 将下次需要执行的任务添加到队列中：下一次需要执行的任务就是第一步设置的sft.outerTask = t，即把自己再次添加到队列中

5、周期性任务的下一次执行操作，

补充：延时线程池的实现其实也不难理解，内部维护了DelayQueue队列，而DelayQueue队列内部又采用优先队列 PriorityQueue，PriorityQueue其内部实现是基于二叉堆的数据结构（对并发包下常见的队列不太熟悉的可以参考我之前的博客：七个常见队列的简单学习，这些队列的概念一定要理解清楚，这对你你整个并发知识的学习至关重要）。加之DelayQueue内部的元素可以基于Comparable 接口，按照规则进行排序，即达到了指定顺序的出队列的效果。

// 延时队列的构造方法
public ScheduledThreadPoolExecutor(int corePoolSize) 
    super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
          new DelayedWorkQueue());

目前有很多的任务调度的组件，除了上面说到的Timer和ScheduledExecutorService，还有Spring Task、Quartz、xxl-job、Elastic-job等等。不过我认为学好JDK自带的组件是研究其他组件的基础！！！