SpringBoot之任务调度

任务调度的应用非常广泛,Java提供的解决方案也非常多,例如SpringBoot内置的Schedule、老一辈的任务调度框架Quartz、许雪里写的分布式调度框架xxl-job、shedlock。使用这些框架都离不开cron表达式,cron表达式可以根据表达式的内容指定任务调度的频次与执行时机,例如每隔一秒执行一次任务、每周五执行一次任务等等。

cron位数描述:

cron表示占位符描述:

cron例子:

1.使用SpringBoot内置的Schedule

     <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.4.2</version>
    </parent>
    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
    </dependencies>

1.1 使用Schedule

使用SpringBoot任务调度很简单,只需两步,第一在SpringBoot启动类上添加@EnableScheduling注解开启任务调度,第二在对应的调度任务上添加@Scheduled注解配置任务调度信息。

(1).在SpringBoot启动类上添加@EnableScheduling注解开启任务调度

package com.fly;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.scheduling.annotation.EnableScheduling;

@SpringBootApplication
@EnableScheduling //开启任务调度
public class AppStart {
    public static void main(String[] args) {
        SpringApplication.run(AppStart.class);
    }
}

(2).在对应的调度任务上添加@Scheduled注解配置任务调度信息。

package com.fly.schedule;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.util.Date;


@Component
public class ScheduleTask {
    private Logger logger= LoggerFactory.getLogger(ScheduleTask.class);

    /**
     * @Scheduled注解主要又以下几个重要的属性:
     * cron:根据corn表达式进行任务调度。
     * fixedDelay:每隔多少毫秒执行一次,必须是上一次任务调度成功后才会执行。
     * fixedRate:每隔多少毫秒执行一次,无论上次任务调度是否执行成功,下次都会执行。
     * initialDelay:表示初始化延迟多少毫秒后开始调度。initialDelay必须搭配fixedDelay或fixedRate使用,不然报错
     */
    //每隔3秒调度一次,且上次调度是成功后才会执行
    @Scheduled(fixedDelay = 3000)
    public void taskOne(){
      logger.info("任务1:"+new Date());
      //故意抛出错误
      System.out.println(1/0);
    }

    //每隔3秒调度一次,无论上次是否调度成功
    @Scheduled(fixedRate = 3000)
    public void taskTwo(){
        logger.info("任务2:"+new Date());
        //故意抛出错误
        System.out.println(1/0);
    }

    @Scheduled(fixedRate =3000, initialDelay = 5000)
    public void taskThree(){
        logger.info("任务3:"+new Date());
    }

    //每隔1秒执行一次
    @Scheduled(cron = "0/1 * * * * ?")
    public void taskFour(){
        logger.info("任务4:"+new Date());
    }
}

1.2 并行调度任务

上面所有任务都是在同一个线程调度的(scheduling-1线程),这样会导致当一个线程挂掉之后,会导致线程阻塞从而影响其他任务调度。若想解决此问题只需加入如下配置类即可实现多线程并行,即使某一个线程挂掉之后,也不会阻塞导致其他任务调度。

package com.fly.config;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.SchedulingConfigurer;
import org.springframework.scheduling.config.ScheduledTaskRegistrar;
import java.util.concurrent.Executors;

@Configuration
public class ScheduleConfig implements SchedulingConfigurer {
    @Override
    public void configureTasks(ScheduledTaskRegistrar scheduledTaskRegistrar) {
        scheduledTaskRegistrar.setScheduler(Executors.newScheduledThreadPool(5));
    }
}

1.3 搭配异步注解并发执行

@Async是SpringBoot提供的异步注解,使用方法就两步,在类上添加@EnableAsync注解启用异步注解,在对应的方法上添加@Async注解表示该方法是异步并行执行的。也可以将@EnableAsync注解添加到Springboot启动类上,避免多个地方添加@EnableAsync注解。

package com.fly.schedule;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.util.Date;


@Component
@EnableAsync
public class ScheduleTask02 {
    private Logger logger= LoggerFactory.getLogger(ScheduleTask02.class);

    /**
     * @Scheduled注解主要又以下几个重要的属性:
     * cron:根据corn表达式进行任务调度。
     * fixedDelay:每隔多少毫秒执行一次。fixedRate下一次执行时间是本次开始时间加间隔时间;而fixedDelay下一次执行时间是本次结束时间加间隔时间
     * fixedRate:每隔多少毫秒执行一次,无论上次任务调度是否执行成功,下次都会执行。
     * initialDelay:表示初始化延迟多少毫秒后开始调度。initialDelay必须搭配fixedDelay或fixedRate使用,不然报错
     */

    /**
     * 每隔3秒调度一次。fixedDelay的下次执行时间是本次任务调度结束加间隔时间,例如taskOne中线程休眠了5秒,
     * taskOne下次调度时间是5s+3s=8s,除了首次调度外其他调度都是每8s调度一次
     * @throws InterruptedException
     */
    @Scheduled(fixedDelay = 3000)
    @Async
    public void taskOne() throws InterruptedException {
      logger.info("任务1:"+new Date());
      Thread.sleep(5000);
    }

