Bootstrap

🍃【SpringBoot技术专题】「开发实战系列」动态化Quartz任务调度机制+实时推送任务数据到前端

2022-03-10 作者: 浩宇天尚

前提介绍

SpringBoot2.0整合quartz实现多定时任务动态配置，实现任务增删改，生成Cron表达式

动态化任务调度

添加依赖包

添加SpringBoot整合Quartz框架的操作启动类



    org.springframework.boot
    spring-boot-starter-quartz

DynamicSchedulerConfig

设置动态Scheduler调度中心的配置服务类

import org.springframework.boot.autoconfigure.quartz.SchedulerFactoryBeanCustomizer;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.quartz.SchedulerFactoryBean;
@Configuration
public class DynamicSchedulerConfig implements SchedulerFactoryBeanCustomizer{
  @Override
  public void customize(SchedulerFactoryBean schedulerFactoryBean) {
          schedulerFactoryBean.setStartupDelay(2); 
          schedulerFactoryBean.setAutoStartup(true);
          schedulerFactoryBean.setOverwriteExistingJobs(true);
        }
}

application.yml配置

server:
    port: 8101
# 默认的profile为dev，其他环境通过指定启动参数使用不同的profile，比如：  
#   测试环境：java -jar quartz-service.jar --spring.profiles.active=test  
#   生产环境：java -jar quartz-service.jar --spring.profiles.active=prod  
spring:
  datasource:
    type: com.alibaba.druid.pool.DruidDataSource   
  #这里是配置druid连接池，以下都是druid的配置信息
    url: jdbc:mysql://127.0.0.1:3306/task?useUnicode=true&characterEncoding=utf-8&useSSL=false
    driver-class-name: com.mysql.jdbc.Driver
    username: root
    password: 123456
#quartz相关属性配置
 quartz:
    properties:
      org:
        quartz:
          scheduler:
            instanceName: clusteredScheduler
            instanceId: AUTO
          jobStore:
            class: org.quartz.impl.jdbcjobstore.JobStoreTX
            driverDelegateClass: org.quartz.impl.jdbcjobstore.StdJDBCDelegate
            tablePrefix: QRTZ_
            isClustered: true
            clusterCheckinInterval: 10000
            useProperties: false
          threadPool:
            class: org.quartz.simpl.SimpleThreadPool
            threadCount: 10
            threadPriority: 5
            threadsInheritContextClassLoaderOfInitializingThread: true
    #数据库方式
    job-store-type: jdbc
# mybatisplus配置
mybatis-plus:
  mapper-locations: classpath*:/mapper/**Mapper.xml
  #把xml文件放在com.XX.mapper.*中可能会出现找到的问题，这里把他放在resource下的mapper中
  typeAliasesPackage: com.task.entity
  #这里是实体类的位置，#实体扫描，多个package用逗号或者分号分隔
  configuration:
    map-underscore-to-camel-case: true
    cache-enabled: false
logging:
  file: task-info.log
  level:
    com.task: debug

TaskInfoService业务服务类

配置相关的业务服务类：

@Service
public class TaskInfoService extends IService {
    /**

    * @Title: getPageJob
    * @Description: TODO(查询定时任务，分页)
    * @param @param search
    * @param @return    参数
    * @return Map    返回类型
    * @throws
    */

    IPage getPageJob(Pageable pageable, MultiValueMap queryParam);

    /**

    * @Title: getPageJobmod
    * @Description: TODO(查询定时任务)
    * @param @return    参数
    * @return TaskInfoBO    返回类型
    * @throws
    */
    TaskInfoBO getPageJobmod();

    /**
    * @Title: addJob
    * @Description: TODO(添加任务)
    * @param @param jobClassName 任务路径名称
    * @param @param jobGroupName 任务分组
    * @param @param cronExpression cron时间规则
    * @param @throws Exception    参数
    * @return void    返回类型
    * @throws
    */
    void addJob(String jobClassName, String jobGroupName, String cronExpression) throws Exception;

    /**
    * @Title: addJob
    * @Description: TODO(添加动态任务)
    * @param @param jobClassName 任务路径名称
    * @param @param jobGroupName 任务分组
    * @param @param cronExpression cron时间规则
    * @param @param jobDescription 参数
    * @param @param params
    * @param @throws Exception  参数说明
    * @return void    返回类型
    * @throws
    */

    void addJob(String jobClassName, String jobGroupName, String cronExpression, String jobDescription, Map params) throws Exception;

    /**
    * @Title: updateJob
    * @Description: TODO(更新定时任务)
    * @param @param jobClassName 任务路径名称
    * @param @param jobGroupName 任务分组
    * @param @param cronExpression cron时间规则
    * @param @throws Exception    参数
    * @return void    返回类型
    * @throws
    */
    void updateJob(String jobClassName, String jobGroupName, String cronExpression) throws Exception;

    /**
    * @Title: deleteJob
    * @Description: TODO(删除定时任务)
    * @param @param jobClassName 任务路径名称
    * @param @param jobGroupName 任务分组
    * @param @throws Exception    参数
    * @return void    返回类型
    * @throws
    */

    void deleteJob(String jobClassName, String jobGroupName) throws Exception;

    /**
    * @Title: pauseJob
    * @Description: TODO(暂停定时任务)
    * @param @param jobClassName 任务路径名称
    * @param @param jobGroupName 任务分组
    * @param @throws Exception    参数
    * @return void    返回类型
    * @throws
    */
    void pauseJob(String jobClassName, String jobGroupName) throws Exception;

    /**
    * @Title: resumejob
    * @Description: TODO(恢复任务)
    * @param @param jobClassName 任务路径名称
    * @param @param jobGroupName 任务分组
    * @param @throws Exception    参数
    * @return void    返回类型
    * @throws
    */
    void resumejob(String jobClassName, String jobGroupName) throws Exception;
}

