Spring Boot集成Spring Scheduler和Quartz Scheduler

本文介紹了Spring Boot集成Spring Scheduler和Quartz Scheduler的基礎知識，利用ShedLock解決Spring Scheduler多實例運行沖突，介紹了Quartz ScheduleBuilder、Calendar，介紹了動態創建Quartz Job的方法。

GitHub源碼

Spring Scheduler

Spring Framework提供了簡單、易用的Job調度框架Spring Scheduler。

示例

在Spring Boot中，只需兩步即可啟用Scheduler：

1. 啟用Scheduling

package org.itrunner.heroes.scheduling;

import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.EnableScheduling;

@Configuration
@EnableScheduling
public class ScheduleConfig {
}

2. 定義Schedule方法

package org.itrunner.heroes.scheduling;

import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

@Component
@Slf4j
public class HelloSpring {

    @Scheduled(cron = "0 */10 * * * *")
    public void sayHello() {
        log.info("Hello Spring Scheduler");
    }

}

@Scheduled支持cron、fixedDelay、fixedRate三種定義方式，方法必須沒有參數，返回void類型。

ShedLock

默認情況下，Spring無法同步多個實例的調度程序，而是在每個節點上同時執行作業。我們可以使用shedlock-spring解決這一問題，確保在同一時間僅調度一次任務。

<dependency>
    <groupId>net.javacrumbs.shedlock</groupId>
    <artifactId>shedlock-spring</artifactId>
    <version>4.1.0</version>
</dependency>
<dependency>
    <groupId>net.javacrumbs.shedlock</groupId>
    <artifactId>shedlock-provider-jdbc-template</artifactId>
    <version>4.1.0</version>
</dependency>

ShedLock是利用數據庫鎖機制實現的，當前支持DynamoDB、Hazelcast、Mongo、Redis、ZooKeeper和任何JDBC Driver。為了使用JDBC，增加下面依賴：

<dependency>
  <groupId>net.javacrumbs.shedlock</groupId>
  <artifactId>shedlock-provider-jdbc-template</artifactId>
  <version>4.1.0</version>
</dependency>

創建Shedlock Entity：

package org.itrunner.heroes.domain;

import lombok.Data;

import javax.persistence.Column;
import javax.persistence.Entity;
import javax.persistence.Id;
import javax.persistence.Table;
import java.time.LocalDateTime;

@Entity
@Table(name = "shedlock")
@Data
public class Shedlock {
    @Id
    @Column(name = "name", length = 64)
    private String name;

    @Column(name = "lock_until")
    private LocalDateTime lockUntil;

    @Column(name = "locked_at")
    private LocalDateTime lockedAt;

    @Column(name = "locked_by")
    private String lockedBy;
}

啟用ShedLock：

package org.itrunner.heroes.scheduling;

import net.javacrumbs.shedlock.core.LockProvider;
import net.javacrumbs.shedlock.provider.jdbctemplate.JdbcTemplateLockProvider;
import net.javacrumbs.shedlock.spring.annotation.EnableSchedulerLock;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.EnableScheduling;

import javax.sql.DataSource;

@Configuration
@EnableScheduling
@EnableSchedulerLock(defaultLockAtMostFor = "PT30S")
public class ScheduleConfig {

    @Bean
    public LockProvider lockProvider(DataSource dataSource) {
        return new JdbcTemplateLockProvider(dataSource);
    }

}

package org.itrunner.heroes.scheduling;

import lombok.extern.slf4j.Slf4j;
import net.javacrumbs.shedlock.spring.annotation.SchedulerLock;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

@Component
@Slf4j
public class HelloSpring {

    @Scheduled(cron = "0 */10 * * * *")
    @SchedulerLock(name = "helloSpringScheduler", lockAtLeastFor = "PT30S", lockAtMostFor = "PT3M")
    public void sayHello() {
        log.info("Hello Spring Scheduler");
    }

}

其中lockAtLeastFor和lockAtMostFor設置lock的最短和最長時間，上例分別為30秒、3分鐘。

Quartz Scheduler