    //每隔3秒调度一次,即使taskTwo()休眠了5秒,依旧还是每隔3秒调度一次
    @Scheduled(fixedRate = 3000)
    @Async
    public void taskTwo() throws InterruptedException {
        logger.info("任务2:"+new Date());
        Thread.sleep(5000);
    }

    @Scheduled(fixedRate =3000, initialDelay = 5000)
    @Async
    public void taskThree(){
        logger.info("任务3:"+new Date());
    }

    //每隔1秒调度一次
    @Scheduled(cron = "0/1 * * * * ?")
    @Async
    public void taskFour(){
        logger.info("任务4:"+new Date());
    }

}

1.4 动态任务

前面几个例子都是把任务的调度频次写死的,为了能动态的构建任务调度频次,我们需要将调度频次信息存储起来,动态任务本质上是通过存储介质存储cron表达式,将cron表达式存储在数据库中即可实现动态任务。

数据表设计:

### 动态任务调度表
drop TABLE if EXISTS scheduleTask;
create table scheduleTask(
id int not null primary key auto_increment,
taskName varchar(30) not null comment '任务名称',
type tinyint not null comment '任务类型,0系统任务,1用户任务',
taskClass varchar(100) not null comment '目标执行任务类,由类全限定名组成',
taskMethod VARCHAR(50) not null comment '目标执行任务方法名称',
targetMethodParams varchar(100)  comment '目标执行任务方法参数,用&号分割',
cron VARCHAR(100) not null comment 'cron表达式',
taskDescribe VARCHAR(200) comment '任务描述', 
createTime TIMESTAMP not null comment '创建时间'
);
insert into scheduleTask(taskName,type,taskClass,taskMethod,targetMethodParams,cron,taskDescribe,createTime) 
values('测试任务1',0,'com.fly.schedule.ScheduleTask03','task01',null,'0/1 * * * * ?','测试任务',now());
insert into scheduleTask(taskName,type,taskClass,taskMethod,targetMethodParams,cron,taskDescribe,createTime) 
values('测试任务2',0,'com.fly.schedule.ScheduleTask03','task02','task02&这是测试2','0/1 * * * * ?','测试任务',now());

动态任务调度配置类,该类主要核心逻辑在于动态读取数据库调度任务表信息,使用taskClass字段通过反射机制即可获得要执行的目标任务类,通过taskMethod字段获取要进行任务调度的方法名,targetMethodParams字段用于任务调度方法的参数(目前只支持String类型传参,这是因为数据库存储的字符串类型原因),cron字段用于存储cron表达式控制任务调度的频次。

package com.fly.config;

import com.fly.entity.ScheduleTask;
import com.fly.mapper.ScheduleTaskMapper;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.SchedulingConfigurer;
import org.springframework.scheduling.config.ScheduledTaskRegistrar;
import org.springframework.scheduling.support.CronTrigger;

import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.List;

@Configuration
public class DynamicScheduleTask implements SchedulingConfigurer {

    @Autowired
    ScheduleTaskMapper taskMapper;

    @Override
    public void configureTasks(ScheduledTaskRegistrar taskRegistrar) {
        List<ScheduleTask> scheduleTasks = taskMapper.queryScheduleTask();
        scheduleTasks.stream().forEach(scheduleTask->{

            taskRegistrar.addTriggerTask(()->{
                try {
                    Class<?> aClass = Class.forName(scheduleTask.getTaskClass());
                    Class[] methodParameterTypes = getMethodParameterTypes(aClass.getDeclaredMethods(),scheduleTask.getTaskMethod());
                    Method method = aClass.getMethod(scheduleTask.getTaskMethod(),methodParameterTypes);
                    Object[] args=null;
                    if(scheduleTask.getTargetMethodParams()!=null){
                        args=scheduleTask.getTargetMethodParams().split("&");
                    }
                    method.invoke(aClass.newInstance(),args);

                } catch (ClassNotFoundException | NoSuchMethodException e ) {
                    e.printStackTrace();
                } catch (IllegalAccessException e) {
                    e.printStackTrace();
                } catch (InvocationTargetException e) {
                    e.printStackTrace();
                } catch (InstantiationException e) {
                    e.printStackTrace();
                }
            },(triggerContext)->new CronTrigger(scheduleTask.getCron()).nextExecutionTime(triggerContext));
        });
    }

    public static Class[] getMethodParameterTypes(Method[] methods,String methodName){
        Class<?>[] parameterTypes = null;
        for (Method method : methods) {
            if(method.getName().equals(methodName)){
                parameterTypes = method.getParameterTypes();
            }
        }
        return parameterTypes;
    }
}

测试类:

package com.fly.schedule;

public class ScheduleTask03 {
    public void task01(){
        System.out.println("动态任务01");
    }
    public void task02(String taskName,String des){
        System.out.println("动态任务02,任务名称是:"+taskName);
        System.out.println("动态任务02,des:"+des);
    }
}