TaskInfoServiceImpl业务服务类

实现相关任务类服务机制的实现类


@Slf4j
@Service
@Transactional
public class TaskInfoServiceImpl  extends ServiceImpl implements TaskInfoService {

    @Autowired

    private Scheduler scheduler;

    @Override

    public IPage getPageJob(Pageable pageable, MultiValueMap queryParam) {

        IPage page = new Page<>(pageable.getPageNumber(), pageable.getPageSize());

        return baseMapper.getJobAndTriggerDetails(page);

    }

    @Override

    public TaskInfoBO getPageJobmod() {
        return baseMapper.getJobAndTriggerDto();
    }

    @Override

    public void addJob(String jobClassName, String jobGroupName, String cronExpression) throws Exception {
        // 启动调度器
        scheduler.start();
        // 构建job信息
        JobDetail jobDetail = JobBuilder.newJob(getClass(jobClassName).getClass())
                .withIdentity(jobClassName, jobGroupName).build();
        // 表达式调度构建器(即任务执行的时间)
        CronScheduleBuilder scheduleBuilder = CronScheduleBuilder.cronSchedule(cronExpression);
        // 按新的cronExpression表达式构建一个新的trigger
        CronTrigger trigger = TriggerBuilder.newTrigger().withIdentity(jobClassName, jobGroupName)
                .withSchedule(scheduleBuilder).build();
        try {
            scheduler.scheduleJob(jobDetail, trigger);
        } catch (SchedulerException e) {
            throw new Exception("创建定时任务失败");
        }
    }
  
    @Override
    public void addJob(String jobClassName, String jobGroupName, String cronExpression, String jobDescription,
                      Map params) throws Exception {
        // 启动调度器
        scheduler.start();
        // 构建job信息
        JobDetail jobDetail = JobBuilder.newJob(TaskInfoServiceImpl.getClass(jobClassName).getClass())
                .withIdentity(jobClassName, jobGroupName).withDescription(jobDescription).build();
        Iterator> var7 = params.entrySet().iterator();
        while(var7.hasNext()) {
            Map.Entry entry = var7.next();
            jobDetail.getJobDataMap().put((String)entry.getKey(), entry.getValue());
        }
        // 表达式调度构建器(即任务执行的时间)
        CronScheduleBuilder scheduleBuilder = CronScheduleBuilder.cronSchedule(cronExpression);
        // 按新的cronExpression表达式构建一个新的trigger
        CronTrigger trigger = (CronTrigger)TriggerBuilder.newTrigger().withIdentity(jobClassName, jobGroupName)
                .withSchedule(scheduleBuilder).build();
        try {
            scheduler.scheduleJob(jobDetail, trigger);
        } catch (SchedulerException e) {
            throw new Exception("创建定时任务失败");
        }
    }

