Spring 任务调度

2023-08-17 17:31 385阅读 0赞

在做项目过程中，一些耗时长的任务可能需要在后台线程池中运行；典型的如发送邮件等，由于需要调用外部的接口来进行实际的发送操作，如果客户端在提交发送请求后一直等待服务器端发送成功后再返回，就会长时间的占用服务器的一个连接；当这类请求过多时，服务器连接数会不够用，新的连接请求可能无法得到满足，从而导致客户端连接失败。因此这类服务一般需要使用到后台线程池来处理。 在这种情况下，我们可以直接使用concurrent包中的线程池来处理，也可以使用其它的方案如Quartz等组件中的线程池来解决；为适配这些不同的方案，Spring引入了TaskExecutor接口作为顶层接口，并提供了几种不同的实现来满足不同的场景。

### 一、TaskExecutor常见实现类 ###

**（1）ThreadPoolTaskExecutor **  
    它是最经常使用的一个，提供了一些Bean属性用于配置java.util.concurrent.ThreadPoolExecutor并且将其包装到TaskExecutor对象中。如果需要适配java.util.concurrent.Executor，请使用ConcurrentTaskExecutor。  
**（2）SimpleAsyncTaskExecutor**   
    线程不会重用，每次调用时都会重新启动一个新的线程；但它有一个最大同时执行的线程数的限制；支持并发，超过最大并发调用数时，会阻塞，直到释放一个槽为止。  
**（3）SyncTaskExecutor **  
    同步的执行任务，任务的执行是在主线程中，不会启动新的线程来执行提交的任务。主要使用在没有必要使用多线程的情况，如较为简单的测试用例。  
**（4）ConcurrentTaskExecutor **  
    它用于适配java.util.concurrent.Executor， 一般情况下请使用ThreadPoolTaskExecutor，如果hreadPoolTaskExecutor不够灵活时可以考虑采用ConcurrentTaskExecutor。  
**（5）SimpleThreadPoolTaskExecutor **  
    它是Quartz中SimpleThreadPool的一个实现，用于监听Spring生命周期回调事件。它主要使用在需要一个线程池来被Quartz和非Quartz中的对象同时共享使用的情况。  
**（6）WorkManagerTaskExecutor **  
    使用CommonJ WorkManager作为它的支持实现，并且是在Spring上下文中设置CommonJ WorkManager引用的中心便利类。与SimpleThreadPoolTaskExecutor类似，这个类实现了WorkManager接口，因此也可以直接作为WorkManager使用。

### 二、使用TaskExecutor实例 ###

/**
     *	测试TaskExecutor
     */
    public class TaskExecutorExample {
    	/**
    	 *	创建一个线程类，该类主要用于打印信息功能
    	 */
    	private class MessagePrinterTask implements Runnable {
    
    		private String message;
    
    		public MessagePrinterTask(String message) {
    			this.message = message;
    		}
    		
    		@Override
    		public void run() {
    			System.out.println(message);
    		}
    
    	}
    	
    	private TaskExecutor taskExecutor;
    	// 构造器 实例化TaskExecutorExample类需要TaskExecutor实例
    	public TaskExecutorExample(TaskExecutor taskExecutor) {
    		this.taskExecutor = taskExecutor;
    	}
    
    	public void printMessages() {
    		for (int i = 0; i < 25; i++) {
    			// 执行任务
    			taskExecutor.execute(new MessagePrinterTask("Message:" + i));
    		}
    	}
    }

@Test
    	public void test() {
    		ApplicationContext context = new ClassPathXmlApplicationContext("spring.xml");
    		TaskExecutorExample taskExecutorExample = context.getBean("taskExecutorExample", TaskExecutorExample.class);
    	}

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwMjk4MzUx_size_16_color_FFFFFF_t_70][]

### 三、TaskScheduler ###

