Dubbo学习系列之十二（Quartz任务调度）

public class HelloJob implements Job {
    @Override
    public void execute(JobExecutionContext context) throws JobExecutionException {
        System.out.println(System.currentTimeMillis()+"helloWorld");
    }
}

public class ScheduledTaskMain {
    public static void main(String[] args) throws SchedulerException {
        /*创建一个jobDetail的实例，将该实例与HelloJob Class绑定*/
        JobDetail jobDetail = JobBuilder.newJob(HelloJob.class).withIdentity("HelloJob").build();
        /*创建一个触发器，每2秒执行一次任务，一直持续下去*/
        SimpleTrigger cronTrigger=  TriggerBuilder.newTrigger().withIdentity("HelloTrigger").startNow()
                .withSchedule(SimpleScheduleBuilder.simpleSchedule().withIntervalInSeconds(2).repeatForever()).build();
        /*创建schedule实例*/
        StdSchedulerFactory factory = new StdSchedulerFactory();
        Scheduler scheduler = factory.getScheduler();
        /*将Job和trigger放入Scheduler容器*/
        scheduler.scheduleJob(jobDetail,cronTrigger);
        scheduler.start();
    }
}

@Component
@DisallowConcurrentExecution //标识这个任务是串行执行，不是并发执行
public class HelloJob2 implements Job, Serializable {
    @Autowired
    private DubboDeliveryService deliveryService;

    /*经测试，下面这种构造方法注入deliveryService的方式会导致NPE！*/
/*    @Autowired
    public HelloJob2(DubboDeliveryService deliveryService){
        this.deliveryService = deliveryService;
    }

    public HelloJob2(){}
*/
    @Override
    public void execute(JobExecutionContext context) throws JobExecutionException {
        int num = deliveryService.checkDelayed();
        System.out.println("delayed num is >>> "+ num);
    }
}

    //检查延误未发的物流单,
    /*仅供逻辑测试，直接查找出所有10天之前的订单，生产逻辑肯定比这复杂*/
    @Override
    public int checkDelayed() {
        QueryWrapper qw = new QueryWrapper();
        LocalDateTime timeNow = LocalDateTime.now(ZoneId.systemDefault());
        qw.lt(true,"gmt_create",timeNow.minusDays(10L));
        List list = deliveryDao.selectList(qw);
        return Objects.isNull(list)? 0: list.size();
    }

@Component
public class MyJobFactory extends AdaptableJobFactory {
    /**
     * AutowireCapableBeanFactory接口是BeanFactory的子类
     * 可以连接和填充那些生命周期不被Spring管理的已存在的bean实例
     */
    private AutowireCapableBeanFactory capableBeanFactory;

    @Autowired
    public MyJobFactory(AutowireCapableBeanFactory capableBeanFactory){
        this.capableBeanFactory = capableBeanFactory;
    }

    @Override
    protected Object createJobInstance(TriggerFiredBundle bundle) throws Exception{
        //调用父类方法
        Object jobInstance = super.createJobInstance(bundle);
        //进行注入（Spring接管该Bean）
        capableBeanFactory.autowireBean(jobInstance);
        return jobInstance;
    }
}

@Configuration
public class QuartzConfig {
    //不推荐这里注解@Autowired，使用构造函数注入
    private MyJobFactory myJobFactory;
    @Autowired
    public QuartzConfig(MyJobFactory myJobFactory){
        this.myJobFactory = myJobFactory;
    }
    @Bean(name = "factoryBean")
    public SchedulerFactoryBean schedulerFactoryBean(){
        // Spring提供SchedulerFactoryBean为Scheduler提供配置信息,并被Spring容器管理其生命周期
        SchedulerFactoryBean factoryBean = new SchedulerFactoryBean();
        factoryBean.setOverwriteExistingJobs(true);
        //设置是否自动启动
        factoryBean.setAutoStartup(false);
        //设置系统启动后，Starting Quartz Scheduler的延迟时间
        factoryBean.setStartupDelay(30);
        // 设置自定义Job Factory，用于Spring管理Job bean
        factoryBean.setJobFactory(myJobFactory);
        return factoryBean;
    }

    @Bean(name = "myScheduler")
    public Scheduler getScheduler(){
        Scheduler scheduler = schedulerFactoryBean().getScheduler();
        return scheduler;
    }
}