    @Override

    public void updateJob(String jobClassName, String jobGroupName, String cronExpression) throws Exception {
        try {
            TriggerKey triggerKey = TriggerKey.triggerKey(jobClassName, jobGroupName);
            // 表达式调度构建器（动态修改后不立即执行）
            CronScheduleBuilder scheduleBuilder = CronScheduleBuilder.cronSchedule(cronExpression).withMisfireHandlingInstructionDoNothing();
            CronTrigger trigger = (CronTrigger) scheduler.getTrigger(triggerKey);
            // 按新的cronExpression表达式重新构建trigger
            trigger = trigger.getTriggerBuilder().withIdentity(triggerKey).withSchedule(scheduleBuilder).build();
            // 按新的trigger重新设置job执行
            scheduler.rescheduleJob(triggerKey, trigger);
        } catch (SchedulerException e) {
            throw new Exception("更新定时任务失败");
        }
    }

    @Override

    public void deleteJob(String jobClassName, String jobGroupName) throws Exception {
        scheduler.pauseTrigger(TriggerKey.triggerKey(jobClassName, jobGroupName));
        scheduler.unscheduleJob(TriggerKey.triggerKey(jobClassName, jobGroupName));
        scheduler.deleteJob(JobKey.jobKey(jobClassName, jobGroupName));
    }

    @Override
    public void pauseJob(String jobClassName, String jobGroupName) throws Exception {
        scheduler.pauseJob(JobKey.jobKey(jobClassName, jobGroupName));
    }

    @Override

    public void resumejob(String jobClassName, String jobGroupName) throws Exception {
        scheduler.resumeJob(JobKey.jobKey(jobClassName, jobGroupName));
    }
  
    public static BaseJob getClass(String classname) throws Exception {
        Class class1 = Class.forName(classname);
        return (BaseJob) class1.newInstance();
    }
}

JobAndTriggerMapper

任务服务中心的数据访问层

public interface JobAndTriggerMapper extends BaseMapper {
    IPage getJobAndTriggerDetails(IPage page);
    TaskInfoBO getTaskInfoModel;
}

JobAndTriggerMapperxml

任务服务中心的数据访问层xml配置文件操作处理

BaseJob类

设置配置相关的BaseJob类模型，作为Job与Quartz的核心机制对象的介入口

public interface BaseJob extends Job {
  public void execute(JobExecutionContext context) throws JobExecutionException;
}

TaskController控制器

任务服务调度的控制器，用于页面或者前端进行交互的任务控制器操作执行类

@RestController
@RequestMapping(value = "/job")
public class JobController  {

  @Autowired
    private TaskInfoService taskInfoService;

    public JobController(TaskInfoService taskInfoService){
        this.taskInfoService = taskInfoService;
    }

    @PostMapping(value = "/page")
    public ResponseEntity> queryjob(Pageable pageable, @RequestParam MultiValueMap queryParams, UriComponentsBuilder uriBuilder) {
        IPage page = taskInfoService.getPageJob(pageable, queryParams);
        HttpHeaders headers = PaginationUtil.generatePaginationHttpHeaders(uriBuilder.queryParams(queryParams), page);
        return ResponseEntity.ok().headers(headers).body(page.getRecords());
    }

    /**
    * @Title: addJob
    * @Description: TODO(添加Job)
    * @param jobClassName
    * 类名
    * @param jobGroupName
    * 组名
    * @param cronExpression
    * 表达式，如：0/5 * * * * ? (每隔5秒)
    */
    @PostMapping(value = "/add")
    public ResponseEntity addJob(
            @RequestParam(value = "jobClassName") String jobClassName,
            @RequestParam(value = "jobGroupName") String jobGroupName,
            @RequestParam(value = "cronExpression") String cronExpression){
        try {
            jobAndTriggerService.addJob(jobClassName, jobGroupName, cronExpression);
            return ResponseEntity.ok().body("操作成功");
        } catch (Exception e) {
            return ResponseEntity.ok().body("操作失败");
        }
    }
  