某些时候我们可能需要在某些固定的时间或者是间隔一定的时间连续执行一些任务，如每天凌晨自动跑一些批次/心跳检测等。Spring通过使用TaskScheduler来完成这些功能。

TaskScheduler用于对Runnable的任务进行调度，它包含有多种触发规则。默认的实现是ThreadPoolTaskScheduler。

public interface TaskScheduler {
    
    	// 提交任务调度请求  Trigger 触发器
    	@Nullable
    	ScheduledFuture<?> schedule(Runnable task, Trigger trigger);
    
        // 提交任务调度请求  注意任务只执行一次，使用startTime指定其启动时间  	
    	ScheduledFuture<?> schedule(Runnable task, Date startTime);
    
    	/* 
            使用fixedRate的方式提交任务调度请求 任务首次启动时间由传入参数指定
            period　两次任务启动时间之间的间隔时间，默认单位是毫秒
         */	
    	ScheduledFuture<?> scheduleAtFixedRate(Runnable task, Date startTime, long period);
    
    	// 使用fixedRate的方式提交任务调度请求 任务首次启动时间未设置，任务池将会尽可能早的启动任务 
    	ScheduledFuture<?> scheduleAtFixedRate(Runnable task, long period);
    
    	/*
         使用fixedDelay的方式提交任务调度请求 任务首次启动时间由传入参数指定
         delay 上一次任务结束时间与下一次任务开始时间的间隔时间，单位默认是毫秒 
        */
    	ScheduledFuture<?> scheduleWithFixedDelay(Runnable task, Date startTime, long delay);
    	
    	// 使用fixedDelay的方式提交任务调度请求 任务首次启动时间未设置，任务池将会尽可能早的启动任务 
    	ScheduledFuture<?> scheduleWithFixedDelay(Runnable task, long delay);
    
    }

**（1）Trigger**

触发器的基本思想是，执行时间可以根据过去的执行结果甚至任意条件来确定。如果这些决定确实考虑了前面执行的结果，那么该信息在TriggerContext中是可用的。

public interface Trigger {
    
    	// 用于计算下一次的执行时间
    	@Nullable
    	Date nextExecutionTime(TriggerContext triggerContext);
    
    }

public interface TriggerContext {
        
    	// 上次任务原本的计划时间
        @Nullable
    	Date lastScheduledExecutionTime();
    
    	// 实际的执行时间
        @Nullable
    	Date lastActualExecutionTime();
    
        // 实际的完成时间
    	@Nullable
    	Date lastCompletionTime();
    
    }

（2）**Trigger的实现类**

![20191001195200846.png][]

**1.CronTrigger**

它通过Cron表达式来生成调度计划。 如：

scheduler.schedule(task, new CronTrigger("0 15 9-17 * * MON-FRI"));

以上表达式表示在工作日的9-17点之间的15分钟执行一次；

1.1 Cron表达式是一个字符串，字符串以5或6个空格隔开，分为6或7个域，每一个域代表一个含义，Cron有如下两种语法格式：

① Seconds Minutes Hours DayofMonth Month DayofWeek Year

② *Seconds Minutes Hours DayofMonth Month DayofWeek*

corn从左到右（用空格隔开）：秒 分 小时 月份中的日期 月份 星期中的日期 年份

1.2 各字段的含义