/**
 * {@link FactoryBean} that creates and configures a Quartz {@link org.quartz.Scheduler},
 * manages its lifecycle as part of the Spring application context, and exposes the
 * Scheduler as bean reference for dependency injection.
 *
 * 
Allows registration of JobDetails, Calendars and Triggers, automatically
 * starting the scheduler on initialization and shutting it down on destruction.
 * In scenarios that just require static registration of jobs at startup, there
 * is no need to access the Scheduler instance itself in application code.
 // 代码省略部分 ... ...
 */
 public class SchedulerFactoryBean extends SchedulerAccessor implements FactoryBean,
    BeanNameAware, ApplicationContextAware, InitializingBean, DisposableBean, SmartLifecycle {
    
  public static final String PROP_THREAD_COUNT = "org.quartz.threadPool.threadCount";
  public static final int DEFAULT_THREAD_COUNT = 10;
  private static final ThreadLocal cOnfigTimeResourceLoaderHolder= new ThreadLocal<>();
  private static final ThreadLocal cOnfigTimeTaskExecutorHolder= new ThreadLocal<>();
  private static final ThreadLocal cOnfigTimeDataSourceHolder= new ThreadLocal<>();
  private static final ThreadLocal cOnfigTimeNonTransactionalDataSourceHolder= new ThreadLocal<>();
   // 代码省略部分 ... ...
  //---------------------------------------------------------------------
  // 实现接口InitializingBean，即SpringBean生命周期中的afterPropertiesSet()方法，dataSource是持久化属性，
  @Override
  public void afterPropertiesSet() throws Exception {
    if (this.dataSource == null && this.nonTransactionalDataSource != null) {
      this.dataSource = this.nonTransactionalDataSource;
    }

    if (this.applicationContext != null && this.resourceLoader == null) {
      this.resourceLoader = this.applicationContext;
    }

    // 初始化Scheduler实例，将Jobs/Triggers注册
    this.scheduler = prepareScheduler(prepareSchedulerFactory());
    try {
      registerListeners();
      registerJobsAndTriggers();
    }
    catch (Exception ex) {
      try {
        this.scheduler.shutdown(true);
      }
      catch (Exception ex2) {
        logger.debug("Scheduler shutdown exception after registration failure", ex2);
      }
      throw ex;
    }
  }
  // 代码省略部分 ... ...
  /**
   * 初始化当前的SchedulerFactory, 应用本地定义的属性值
   * @param参数schedulerFactory为需要初始化的对象
   */
  private void initSchedulerFactory(StdSchedulerFactory schedulerFactory) throws SchedulerException, IOException {
    Properties mergedProps = new Properties();
    if (this.resourceLoader != null) {
      mergedProps.setProperty(StdSchedulerFactory.PROP_SCHED_CLASS_LOAD_HELPER_CLASS,
          ResourceLoaderClassLoadHelper.class.getName());
    }

    if (this.taskExecutor != null) {
      mergedProps.setProperty(StdSchedulerFactory.PROP_THREAD_POOL_CLASS,
          LocalTaskExecutorThreadPool.class.getName());
    }
    else {
      // 设置必要的默认属性，Quartz会使用显式属性设置覆盖默认属性 
      mergedProps.setProperty(StdSchedulerFactory.PROP_THREAD_POOL_CLASS, SimpleThreadPool.class.getName());
      mergedProps.setProperty(PROP_THREAD_COUNT, Integer.toString(DEFAULT_THREAD_COUNT));
    }

    if (this.configLocation != null) {
      if (logger.isInfoEnabled()) {
        logger.info("Loading Quartz config from [" + this.configLocation + "]");
      }
      PropertiesLoaderUtils.fillProperties(mergedProps, this.configLocation);
    }

    CollectionUtils.mergePropertiesIntoMap(this.quartzProperties, mergedProps);
    if (this.dataSource != null) {
      mergedProps.put(StdSchedulerFactory.PROP_JOB_STORE_CLASS, LocalDataSourceJobStore.class.getName());
    }
    if (this.schedulerName != null) {
      mergedProps.put(StdSchedulerFactory.PROP_SCHED_INSTANCE_NAME, this.schedulerName);
    }

    schedulerFactory.initialize(mergedProps);
  }
 // 代码省略部分 ... ...
  /**
   * 根据给定的factory 和scheduler name生成Scheduler实例，由afterPropertiesSet调用
    缺省实现将触发SchedulerFactory的getScheduler方法
   * @param 参数schedulerFactory生产Scheduler的工厂
   * @param schedulerName the name of the scheduler to create
   * @return the Scheduler instance
   * @throws SchedulerException if thrown by Quartz methods
   * @see #afterPropertiesSet
   * @see org.quartz.SchedulerFactory#getScheduler
   */
  protected Scheduler createScheduler(SchedulerFactory schedulerFactory, @Nullable String schedulerName)
      throws SchedulerException {

    // Override thread context ClassLoader to work around native Quartz ClassLoadHelper loading.
    Thread currentThread = Thread.currentThread();
    ClassLoader threadContextClassLoader = currentThread.getContextClassLoader();
    boolean overrideClassLoader = (this.resourceLoader != null &&
        this.resourceLoader.getClassLoader() != threadContextClassLoader);
    if (overrideClassLoader) {
      currentThread.setContextClassLoader(this.resourceLoader.getClassLoader());
    }
    try {
      SchedulerRepository repository = SchedulerRepository.getInstance();
      synchronized (repository) {
        Scheduler existingScheduler = (schedulerName != null ? repository.lookup(schedulerName) : null);
        Scheduler newScheduler = schedulerFactory.getScheduler();
        if (newScheduler == existingScheduler) {
          throw new IllegalStateException("Active Scheduler of name ' " + schedulerName + " ' already registered " +
              "in Quartz SchedulerRepository. Cannot create a new Spring-managed Scheduler of the same name!");
        }
        if (!this.exposeSchedulerInRepository) {
          // Need to remove it in this case, since Quartz shares the Scheduler instance by default!
          SchedulerRepository.getInstance().remove(newScheduler.getSchedulerName());
        }
        return newScheduler;
      }
    }
    finally {
      if (overrideClassLoader) {
        // 重置初始的线程ClassLoader.
        currentThread.setContextClassLoader(threadContextClassLoader);
      }
    }
  }
  /**
   * 将指定的或当前的ApplicationContext暴露给Quartz SchedulerContext.
   */
  private void populateSchedulerContext(Scheduler scheduler) throws SchedulerException {
    // 将对象放入Scheduler context.
    if (this.schedulerContextMap != null) {
      scheduler.getContext().putAll(this.schedulerContextMap);
    }
    // 在Scheduler context中注册ApplicationContext.
    if (this.applicationContextSchedulerContextKey != null) {
      if (this.applicatiOnContext== null) {
        throw new IllegalStateException(
          "SchedulerFactoryBean needs to be set up in an ApplicationContext " +
          "to be able to handle an ' applicationContextSchedulerContextKey'");
      }
      scheduler.getContext().put(this.applicationContextSchedulerContextKey, this.applicationContext);
    }
  }
  /**
   * 根据startupDelay设置启动Scheduler，注意这里是异步启动
   * @param scheduler the Scheduler to start
   * @param startupDelay the number of seconds to wait before starting
   * the Scheduler asynchronously
   */
  protected void startScheduler(final Scheduler scheduler, final int startupDelay) throws SchedulerException {
    if (startupDelay <= 0) {
      logger.info("Starting Quartz Scheduler now");
      scheduler.start();
    }
    else {
      if (logger.isInfoEnabled()) {
        logger.info("Will start Quartz Scheduler [" + scheduler.getSchedulerName() +
            "] in " + startupDelay + " seconds");
      }
      // 因这里明确需要一个守护线程，所以不使用Quartz的startDelayed方法，
      // 这样当其他线程全部终止时，应用就终止，JVM也会退出
      Thread schedulerThread = new Thread() {
        @Override
        public void run() {
          try {
            Thread.sleep(TimeUnit.SECONDS.toMillis(startupDelay));
          }
          catch (InterruptedException ex) {
            Thread.currentThread().interrupt();
            // 简单处理
          }
          if (logger.isInfoEnabled()) {
            logger.info("Starting Quartz Scheduler now, after delay of " + startupDelay + " seconds");
          }
          try {
            scheduler.start();
          }
          catch (SchedulerException ex) {
            throw new SchedulingException("Could not start Quartz Scheduler after delay", ex);
          }
        }
      };
      schedulerThread.setName("Quartz Scheduler [" + scheduler.getSchedulerName() + "]");
      schedulerThread.setDaemon(true); // 指定thread为deamon类型
      schedulerThread.start();
    }
  }
  // 代码省略部分 ... ...
    }