    /**
    * @Title: pauseJob
    * @Description: TODO(暂停Job)
    * @param jobClassName
    *            类名
    * @param jobGroupName
    *            组名
    */

    @PostMapping(value = "/pause")
    public ResponseEntity pauseJob(
            @RequestParam(value = "jobClassName") String jobClassName,
            @RequestParam(value = "jobGroupName") String jobGroupName) {
        try {
            taskInfoService.pauseJob(jobClassName, jobGroupName);
            return ResponseEntity.ok().body("操作成功");
        } catch (Exception e) {
            return ResponseEntity.ok().body("操作失败");
        }
    }
    /**
    * @Title: resumeJob
    * @Description: TODO(恢复Job)
    * @param jobClassName
    *            类名
    * @param jobGroupName
    *            组名
    */
    @PostMapping(value = "/resume")
    public ResponseEntity resumeJob(
            @RequestParam(value = "jobClassName") String jobClassName,
            @RequestParam(value = "jobGroupName") String jobGroupName) {
        try {
            taskInfoService.resumejob(jobClassName, jobGroupName);
            return ResponseEntity.ok().body("操作成功");
        } catch (Exception e) {
            return ResponseEntity.ok().body("操作失败");
        }
    }
    /**
    * @Title: rescheduleJob
    * @Description: TODO(重新设置Job)
    * @param jobClassName
    *            类名
    * @param jobGroupName
    *            组名
    * @param cronExpression
    *            表达式
    */

    @PostMapping(value = "/reschedule")
    public ResponseEntity rescheduleJob(
            @RequestParam(value = "jobClassName") String jobClassName,
            @RequestParam(value = "jobGroupName") String jobGroupName,
            @RequestParam(value = "cronExpression") String cronExpression) {
        try {
            taskInfoService.updateJob(jobClassName, jobGroupName, cronExpression);
            return ResponseEntity.ok().body("操作成功");
        } catch (Exception e) {
            return ResponseEntity.ok().body("操作失败");
        }
    }

    /**
    * @Title: deleteJob
    * @Description: TODO(删除Job)
    * @param jobClassName
    *            类名
    * @param jobGroupName
    *            组名
    */

    @RequestMapping(value = "/del", method = RequestMethod.POST)

    public ResponseEntity deleteJob(@RequestParam(value = "jobClassName") String jobClassName, @RequestParam(value = "jobGroupName") String jobGroupName) {
        try {
            taskInfoService.deleteJob(jobClassName, jobGroupName);
            return ResponseEntity.ok().body("操作成功");
        } catch (Exception e) {
            return ResponseEntity.ok().body("操作失败");
        }
  }
}

下面是关于任务调度列表的图：

DeferredResult实现实时推送

  • 浏览器要实时展示服务端计算出来的数据。一种可能的实现是：浏览器频繁向服务端发起请求以获得服务端数据。

  • 若定时周期为S，则数据延迟周期最大即为S。若想缩短数据延迟周期，则应使S尽量小，而S越小，浏览器向服务端发起请求的频率越高，又造成网络握手次数越多，影响了效率。因此，此场景应使用服务端实时推送技术。

  • 这里说是推送，其实还是基于请求-响应机制，只不过发起的请求会在服务端挂起，直到请求超时或服务端有数据推送时才会做出响应，响应的时机完全由服务端控制。所以，整体效果看起来就像是服务端真的在“实时推送”一样。

可以利用DeferredResult来实现异步长连接的服务端实时推送。

使用案例
@RequestMapping("/call")
@ResponseBody
public DeferredResult call() {
  // 泛型Object表示返回结果的类型
    DeferredResult response = new DeferredResult(10000, // 请求的超时时间
  null); // 超时后响应的结果
    response.onCompletion(new Runnable() {
        @Override
        public void run() {
            // 请求处理完成后所做的一些工作
        }
    });
    // 设置响应结果
    // 调用此方法时立即向浏览器发出响应；未调用时请求被挂起
    response.setResult(new Object());
    return response;
}

执行逻辑
实现DeferResult传输模型
DeferredResult对象模型接口抽象接口
public interface DeferredData {
    String getId(); // 唯一标识
}