<table> 
 <tbody> 
  <tr> 
   <td>字段</td> 
   <td>允许值</td> 
   <td>允许的特殊字符</td> 
  </tr> 
  <tr> 
   <td>秒（Seconds）</td> 
   <td>0~59的整数</td> 
   <td>, - * / &nbsp; &nbsp;四个字符</td> 
  </tr> 
  <tr> 
   <td>分（<em>Minutes</em>）</td> 
   <td>0~59的整数</td> 
   <td>, - * / &nbsp; &nbsp;四个字符</td> 
  </tr> 
  <tr> 
   <td>小时（<em>Hours</em>）</td> 
   <td>0~23的整数</td> 
   <td>, - * / &nbsp; &nbsp;四个字符</td> 
  </tr> 
  <tr> 
   <td>日期（<em>DayofMonth</em>）</td> 
   <td>1~31的整数（但是你需要考虑你月的天数）</td> 
   <td>,- * ? / L W C &nbsp; &nbsp; 八个字符</td> 
  </tr> 
  <tr> 
   <td>月份（<em>Month</em>）</td> 
   <td>1~12的整数或者 JAN-DEC</td> 
   <td>, - * / &nbsp; &nbsp;四个字符</td> 
  </tr> 
  <tr> 
   <td>星期（<em>DayofWeek</em>）</td> 
   <td>1~7的整数或者 SUN-SAT （1=SUN）</td> 
   <td>, - * ? / L C # &nbsp; &nbsp; 八个字符</td> 
  </tr> 
  <tr> 
   <td>年(可选，留空)（<em>Year</em>）</td> 
   <td>1970~2099</td> 
   <td>, - * / &nbsp; &nbsp;四个字符</td> 
  </tr> 
 </tbody> 
</table>

1.3  特殊符号

① \*：表示匹配该域的任意值。假如在Minutes域使用\*, 即表示每分钟都会触发事件。

②  ?：只能用在DayofMonth和DayofWeek两个域。它也匹配域的任意值，但实际不会。因为DayofMonth和DayofWeek会相互影响。例如想在每月的20日触发调度，不管20日到底是星期几，则只能使用如下写法： 13 13 15 20 \* ?, 其中最后一位只能用？，而不能使用\*，如果使用\*表示不管星期几都会触发，实际上并不是这样。

③  -：表示范围。例如在Minutes域使用5-20，表示从5分到20分钟每分钟触发一次

④ /：表示起始时间开始触发，然后每隔固定时间触发一次。例如在Minutes域使用5/20,则意味着5分钟触发一次，而25，45等分别触发一次.

⑤ ,：表示列出枚举值。例如：在Minutes域使用5,20，则意味着在5和20分每分钟触发一次。

⑥ L：表示最后，只能出现在DayofWeek和DayofMonth域。如果在DayofWeek域使用5L,意味着在最后的一个星期四触发。

⑦ W:表示有效工作日(周一到周五),只能出现在DayofMonth域，系统将在离指定日期的最近的有效工作日触发事件。例如：在 DayofMonth使用5W，如果5日是星期六，则将在最近的工作日：星期五，即4日触发。如果5日是星期天，则在6日(周一)触发；如果5日在星期一到星期五中的一天，则就在5日触发。另外一点，W的最近寻找不会跨过月份 。

⑧ LW:这两个字符可以连用，表示在某个月最后一个工作日，即最后一个星期五。

⑨ \#:用于确定每个月第几个星期几，只能出现在DayofMonth域。例如在4\#2，表示某月的第二个星期三。

**2.PeriodicTrigger **

用于定期执行的Trigger；它有两种模式：

fixedRate：两次任务开始时间之间间隔指定时长  
    fixedDelay: 上一次任务的结束时间与下一次任务开始时间间隔指定时长  
    可见这两种情况的区别就在于，在决定下一次的执行计划时是否要考虑上次任务在什么时间执行完成。   
    默认情况下PeriodicTrigger使用了fixedDelay模式。

PeriodicTrigger提供以下参数来达成目的：

period: long类型，表示间隔时长，注意在fixedRate与fixedDelay两种模式下的不同含义  
    timeUnit: TimeUnit类型，表示间隔时长的单位，如毫秒等；默认是毫秒  
    initialDelay: long类型，表示启动任务后间隔多长时间开始执行第一次任务  
    fixedRate: boolean类型，表示是否是fixedRate，为True时是fixedRate，否则是fixedDelay，默认为False

**（3）TaskScheduler的实现ThreadPoolTaskScheduler**