@Service
public class QuartzService {
    private final String groupName = "group1";

    @Autowired
    @Qualifier(value = "factoryBean")
    private SchedulerFactoryBean factoryBean;

    //以下方式也可以获取bean
    // Scheduler scheduler = SpringUtil.getBean("myScheduler");
    @Autowired
    @Qualifier(value = "myScheduler")
    private Scheduler scheduler;

    // 启动 Scheduler
    public void startScheduleJobs() throws SchedulerException {
        if (this.scheduler.isStarted()){
            return;
        }
        this.setCheckScheduler(scheduler);
        this.scheduler.start();

    }

    // 停止 Scheduler
    public void stopScheduleJobs() {
        scheduler = factoryBean.getScheduler();
        try {
            if (!scheduler.isShutdown()){
                scheduler.shutdown();
            }
        } catch (SchedulerException e) {
            e.printStackTrace();
        }
    }

    // 添加 Job 并替换
    public void addJobandReplace(){
        //打印hellowrold的job
        JobDetail jobDetail = JobBuilder.newJob(HelloJob.class).withIdentity("HelloJob").storeDurably(true).build();
        // 第二个参数为replace，是否替换存在的同名job
        // jobDetail必须是durable属性，表示任务完成之后是否依然保留到数据库，且无定义关联的trigger
        try {
            this.scheduler.addJob(jobDetail,true);
        } catch (SchedulerException e) {
            e.printStackTrace();
        }
    }
    // 添加 Job 不替换
    public void addJobwithoutReplace(){
        //打印hellowrold的job
        JobDetail jobDetail = JobBuilder.newJob(HelloJob.class).withIdentity("HelloJob").storeDurably(true).build();
        // 第二个参数为replace，是否替换存在的同名job
        try {
            this.scheduler.addJob(jobDetail,false);
        } catch (SchedulerException e) {
            e.printStackTrace();
        }
    }
    // 暂停所有 Job，还可指定具体的Job
    public void pauseScheduler(){
        try {
            this.scheduler.pauseAll();
        } catch (SchedulerException e) {
            e.printStackTrace();
        }
    }
    // 恢复并继续所有Job执行，还可指定具体的Job
    public void resumeJobs(){
        try {
            this.scheduler.resumeAll();
        } catch (SchedulerException e) {
            e.printStackTrace();
        }
    }