DeferredResult的持有者
DeferredResult对象的数据接口容器机制
public interface IDeferredResultHolder {
    DeferredResult newDeferredResult(String key, long timeout, Object timeoutResult);
    void add(String key, DeferredResult deferredResult);
    DeferredResult get(String key);
    void remove(String key);
    void handleDeferredData(DeferredData deferredData);
}

DeferredResult的持有者实现
实现类针对于接口容器机制
public class DeferredResultHolder implements IDeferredResultHolder {

    private Map> deferredResults = new ConcurrentHashMap>();

    public DeferredResult newDeferredResult(String key) {
        return new DeferredResult(key, 30 * 1000L, null);
    }
    public DeferredResult newDeferredResult(String key, long timeout) {
        return new DeferredResult(key, timeout, null);
    }
    public DeferredResult newDeferredResult(String key, Object timeoutResult) {
        return new DeferredResult(key, 30 * 1000L, timeoutResult);
    }
    @Override
    public DeferredResult newDeferredResult(String key, long timeout, Object timeoutResult) {
        DeferredResult deferredResult = newDeferredResult(timeout, timeoutResult);
        add(key, deferredResult);
        deferredResult.onCompletion(new Runnable() {
            @Override
            public void run() {
                remove(key);
            }
        });
        return deferredResult;
    }

    @Override
    public void add(String key, DeferredResult deferredResult) {
        deferredResults.put(key, deferredResult);
    }

    @Override
    public DeferredResult get(String key) {
        return deferredResults.get(key);
    }

    @Override
    public void remove(String key) {
        deferredResults.remove(key);
    }

    @Override
    public void handleDeferredData(DeferredData deferredData) {
        String key = deferredData.getId();
        DeferredResult deferredResult = get(key);
        if (deferredResult != null) {
            deferredResult.setResult(deferredData);
        }
    }
}

调用端
服务页面的调用操作，不会立刻返回，而是被阻塞状态化
@RequestMapping
@Controller
public class CallController {
    @Autowired
    private DeferredResultHolder deferredResultHolder;
    @RequestMapping("/call")
    @ResponseBody
    public DeferredResult call() {
        String id = "abc";
        return deferredResultHolder.newDeferredResult(id, 10 * 1000L, null);
    }
}

触发返回端
完成后，调用此方法，会促使前端进行解除阻塞，并且立刻返回！
@RequestMapping
@Controller
public class CallController {
    @Autowired
    private DeferredResultHolder deferredResultHolder;
    @RequestMapping("/finished")
    @ResponseBody
    public void finished() {
        String id = "abc";
        DeferredData defdatq = new CustomerDeferredData(id); // 此处的CustomerDeferredData为实现了DeferredData接口的实现模型
        return deferredResultHolder.handleDeferredData(defdatq);
    }
}

补充Spring泛型的tips
随着泛型用的越来越多，获取泛型实际类型信息的需求也会出现，如果用原生API，需要很多步操作才能获取到泛型，比如：
ParameterizedType parameterizedType =   
      (ParameterizedType) ABService.class.getGenericInterfaces()[0];  
Type genericType = parameterizedType.getActualTypeArguments()[1];  

Spring提供的ResolvableType API，提供了更加简单易用的泛型操作支持，如：
ResolvableType resolvableType1 = ResolvableType.forClass(ABService.class);  
resolvableType1.as(Service.class).getGeneric(1).resolve()  

对于获取更复杂的泛型操作ResolvableType更加简单。
假设我们的API是：
public interface Service {    }  

@org.springframework.stereotype.Service  
public class ABService implements Service {}  
    
@org.springframework.stereotype.Service  
public class CDService implements Service {}

如上泛型类非常简单。
ResolvableType resolvableType1 = ResolvableType.forClass(ABService.class);  