Quartz Scheduler是功能強大的任務調度框架，在Spring Scheduler不能滿足需求時可以使用Quartz。

集成Quartz

1. POM依賴

Spring Boot項目中僅需引入依賴spring-boot-starter-quartz：

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-quartz</artifactId>
</dependency>

2. 配置Quartz

默認，使用內存JobStore，在生產環境中應配置使用數據庫：

spring:
  quartz:
    auto-startup: true
    job-store-type: jdbc
    jdbc:
      initialize-schema: always
    overwrite-existing-jobs: true
    properties:
      org.quartz.threadPool.threadCount: 5

在spring.quartz.properties中可以配置Quartz高級屬性。

3. 定義Quartz Job

package org.itrunner.heroes.scheduling;

import lombok.extern.slf4j.Slf4j;
import org.quartz.JobExecutionContext;
import org.quartz.JobExecutionException;
import org.springframework.scheduling.quartz.QuartzJobBean;

@Slf4j
public class HelloQuartz extends QuartzJobBean {

    @Override
    protected void executeInternal(JobExecutionContext jobExecutionContext) throws JobExecutionException {
        log.info("Hello Quartz Scheduler");
    }

}

4. 注冊Job

集成Quartz后，一個Scheduler被自動配置。SchedulerFactoryBean負責創建和配置Quartz Scheduler，作為Spring application context的一部分管理其生命周期。scheduler可以在其它組件中注入。

所有的JobDetail、Calendar和Trigger Bean自動與scheduler關聯，在Spring Boot初始化時自動啟動scheduler，并在銷毀時將其關閉。

靜態注冊Job
僅需在啟動時靜態注冊Job的情況下，只需聲明Bean，無需在程序中訪問scheduler實例本身，如下：

package org.itrunner.heroes.scheduling;

import org.itrunner.heroes.util.DateUtils;
import org.quartz.*;
import org.quartz.impl.calendar.HolidayCalendar;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.time.LocalDate;

@Configuration
public class QuartzConfig {
    private static final String CRON_EXPRESSION = "0 0/5 * * * ?";
    private static final String GROUP = "iTRunner";

    @Bean
    public Trigger helloJobTrigger(JobDetail helloJob) {
        CronScheduleBuilder scheduleBuilder = CronScheduleBuilder.cronSchedule(CRON_EXPRESSION);
        return TriggerBuilder.newTrigger().forJob(helloJob).withIdentity(getTriggerKey(helloJob.getKey())).withSchedule(scheduleBuilder).modifiedByCalendar("holidayCalendar").build();
    }

    @Bean
    public JobDetail helloJob() {
        return JobBuilder.newJob(HelloQuartz.class).withIdentity(getJobKey(HelloQuartz.class)).storeDurably().build();
    }

    @Bean
    public Calendar holidayCalendar() {
        HolidayCalendar calendar = new HolidayCalendar();
        LocalDate date = LocalDate.of(2020, 1, 1);
        calendar.addExcludedDate(DateUtils.toDate(date));
        return calendar;
    }

    private static <T> JobKey getJobKey(Class<T> cls) {
        return new JobKey(cls.getSimpleName(), GROUP);
    }

    private static TriggerKey getTriggerKey(JobKey jobKey) {
        return new TriggerKey(jobKey.getName(), GROUP);
    }
}

ScheduleBuilder

上例，我們使用了CronScheduleBuilder，Quartz還支持SimpleScheduleBuilder、DailyTimeIntervalScheduleBuilder、CalendarIntervalScheduleBuilder。

• SimpleScheduleBuilder 按指定的時間間隔執行任務，單位可以為毫秒、秒、分或小時；可以指定執行任務的總次數。
• DailyTimeIntervalScheduleBuilder 每天按指定的時間間隔執行任務，單位可以為秒、分或小時；可以指定每天執行任務的起止時間；可以指定在星期幾執行；可以指定每天執行任務的總次數
• CalendarIntervalScheduleBuilder 按指定的時間間隔執行任務，單位可以為秒、分、時、日、周、月、年
• CronScheduleBuilder 使用Cron表達式定義任務執行時間，也支持dailyAtHourAndMinute()、atHourAndMinuteOnGivenDaysOfWeek()、weeklyOnDayAndHourAndMinute()、monthlyOnDayAndHourAndMinute()等便利方法。