包装Java Concurrent中的ScheduledThreadPoolExecutor类，大多数场景下都使用它来进行任务调度。   
    除实现了TaskScheduler接口中的方法外，它还包含了一些对ScheduledThreadPoolExecutor进行操作的接口。

@SuppressWarnings("serial")
    public class ThreadPoolTaskScheduler extends ExecutorConfigurationSupport
    		implements AsyncListenableTaskExecutor, SchedulingTaskExecutor, TaskScheduler {
    
    	// 设置线程池大小，最小为1，默认情况下也为1
    	public void setPoolSize(int poolSize) {
    		Assert.isTrue(poolSize > 0, "'poolSize' must be 1 or higher");
    		this.poolSize = poolSize;
    		if (this.scheduledExecutor instanceof ScheduledThreadPoolExecutor) {
    			((ScheduledThreadPoolExecutor) this.scheduledExecutor).setCorePoolSize(poolSize);
    		}
    	}
    
        // 设置异常处理器
    	public void setErrorHandler(ErrorHandler errorHandler) {
    		this.errorHandler = errorHandler;
    	}
    
    	// 获取ScheduledExecutor，默认是ScheduledThreadPoolExecutor类型
    	public ScheduledThreadPoolExecutor getScheduledThreadPoolExecutor() throws IllegalStateException {
    		Assert.state(this.scheduledExecutor instanceof ScheduledThreadPoolExecutor,
    				"No ScheduledThreadPoolExecutor available");
    		return (ScheduledThreadPoolExecutor) this.scheduledExecutor;
    	}
    
    	// 获取线程池大小
    	public int getPoolSize() {
    		if (this.scheduledExecutor == null) {
    			// Not initialized yet: assume initial pool size.
    			return this.poolSize;
    		}
    		return getScheduledThreadPoolExecutor().getPoolSize();
    	}
    
    	// 获取当前活动的线程数
    	public int getActiveCount() {
    		if (this.scheduledExecutor == null) {
    			// Not initialized yet: assume no active threads.
    			return 0;
    		}
    		return getScheduledThreadPoolExecutor().getActiveCount();
    	}
    
    }

举个例子

/**
     *	TaskScheduler例子
     */
    public class TaskSchedulerExample {
    	
    	public class MessagePrinterTask implements Runnable{
    
    		private int message;
    		
    		@Override
    		public void run() {
    			System.out.println("--------------------"+message++);
    		}
    	}
    	
    	private TaskScheduler  taskScheduler;
    	
    	public void printMessages() {
    		// 定时执行任务	
    		taskScheduler.schedule(new MessagePrinterTask(),new CronTrigger("* * * * * ?"));
    	}
    
    	public TaskScheduler getTaskScheduler() {
    		return taskScheduler;
    	}
    
    	public void setTaskScheduler(TaskScheduler taskScheduler) {
    		this.taskScheduler = taskScheduler;
    	}
    		
    }

@Test
    	public void test() {
    		ApplicationContext context = new ClassPathXmlApplicationContext("spring.xml");
    		TaskSchedulerExample taskSchedulerExample = context.getBean("taskSchedulerExample", TaskSchedulerExample.class);
    		
    		try {//模拟系统持续运行
    			Thread.sleep(10000);
    		} catch (InterruptedException e) {
    			e.printStackTrace();
    		}
    	}

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwMjk4MzUx_size_16_color_FFFFFF_t_70 1][]

**（4）Scheduled注解支持**

**   ** 1.@Scheduled注解属性

**    ① **cron：该参数接收一个`cron表达式`，`cron表达式`是一个字符串，字符串以5或6个空格隔开，分开共6或7个域，每一个域代表一个含义。另外，`cron`属性接收的`cron表达式`支持占位符。eg：

配置文件：

time:
      cron: */5 * * * * *
      interval: 5

每5秒执行一次：