通过如上API，可以得到类型的ResolvableType，如果类型被Spring AOP进行了CGLIB代理，请使用ClassUtils.getUserClass(ABService.class)得到原始类型。
可以通过如下得到泛型参数的第1个位置（从0开始）的类型信息resolvableType1.getInterfaces()[0].getGeneric(1).resolve()。
因为我们泛型信息放在 Service 上，所以需要resolvableType1.getInterfaces()[0]得到；通过getGeneric（泛型参数索引）得到某个位置的泛型，resolve()把实际泛型参数解析出来
得到字段级别的泛型信息
假设我们的字段如下：
  @Autowired  
  private Service abService;  
  @Autowired  
  private Service cdService;      
  private List> list;      
  private Map> map;      
  private List[] array;  

通过如下API可以得到字段级别的ResolvableType
ResolvableType resolvableType2 =  
ResolvableType.forField(ReflectionUtils.findField(GenricInjectTest.class, "cdService"));  

然后通过如下API得到Service的第0个位置上的泛型实参类型，即C：resolvableType2.getGeneric(0).resolve()，比如List> list;是一种嵌套的泛型用例，我们可以通过如下操作获取String类型：
ResolvableType resolvableType3 =  
ResolvableType.forField(ReflectionUtils.findField(GenricInjectTest.class, "list"));  
resolvableType3.getGeneric(0).getGeneric(0).resolve();  

更简单的写法
resolvableType3.getGeneric(0, 0).resolve()，List> 即String   

比如Map> map;我们想得到Integer，可以使用：
ResolvableType resolvableType4 =  
ResolvableType.forField(ReflectionUtils.findField(GenricInjectTest.class, "map"));  
resolvableType4.getGeneric(1).getGeneric(1).resolve(); 

更简单的写法
resolvableType4.getGeneric(1, 1).resolve()  

得到方法返回值的泛型信息**
假设我们的方法如下：
private HashMap> method() {  
      return null;  
}  

得到Map中的List中的String泛型实参：
ResolvableType resolvableType5 =  ResolvableType.forMethodReturnType(ReflectionUtils.findMethod(GenricInjectTest.class, "method"));  
resolvableType5.getGeneric(1, 0).resolve();  

得到构造器参数的泛型信息
假设我们的构造器如下：
public Const(List> list, Map> map) {  }  

我们可以通过如下方式得到第1个参数（ Map>）中的Integer：
ResolvableType resolvableType6 = ResolvableType.forConstructorParameter(ClassUtils.getConstructorIfAvailable(Const.class, List.class, Map.class), 1);  

resolvableType6.getGeneric(1, 0).resolve();  

得到数组组件类型的泛型信息
如对于private List[] array; 可以通过如下方式获取List的泛型实参String：
ResolvableType resolvableType7 =  ResolvableType.forField(ReflectionUtils.findField(GenricInjectTest.class, "array"));  
resolvableType7.isArray();//判断是否是数组  
resolvableType7.getComponentType().getGeneric(0).resolve();  

自定义泛型类型
ResolvableType resolvableType8 = ResolvableType.forClassWithGenerics(List.class, String.class);   ResolvableType resolvableType9 = ResolvableType.forArrayComponent(resolvableType8);  
resolvableType9.getComponentType().getGeneric(0).resolve();  

ResolvableType.forClassWithGenerics(List.class, String.class)相当于创建一个List类型；
ResolvableType.forArrayComponent(resolvableType8);：相当于创建一个List[]数组；
resolvableType9.getComponentType().getGeneric(0).resolve()：得到相应的泛型信息；
从如上操作可以看出其泛型操作功能十分完善，尤其在嵌套的泛型信息获取上相当简洁。目前整个Spring4环境都使用这个API来操作泛型信息。


            

            

        

        

            

                

                    
关于我们

                    
Customize this section to tell your visitors a little bit about your publication,
                        writers, content, or something else entirely. Totally up to you.

                


                

                    
热门标签

                    
                        
                    
                        
                    
                        
                    
                        
                    
                        
                    
                        
                    
                        
                    
                        
                    
                        
                    
                        
                    
                        
                    
                        
                    
                        
                    
                        
                    
                        
                    
                        
                    
                        
                    
                        
                    
                        
                    
                        
                    
                        
                    
                        
                    
                        
                    
                        
                    
                        
                    
                        
                    
                        
                    
                        
                    
                        
                    
                        
                    
                


                

                    
Elsewhere

                    
    
                        
  1. GitHub
    2. 
                        
  2. 竹心导航
    3. 
                    

                    

                        返回顶部