Cron-Expression
Cron 表達式由 6 個必選字段和一個可選字段組成，字段間由空格分隔。

* 可用在所有字段，例如，在分鐘字段表示每分鐘
? 允許應用在日和周字段，用于指定“非特定值”，相當于占位符
- 用于指定范圍，如在小時字段，“10-12”表示10，11，12
, 指定列表值，如在周字段，"MON,WED,FRI"表示周一，周三，周五
/ 指定步長，如在秒字段，"0/15"表示0，15，30，45；"5/15"表示5, 20, 35, 50。如使用*/x，相當于0/x。
L 只用于日和周字段，意為“last”。在日字段，如一月的31號，非閏年二月的28號；在周字段，表示7或"SAT"，若在L前還有一個值，如6L，表示這個月最后的周五。
W 僅用于日字段，表示離指定日期最近的工作日（周一至周五），如15W，表示離該月15號最近的工作日，注意不能跨月。
LW組合， 表示當月最后一個工作日
# 僅用于周字段，表示第幾，如“6#3”，表示本月第3個周五

示例：

0 0/5 * * * ?    每5分鐘
10 0/5 * * * ?  每5分鐘，10秒時執行，如10:00:10, 10:05:10
0 30 10-13 ? * WED,FRI  每周三和周五的10:30, 11:30, 12:30 和 13:30
0 0/30 8-9 5,20 * ?  每月5號和20號的8:00, 8:30, 9:00 和 9:30

Calendar

Calendar不定義實際的觸發時間，而是與Trigger結合使用，用于排除特定的時間。

AnnualCalendar 排除每年中的一天或多天
CronCalendar 使用Cron表達式定義排除的時間，如"* * 0-7,18-23 ? * *"，排除每天的8點至17點
DailyCalendar 排除每天指定的時間段
HolidayCalendar 排除節假日，需要指定確切的日期
MonthlyCalendar 排除每月的一天或多天
WeeklyCalendar 排除每周的一天或多天，默認排除周六、周日

動態Job

在很多情況下我們需要動態創建或啟停Job，比如Job數據是動態的、Job間有依賴關系、根據條件啟停Job等。

下面示例簡單演示了動態創建Job、添加calendar、添加listener、啟停job的方法：

package org.itrunner.heroes.scheduling;

import lombok.extern.slf4j.Slf4j;
import org.quartz.*;
import org.quartz.impl.calendar.WeeklyCalendar;
import org.quartz.impl.matchers.GroupMatcher;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import static org.itrunner.heroes.scheduling.Constants.*;

@Service
@Slf4j
public class ScheduleService {
    private final Scheduler scheduler;

    @Autowired
    public ScheduleService(Scheduler scheduler) { // 注入scheduler
        this.scheduler = scheduler;

        try {
            addJobListener();
            addCalendar();
            scheduleJob();
        } catch (SchedulerException e) {
            log.error(e.getMessage(), e);
        }
    }

    public void unscheduleJob(String jobName) throws SchedulerException {
        scheduler.pauseJob(JobKey.jobKey(jobName, GROUP_NAME));
        scheduler.unscheduleJob(TriggerKey.triggerKey(jobName, GROUP_NAME));
    }

     /**
     * 立即觸發job
     */
    public void triggerJob(String jobName) throws SchedulerException {
        scheduler.triggerJob(JobKey.jobKey(jobName, GROUP_NAME));
    }

    private void addJobListener() throws SchedulerException {
        UnscheduleJobListener jobListener = new UnscheduleJobListener();
        GroupMatcher<JobKey> groupMatcher = GroupMatcher.jobGroupEquals(GROUP_NAME);
        this.scheduler.getListenerManager().addJobListener(jobListener, groupMatcher);
    }

    private void addCalendar() throws SchedulerException {
        WeeklyCalendar calendar = new WeeklyCalendar();
        calendar.setDayExcluded(1, true); // 排除周日
        calendar.setDayExcluded(7, false);
        this.scheduler.addCalendar("weekly", calendar, false, false);
    }