    //配置一个自定义的scheduler
    private void setCheckScheduler(Scheduler scheduler) throws SchedulerException {
        //添加HelloJob2作为任务内容
        JobDetail jobDetail = JobBuilder.newJob(HelloJob2.class)
                .withIdentity("job1",groupName).build();
        //cron表达式制定触发规则，每10秒执行一次
        CronScheduleBuilder scheduleBuilder = CronScheduleBuilder.cronSchedule("0/10 * * * * ?");
        CronTrigger cronTrigger = TriggerBuilder.newTrigger().withIdentity("trigger1",groupName)
                .withSchedule(scheduleBuilder).build();
        scheduler.scheduleJob(jobDetail,cronTrigger);
    }
}

@RestController
public class DubboDeliveryController {

   // 代码省略部分 ... ...
    
    @RequestMapping("/delivery/start")
    public String start() throws SchedulerException {
        quartzService.startScheduleJobs();
        return "startScheduleJobs success";
    }
    @RequestMapping("/delivery/stop")
    public String stop(){
        quartzService.stopScheduleJobs();
        return "stopScheduleJobs success";
    }
    @RequestMapping("/delivery/add")
    public String addJob(){
        quartzService.addJobandReplace();
        return "addJobandReplace success";
    }
    @RequestMapping("/delivery/pause")
    public String pauseJob(){
        quartzService.pauseScheduler();
        return "pauseScheduler success";
    }
    @RequestMapping("/delivery/resume")
    public String resumeJob(){
        quartzService.resumeJobs();
        return "resumeJobs success";
    }
}