@Scheduled(cron="${time.cron}")
        void testPlaceholder1() {
            System.out.println("Execute at " + System.currentTimeMillis());
        }
    
        @Scheduled(cron="*/${time.interval} * * * * *")
        void testPlaceholder2() {
            System.out.println("Execute at " + System.currentTimeMillis());
        }

② zone：时区，接收一个`java.util.TimeZone#ID`。`cron表达式`会基于该时区解析。默认是一个空字符串，即取服务器所在地的时区。比如我们一般使用的时区`Asia/Shanghai`。该字段我们一般留空。

③ fixedDelay：上一次执行完毕时间点之后多长时间再执行。如：

@Scheduled(fixedDelay = 5000) //上一次执行完毕时间点之后5秒再执行

④ fixedDelayString：与`fixedDelay` 意思相同，只是使用字符串的形式。唯一不同的是支持占位符。如：

@Scheduled(fixedDelayString = "5000") //上一次执行完毕时间点之后5秒再执行

占位符的使用(配置文件中有配置：time.fixedDelay=5000)：

@Scheduled(fixedDelayString = "${time.fixedDelay}")
        void testFixedDelayString() {
            System.out.println("Execute at " + System.currentTimeMillis());
        }

⑤ fixedRate ：上一次开始执行时间点之后多长时间再执行。如：

@Scheduled(fixedRate = 5000) //上一次开始执行时间点之后5秒再执行

⑥ fixedRateString：与`fixedRate` 意思相同，只是使用字符串的形式。唯一不同的是支持占位符。

⑦ initialDelay：第一次延迟多长时间后再执行。如：

@Scheduled(initialDelay=1000, fixedRate=5000) //第一次延迟1秒后执行，之后按fixedRate的规则每5秒执行一次

⑧ initialDelayString：与`initialDelay` 意思相同，只是使用字符串的形式。唯一不同的是支持占位符。

2.启用Scheduled注解支持

必须要使用@EnableScheduling注解来启用对@Scheduled注解的支持，@EnableScheduling必须使用在工程中某一个被Configuration注解的类上，如：

@Configuration
    @EnableScheduling
    public class MainConfiguration {
    }

xml文件启动方法

引入task命名空间

xmlns:task="http://www.springframework.org/schema/task"
    
    http://www.springframework.org/schema/task
    http://www.springframework.org/schema/task/spring-task-4.3.xsd

xml文件配置

<context:component-scan base-package="com.buba.task.*" />
    
    	<task:annotation-driven executor="executor" scheduler="scheduler"/>
    	<task:executor id="executor" pool-size="5"/>
    	<task:scheduler id="scheduler" pool-size="10"/>
    	
    	<bean id="taskSchedulerExample2"
    		class="com.buba.task.TaskSchedulerExample2">
    	</bean>

public class TaskSchedulerExample2 {
    	
    	private int a;
    	
    	@Scheduled(cron="* * * * * ?")
    	public void test() {
    		System.out.println("----------------"+a++);
    	}
    }

@Test
    	public void test() {
    		ApplicationContext context = new ClassPathXmlApplicationContext("spring.xml");
    		TaskSchedulerExample2 bean = context.getBean("taskSchedulerExample2",TaskSchedulerExample2.class);
    		try {//模拟系统持续运行
    			Thread.sleep(10000);
    		} catch (InterruptedException e) {
    			e.printStackTrace();
    		}
    	}

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwMjk4MzUx_size_16_color_FFFFFF_t_70 1][]

**如果本文对您有很大的帮助，还请点赞关注一下。**

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwMjk4MzUx_size_16_color_FFFFFF_t_70]: /images/20230808/96391071928b4e218a41d36d6f05b580.png
[20191001195200846.png]: /images/20230808/8794e64e743b4d89b6c7a1537710fcdb.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwMjk4MzUx_size_16_color_FFFFFF_t_70 1]: /images/20230808/5b4d7f08329f473f8ba4cdfa20fa44ec.png