    private void scheduleJob() throws SchedulerException {
        JobDetail jobDetail = createJobDetail();
        Trigger trigger = createTrigger(jobDetail);
        scheduler.scheduleJob(jobDetail, trigger);
    }

    private JobDetail createJobDetail() {
        JobDataMap jobDataMap = new JobDataMap(); // 添加Job數據
        jobDataMap.put(JOB_NAME, "getHeroes");
        jobDataMap.put(JOB_REST_URI, "http://localhost:8080/api/heroes");
        jobDataMap.put(JOB_REQUEST_METHOD, "GET");

        return JobBuilder.newJob(RestJob.class).withIdentity("getHeroes", GROUP_NAME).usingJobData(jobDataMap).storeDurably().build();
    }

    private Trigger createTrigger(JobDetail jobDetail) {
        DailyTimeIntervalScheduleBuilder scheduleBuilder = DailyTimeIntervalScheduleBuilder.dailyTimeIntervalSchedule().withIntervalInMinutes(1).onEveryDay();
        return TriggerBuilder.newTrigger().forJob(jobDetail).withIdentity("getHeroes", GROUP_NAME).withSchedule(scheduleBuilder).modifiedByCalendar("weekly").build();
    }

}

Job定義

下面Job調用了REST服務，調用成功后在JobExecutionContext中添加stop標志：

package org.itrunner.heroes.scheduling;

import lombok.extern.slf4j.Slf4j;
import org.quartz.JobDataMap;
import org.quartz.JobExecutionContext;
import org.quartz.JobExecutionException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.HttpMethod;
import org.springframework.http.ResponseEntity;
import org.springframework.scheduling.quartz.QuartzJobBean;

import java.util.List;

import static org.itrunner.heroes.scheduling.Constants.JOB_REST_URI;
import static org.itrunner.heroes.scheduling.Constants.JOB_STOP_FLAG;

@Slf4j
public class RestJob extends QuartzJobBean {
    @Autowired
    private RestService restService;

    @Override
    protected void executeInternal(JobExecutionContext jobExecutionContext) throws JobExecutionException {
        JobDataMap jobDataMap = jobExecutionContext.getMergedJobDataMap();
        String restUri = jobDataMap.getString(JOB_REST_URI);

        ResponseEntity<List> responseEntity = restService.requestForEntity(restUri, HttpMethod.GET, List.class);
        log.info(responseEntity.getBody().toString());

        // set stop flag
        jobExecutionContext.put(JOB_STOP_FLAG, true);
    }
}

JobListener
UnscheduleJobListener檢查JobExecutionContext中是否有stop標志，如有則停止Job：

package org.itrunner.heroes.scheduling;

import org.quartz.JobExecutionContext;
import org.quartz.JobExecutionException;
import org.quartz.JobListener;
import org.quartz.SchedulerException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class UnscheduleJobListener implements JobListener {
    private static Logger log = LoggerFactory.getLogger(UnscheduleJobListener.class);

    @Override
    public String getName() {
        return "HERO_UnscheduleJobListener";
    }

    @Override
    public void jobToBeExecuted(JobExecutionContext context) {
        log.info(getJobName(context) + " is about to be executed.");
    }

    @Override
    public void jobExecutionVetoed(JobExecutionContext context) {
        log.info(getJobName(context) + " Execution was vetoed.");
    }

    @Override
    public void jobWasExecuted(JobExecutionContext context, JobExecutionException jobException) {
        log.info(getJobName(context) + " was executed.");

        Boolean stop = (Boolean) context.get(Constants.JOB_STOP_FLAG);

        if (stop == null || !stop) {
            return;
        }

        String jobName = getJobName(context);

        log.info("Unschedule " + jobName);

        try {
            context.getScheduler().unscheduleJob(context.getTrigger().getKey());
        } catch (SchedulerException e) {
            log.error("Unable to unschedule " + jobName, e);
        }
    }

    private String getJobName(JobExecutionContext context) {
        return "Hero job " + context.getJobDetail().getKey().getName();
    }
}