手擼XXL-JOB（二）——定時任務管理

在上一節中，我們介紹了SpringBoot中關于定時任務的執行方式，以及ScheduledExecutorService接口提供的定時任務執行方法。假設我們現在要寫類似XXL-JOB這樣的任務調度平臺，那么，對于任務的管理，是尤為重要的。接下來我們將一步一步，實現一個任務調度管理類。

YangJobManager類基礎實現

假設我們現在的任務管理類，名為YangJobManager類。對于定時任務的執行，我們最終會調用到ScheduledExecutorService的相關方法，因此，我們的YangJobManager類，需要有ScheduledExecutorService屬性，其次，我們希望能對要執行的定時線程任務，其命名進行修改，因此，我們需要有一個線程工廠的屬性。基于上述兩點，我們對YangJobManager類進行實現：

package com.yang.job;import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;public class YangJobManager {private ScheduledExecutorService scheduledExecutorService;private ThreadFactory threadFactory;public YangJobManager(ScheduledExecutorService scheduledExecutorService, ThreadFactory threadFactory) {this.scheduledExecutorService = scheduledExecutorService;this.threadFactory = threadFactory;}public void schedule(Runnable runnable, Long delay) {Thread thread = threadFactory.newThread(runnable);scheduledExecutorService.schedule(thread, delay, TimeUnit.SECONDS);}public void scheduleWithFixedDelay(Runnable runnable, Long delay, Long period) {Thread thread = threadFactory.newThread(runnable);scheduledExecutorService.scheduleWithFixedDelay(thread, delay, period, TimeUnit.SECONDS);}public void scheduleWithFixedRate(Runnable runnable, Long delay, Long period) {Thread thread = threadFactory.newThread(runnable);scheduledExecutorService.scheduleAtFixedRate(thread, delay, period, TimeUnit.SECONDS);}public void shutdown() {if (this.scheduledExecutorService == null) {return;}if (this.scheduledExecutorService.isShutdown()) {return;}scheduledExecutorService.shutdown();try {if (!scheduledExecutorService.awaitTermination(10, TimeUnit.SECONDS)) {scheduledExecutorService.shutdownNow();}} catch (InterruptedException e) {e.printStackTrace();}}
}

然后，我們實現YangJobThreadFactory，完成對線程的命名

public class YangJobThreadFactory implements ThreadFactory {private String poolName;private String threadPrefixName;private static AtomicInteger poolNumber = new AtomicInteger(1);private AtomicInteger threadNumber = new AtomicInteger(1);public YangJobThreadFactory(String poolName) {this.poolName = poolName;this.threadPrefixName = poolName + "-pool-" + poolNumber.getAndIncrement() + "-thread-";}public String getPoolName() {return this.poolName;}@Overridepublic Thread newThread(Runnable r) {Thread thread = new Thread(r);thread.setName(this.threadPrefixName + threadNumber.getAndIncrement());return thread;}}

然后我們添加測試方法：

 public static void main(String[] args) {ThreadFactory threadFactory = new YangJobThreadFactory("yang");ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(4, threadFactory);YangJobManager yangJobManager = new YangJobManager(scheduledExecutorService, threadFactory);yangJobManager.schedule(() -> {System.out.println(Thread.currentThread().getName() + "schedule定時任務開始執行：" + new Date());}, 1L);yangJobManager.scheduleWithFixedDelay(() -> {System.out.println(Thread.currentThread().getName() + "withFixedDelay定時任務開始執行：" + new Date());}, 0L, 1L);yangJobManager.scheduleWithFixedRate(() -> {System.out.println(Thread.currentThread().getName() + "withFixedRate定時任務開始執行：" + new Date());}, 0L, 1L);try {Thread.sleep(20000);} catch (InterruptedException e) {throw new RuntimeException(e);}yangJobManager.shutdown();}

執行結果如下：

提供統一的schedule方法

雖然我們能順利將任務提交給YangJobManager執行，當感覺還不夠收斂，因為我們創建了三個方法：schedule，scheduleWithFixedDelay, shceduleWithFixedRate，每個方法執行邏輯都差不多，最后都是調用scheduledExecutorService的相關方法，我們可以將這些方法都收斂到一個入口——schedule，然后在入參中添加一個參數，表示要執行的策略，根據入參的參數，選擇對應的方法執行。
首先，我們添加一個執行策略枚舉：

package com.yang.job.enums;public enum JobExecuteStrategyEnum {IMMEDIATE_EXECUTE("immediate", "立即執行"),ONCE("once", "執行一次"),WITH_FIXED_DELAY("withFixedDelay", "任務執行完畢后間隔執行"),WITH_FIXED_RATE("withFixedRate", "任務執行開始后間隔執行");private String name;private String description;JobExecuteStrategyEnum(String name, String description) {this.name = name;this.description = description;}public String getName() {return this.name;}public static JobExecuteStrategyEnum getJobExecuteStrategyByName(String name) {if (name == null) {return null;}for (JobExecuteStrategyEnum value : values()) {if (name.equals(value.getName())) {return value;}}return null;}public static boolean isLegal(String name) {JobExecuteStrategyEnum jobExecuteStrategyByName = getJobExecuteStrategyByName(name);return jobExecuteStrategyByName != null;}public String getDescription() {return description;}
}

然后添加YangJobManager的schedule方法的入參類：

package com.yang.job.request;import com.yang.job.enums.JobExecuteStrategyEnum;
import lombok.Data;import java.io.Serializable;@Data
public class YangJobSubmitParam implements Serializable {private Runnable runnable;private Integer initialDelay;private Integer period;private JobExecuteStrategyEnum jobExecuteStrategy;
}

最后，修改YangJobManager類，將執行定時任務收斂到schedule方法，進入該方法，首先根據入參判斷執行策略，如果是immediate，那么直接對入參的runnable調用run方法執行接口，其他的策略則分別對應scheduledExecutorService的schedule、scheduledWithFixedDelay、scheduledWithFixedRate方法，此外，這里對屬性也進行修改，去除ThreadFactory屬性。

package com.yang.job;import com.yang.job.enums.JobExecuteStrategyEnum;
import com.yang.job.request.YangJobSubmitParam;import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;public class YangJobManager {private ScheduledExecutorService scheduledExecutorService;public YangJobManager(ScheduledExecutorService scheduledExecutorService) {this.scheduledExecutorService = scheduledExecutorService;}public void schedule(YangJobSubmitParam yangJobSubmitParam) {JobExecuteStrategyEnum jobExecuteStrategy = yangJobSubmitParam.getJobExecuteStrategy();if (jobExecuteStrategy == null) {throw new RuntimeException("缺少執行策略=========");}Runnable runnable = yangJobSubmitParam.getRunnable();Integer initialDelay = yangJobSubmitParam.getInitialDelay();Integer period = yangJobSubmitParam.getPeriod();switch (jobExecuteStrategy) {case IMMEDIATE_EXECUTE:runnable.run();break;case ONCE:scheduledExecutorService.schedule(runnable, initialDelay, TimeUnit.SECONDS);break;case WITH_FIXED_DELAY:scheduledExecutorService.scheduleWithFixedDelay(runnable, initialDelay, period, TimeUnit.SECONDS);break;case WITH_FIXED_RATE:scheduledExecutorService.scheduleAtFixedRate(runnable, initialDelay, period, TimeUnit.SECONDS);break;}}public void shutdown() {if (this.scheduledExecutorService == null) {return;}if (this.scheduledExecutorService.isShutdown()) {return;}scheduledExecutorService.shutdown();try {if (!scheduledExecutorService.awaitTermination(10, TimeUnit.SECONDS)) {scheduledExecutorService.shutdownNow();}} catch (InterruptedException e) {e.printStackTrace();}}
}

最后，我們添加測試方法：

public static void main(String[] args) {ThreadFactory threadFactory = new YangJobThreadFactory("yang");ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(4, threadFactory);YangJobManager yangJobManager = new YangJobManager(scheduledExecutorService);YangJobSubmitParam yangJobSubmitParam1 = new YangJobSubmitParam();yangJobSubmitParam1.setRunnable(() -> System.out.println("立即執行======" + new Date()));yangJobSubmitParam1.setJobExecuteStrategy(JobExecuteStrategyEnum.IMMEDIATE_EXECUTE);YangJobSubmitParam yangJobSubmitParam2 = new YangJobSubmitParam();yangJobSubmitParam2.setRunnable(() -> System.out.println("執行一次======" + new Date()));yangJobSubmitParam2.setInitialDelay(1);yangJobSubmitParam2.setJobExecuteStrategy(JobExecuteStrategyEnum.ONCE);YangJobSubmitParam yangJobSubmitParam3 = new YangJobSubmitParam();yangJobSubmitParam3.setRunnable(() -> System.out.println("withFixedDelay=====" + new Date()));yangJobSubmitParam3.setInitialDelay(1);yangJobSubmitParam3.setPeriod(2);yangJobSubmitParam3.setJobExecuteStrategy(JobExecuteStrategyEnum.WITH_FIXED_DELAY);YangJobSubmitParam yangJobSubmitParam4 = new YangJobSubmitParam();yangJobSubmitParam4.setRunnable(() -> System.out.println("withFixedRate=====" + new Date()));yangJobSubmitParam4.setInitialDelay(1);yangJobSubmitParam4.setPeriod(2);yangJobSubmitParam4.setJobExecuteStrategy(JobExecuteStrategyEnum.WITH_FIXED_RATE);yangJobManager.schedule(yangJobSubmitParam1);yangJobManager.schedule(yangJobSubmitParam2);yangJobManager.schedule(yangJobSubmitParam3);yangJobManager.schedule(yangJobSubmitParam4);try {Thread.sleep(20000);} catch (InterruptedException e) {throw new RuntimeException(e);}yangJobManager.shutdown();}

執行結果如下：

提交任務和取消任務

任務的提交對應的是schedule方法，但我們的YangJobManager類缺少了關于任務的取消邏輯。在ScheduledExecutorService的各個定時執行方法中，其返回值是一個ScheduleFuture類，我們可以通過該類的cancel方法，來將對應的線程任務進行取消。此外，對于每一個任務，我們需要有一個任務標識，所以，我們先修改YangJobSubmitParam類：

package com.yang.job.request;import com.yang.job.enums.JobExecuteStrategyEnum;
import lombok.Data;import java.io.Serializable;@Data
public class YangJobSubmitParam implements Serializable {private Integer jobId;private Runnable runnable;private Integer initialDelay;private Integer period;private JobExecuteStrategyEnum jobExecuteStrategy;
}

然后，我們修改YangJobManager類，首先將schedule方法改為submit方法，這樣更見名知義，在submit方法中，除了理解執行策略外，其他策略都會獲取返回的ScheduleFuture，然后存入對應的map，在取消的時候，我們根據jobId從map中找到對應的ScheduleFuture，并執行cancel方法，以此來取消任務。

package com.yang.job;import com.yang.job.enums.JobExecuteStrategyEnum;
import com.yang.job.request.YangJobSubmitParam;import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;public class YangJobManager {private ScheduledExecutorService scheduledExecutorService;private Map<String, ScheduledFuture> jobId2ScheduleFutureMap = new ConcurrentHashMap<>();public YangJobManager(ScheduledExecutorService scheduledExecutorService) {this.scheduledExecutorService = scheduledExecutorService;}public void submitJob(YangJobSubmitParam yangJobSubmitParam) {Integer jobId = yangJobSubmitParam.getJobId();if (jobId == null) {throw new RuntimeException("缺少任務標識=========");}ScheduledFuture scheduledFuture = jobId2ScheduleFutureMap.get(jobId.toString());if (scheduledFuture != null && !scheduledFuture.isCancelled()) {// jobId存在對應的任務return;}JobExecuteStrategyEnum jobExecuteStrategy = yangJobSubmitParam.getJobExecuteStrategy();if (jobExecuteStrategy == null) {throw new RuntimeException("缺少執行策略=========");}if (jobExecuteStrategy == JobExecuteStrategyEnum.IMMEDIATE_EXECUTE) {yangJobSubmitParam.getRunnable().run();return;}scheduledFuture = scheduleJob(yangJobSubmitParam);jobId2ScheduleFutureMap.put(jobId.toString(), scheduledFuture);}public void cancelJob(Integer jobId) {if (jobId == null) {return;}ScheduledFuture scheduledFuture = jobId2ScheduleFutureMap.get(jobId.toString());if (scheduledFuture == null) {return;}if (!scheduledFuture.isCancelled()) {scheduledFuture.cancel(true);}jobId2ScheduleFutureMap.remove(jobId.toString());}private ScheduledFuture scheduleJob(YangJobSubmitParam yangJobSubmitParam) {Runnable runnable = yangJobSubmitParam.getRunnable();Integer initialDelay = yangJobSubmitParam.getInitialDelay();Integer period = yangJobSubmitParam.getPeriod();JobExecuteStrategyEnum jobExecuteStrategy = yangJobSubmitParam.getJobExecuteStrategy();switch (jobExecuteStrategy) {case ONCE:return scheduledExecutorService.schedule(runnable, initialDelay, TimeUnit.SECONDS);case WITH_FIXED_DELAY:return scheduledExecutorService.scheduleWithFixedDelay(runnable, initialDelay, period, TimeUnit.SECONDS);case WITH_FIXED_RATE:return scheduledExecutorService.scheduleAtFixedRate(runnable, initialDelay, period, TimeUnit.SECONDS);}throw new RuntimeException("執行策略有誤===========");}public void shutdown() {if (this.scheduledExecutorService == null) {return;}if (this.scheduledExecutorService.isShutdown()) {return;}scheduledExecutorService.shutdown();try {if (!scheduledExecutorService.awaitTermination(10, TimeUnit.SECONDS)) {scheduledExecutorService.shutdownNow();}} catch (InterruptedException e) {e.printStackTrace();}}
}

最后，我們添加對應的測試方法：

 public static void main(String[] args) {ThreadFactory threadFactory = new YangJobThreadFactory("yang");ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(4, threadFactory);YangJobManager yangJobManager = new YangJobManager(scheduledExecutorService);YangJobSubmitParam yangJobSubmitParam = new YangJobSubmitParam();yangJobSubmitParam.setJobId(1);yangJobSubmitParam.setRunnable(() -> System.out.println("執行任務=====" + new Date()));yangJobSubmitParam.setInitialDelay(0);yangJobSubmitParam.setPeriod(2);yangJobSubmitParam.setJobExecuteStrategy(JobExecuteStrategyEnum.WITH_FIXED_RATE);yangJobManager.submitJob(yangJobSubmitParam);try {Thread.sleep(10000);} catch (InterruptedException e) {e.printStackTrace();}System.out.println("取消任務==========");yangJobManager.cancelJob(1);try {Thread.sleep(10000);} catch (InterruptedException e) {e.printStackTrace();}yangJobManager.shutdown();}

在該方法中，我們提交任務，該任務間隔時間為2秒，10秒過后，取消任務，取消任務過后，再睡眠10秒，在后面10秒鐘，不會執行任務（或執行一次，因為在cancel之前剛好有任務沒執行完），執行結果如下：

YangJobManager建造者

對于YangJobManager，目前我們所擁有的屬性、方法都比較簡單，但是如果后續這個類進一步擴展，構造該類可能會變得很麻煩，因此，我們添加一個YangJobBuilder建造者類，用于構造YangJobManager，此外，我們將YangJobManager的構造方法設置為private，從而將構造YangJobManager的職責，徹底收斂到YangJobManagerBuilder類中，我們修改YangJobManager類如下：

package com.yang.job;import com.yang.job.enums.JobExecuteStrategyEnum;
import com.yang.job.factory.YangJobThreadFactory;
import com.yang.job.request.YangJobSubmitParam;import java.util.Map;
import java.util.concurrent.*;public class YangJobManager {private ScheduledExecutorService scheduledExecutorService;private Map<String, ScheduledFuture> jobId2ScheduleFutureMap = new ConcurrentHashMap<>();private YangJobManager(ScheduledExecutorService scheduledExecutorService) {this.scheduledExecutorService = scheduledExecutorService;}public void submitJob(YangJobSubmitParam yangJobSubmitParam) {Integer jobId = yangJobSubmitParam.getJobId();if (jobId == null) {throw new RuntimeException("缺少任務標識=========");}ScheduledFuture scheduledFuture = jobId2ScheduleFutureMap.get(jobId.toString());if (scheduledFuture != null && !scheduledFuture.isCancelled()) {// jobId存在對應的任務return;}JobExecuteStrategyEnum jobExecuteStrategy = yangJobSubmitParam.getJobExecuteStrategy();if (jobExecuteStrategy == null) {throw new RuntimeException("缺少執行策略=========");}if (jobExecuteStrategy == JobExecuteStrategyEnum.IMMEDIATE_EXECUTE) {yangJobSubmitParam.getRunnable().run();return;}scheduledFuture = scheduleJob(yangJobSubmitParam);jobId2ScheduleFutureMap.put(jobId.toString(), scheduledFuture);}public void cancelJob(Integer jobId) {if (jobId == null) {return;}ScheduledFuture scheduledFuture = jobId2ScheduleFutureMap.get(jobId.toString());if (scheduledFuture == null) {return;}if (!scheduledFuture.isCancelled()) {scheduledFuture.cancel(true);}jobId2ScheduleFutureMap.remove(jobId.toString());}private ScheduledFuture scheduleJob(YangJobSubmitParam yangJobSubmitParam) {Runnable runnable = yangJobSubmitParam.getRunnable();Integer initialDelay = yangJobSubmitParam.getInitialDelay();Integer period = yangJobSubmitParam.getPeriod();JobExecuteStrategyEnum jobExecuteStrategy = yangJobSubmitParam.getJobExecuteStrategy();switch (jobExecuteStrategy) {case ONCE:return scheduledExecutorService.schedule(runnable, initialDelay, TimeUnit.SECONDS);case WITH_FIXED_DELAY:return scheduledExecutorService.scheduleWithFixedDelay(runnable, initialDelay, period, TimeUnit.SECONDS);case WITH_FIXED_RATE:return scheduledExecutorService.scheduleAtFixedRate(runnable, initialDelay, period, TimeUnit.SECONDS);}throw new RuntimeException("執行策略有誤===========");}public void shutdown() {if (this.scheduledExecutorService == null) {return;}if (this.scheduledExecutorService.isShutdown()) {return;}scheduledExecutorService.shutdown();try {if (!scheduledExecutorService.awaitTermination(10, TimeUnit.SECONDS)) {scheduledExecutorService.shutdownNow();}} catch (InterruptedException e) {e.printStackTrace();}}public static class YangJobManagerBuilder {private ThreadFactory threadFactory;private ScheduledExecutorService scheduledExecutorService;public YangJobManagerBuilder() {}public YangJobManagerBuilder setThreadFactory(ThreadFactory threadFactory) {this.threadFactory = threadFactory;return this;}public YangJobManagerBuilder setScheduledExecutorService(ScheduledExecutorService scheduledExecutorService) {this.scheduledExecutorService = scheduledExecutorService;return this;}public YangJobManager build() {if (this.threadFactory == null) {this.threadFactory = new YangJobThreadFactory("yang");}if (this.scheduledExecutorService == null) {this.scheduledExecutorService = Executors.newScheduledThreadPool(Runtime.getRuntime().availableProcessors(),this.threadFactory);} else {if (this.scheduledExecutorService instanceof ScheduledThreadPoolExecutor) {ScheduledThreadPoolExecutor scheduledThreadPoolExecutor = (ScheduledThreadPoolExecutor) this.scheduledExecutorService;scheduledThreadPoolExecutor.setThreadFactory(this.threadFactory);}}return new YangJobManager(this.scheduledExecutorService);}}
}

任務執行類

在之前的代碼中，我們的Runnable都是匿名函數類，但是在我們的定時任務調度平臺中，一般情況下，這個任務是會持久化到數據庫中的，我們一般不會說把這個Runnable的代碼也存到數據庫吧，一般存儲的，應該就是某個任務執行類的類路徑，和方法名，以及入參，然后在啟動項目時，從數據庫中加載這些數據，并通過反射或代理等方式，來構造這個Runnable。
首先，我們定義一個任務執行類，來規范任務的執行方法和入參格式：

// 任務執行類
package com.yang.job.execute;public interface IYangJobExecutor {void execute(YangJobExecuteRequest yangJobExecuteRequest);
}// 任務執行方法入參
package com.yang.job.execute;import lombok.Data;import java.io.Serializable;
import java.util.HashMap;
import java.util.Map;@Data
public class YangJobExecuteRequest implements Serializable {private String jobId;private Map<String, String> params = new HashMap<>();public void addParam(String key, String value) {params.put(key, value);}public String getParam(String key) {return params.get(key);}
}

接著，我們創建這個YangJobExecutor的實現類，用于測試，在該類中，執行任務的方法很簡單，打印當前類的名字以及入參。

package com.yang.task;import com.yang.job.execute.IYangJobExecutor;
import com.yang.job.execute.YangJobExecuteRequest;import java.util.Date;public class TestJobExecutor implements IYangJobExecutor {@Overridepublic void execute(YangJobExecuteRequest yangJobExecuteRequest) {System.out.println(String.format("%s 任務執行類執行了，入參為：%s, 當前時間:%s",this.getClass().getName(), yangJobExecuteRequest.toString(),new Date().toString()));}
}

然后我們創建一個YangJobData，假設我們從數據庫中獲取的數據格式如下：

package com.yang.job.data;import lombok.Data;import java.io.Serializable;@Data
public class YangJobData implements Serializable {private Integer jobId;private String cron;private String executeStrategy;private String executeClassPath;private String executeParams;
}

executeStrategy表示任務的執行策略，executeClassPath表示要執行的任務類的路徑，executeParams表示執行任務方法的入參。
在XXL-JOB中，我們可以使用cron來設置定時任務的執行時間，因此我們這里，也使用cron作為定時任務的執行時間設置，為了解析cron表達式，我們添加下列依賴：

  <dependency><groupId>com.cronutils</groupId><artifactId>cron-utils</artifactId><version>9.2.0</version></dependency>

然后創建一個CronUtils工具類，用于解析cron表達式。

package com.yang.demo.infra.utils;import com.cronutils.model.CronType;
import com.cronutils.model.definition.CronDefinition;
import com.cronutils.model.definition.CronDefinitionBuilder;
import com.cronutils.model.time.ExecutionTime;
import com.cronutils.parser.CronParser;import java.time.ZonedDateTime;
import java.util.Optional;public class CronUtils {private static final CronDefinition CRON_DEFINITION = CronDefinitionBuilder.instanceDefinitionFor(CronType.QUARTZ);private static final CronParser CRON_PARSER = new CronParser(CRON_DEFINITION);public static ZonedDateTime nextExecutionTime(String cron, ZonedDateTime startTime) {ExecutionTime executionTime = ExecutionTime.forCron(CRON_PARSER.parse(cron));Optional<ZonedDateTime> zonedDateTime = executionTime.nextExecution(startTime);return zonedDateTime.get();}
}

對于執行方法的入參，一般情況下，就是任務的id，以及一些擴展信息，這些擴展信息一般以鍵值對的形式存儲，即"key:value;key:value;"這些形式，所以這里添加一個FeaturesUtils類，用于解析這些鍵值對信息：

package com.yang.job.utils;import java.util.HashMap;
import java.util.Map;public class FeaturesUtils {private final static String KEY_KEY_SEPARATOR = ";";private final static String KEY_VALUE_SEPARATOR = ":";public static Map<String, String> convert2FeatureMap(String features) {Map<String, String> featureMap = new HashMap<>();if (features == null || features.isEmpty()) {return featureMap;}String[] keyValues = features.split(KEY_KEY_SEPARATOR);for (String keyValue : keyValues) {String[] split = keyValue.split(KEY_VALUE_SEPARATOR);String key = split[0];String value = split[1];featureMap.put(key, value);}return featureMap;}public static String convert2Features(Map<String, String> featureMap) {if (featureMap == null || featureMap.isEmpty()) {return "";}StringBuilder stringBuilder = new StringBuilder();featureMap.forEach((key, value) -> {stringBuilder.append(key).append(KEY_VALUE_SEPARATOR).append(value).append(KEY_KEY_SEPARATOR);});return stringBuilder.toString();}
}

然后我們添加測試方法，模擬從數據庫中獲取數據，并根據任務類路徑，獲取對應的runnable并提交到YangJobManager中。

  public static void main(String[] args) {YangJobData yangJobData = mockYangJobData();YangJobSubmitParam yangJobSubmitParam = convert2YangJobSubmitParam(yangJobData);YangJobManager yangJobManager = new YangJobManager.YangJobManagerBuilder().setThreadFactory(new YangJobThreadFactory("yang")).build();yangJobManager.submitJob(yangJobSubmitParam);try {Thread.sleep(20000);} catch (InterruptedException e) {throw new RuntimeException(e);}yangJobManager.shutdown();}private static YangJobSubmitParam convert2YangJobSubmitParam(YangJobData yangJobData) {YangJobSubmitParam yangJobSubmitParam = new YangJobSubmitParam();yangJobSubmitParam.setJobId(yangJobData.getJobId());yangJobSubmitParam.setJobExecuteStrategy(JobExecuteStrategyEnum.getJobExecuteStrategyByName(yangJobData.getExecuteStrategy()));ZonedDateTime nextExecutionTime = CronUtils.nextExecutionTime(yangJobData.getCron(), ZonedDateTime.now());ZonedDateTime nextNextExecutionTime = CronUtils.nextExecutionTime(yangJobData.getCron(), nextExecutionTime);long nowEochMill = ZonedDateTime.now().toInstant().toEpochMilli();long executeEochMill = nextExecutionTime.toInstant().toEpochMilli();long secondExecuteEochMill = nextNextExecutionTime.toInstant().toEpochMilli();yangJobSubmitParam.setInitialDelay((int)(executeEochMill - nowEochMill) / 1000);yangJobSubmitParam.setPeriod((int)(secondExecuteEochMill - executeEochMill) / 1000);try {Class<?> aClass = Class.forName(yangJobData.getExecuteClassPath());if (!IYangJobExecutor.class.isAssignableFrom(aClass)) {throw new RuntimeException("任務類必須實現IYangJobExecutor接口");}IYangJobExecutor executor = (IYangJobExecutor) aClass.newInstance();YangJobExecuteRequest yangJobExecuteRequest = convert2YangJobExecuteRequest(yangJobData);Runnable runnable = () -> executor.execute(yangJobExecuteRequest);yangJobSubmitParam.setRunnable(runnable);} catch (InstantiationException | IllegalAccessException e) {e.printStackTrace();} catch (ClassNotFoundException e) {e.printStackTrace();}return yangJobSubmitParam;}private static YangJobExecuteRequest convert2YangJobExecuteRequest(YangJobData yangJobData) {YangJobExecuteRequest yangJobExecuteRequest = new YangJobExecuteRequest();yangJobExecuteRequest.setJobId(yangJobData.getJobId().toString());yangJobExecuteRequest.setParams(FeaturesUtils.convert2FeatureMap(yangJobData.getExecuteParams()));return yangJobExecuteRequest;}private static YangJobData mockYangJobData() {YangJobData yangJobData = new YangJobData();yangJobData.setJobId(1);yangJobData.setCron("0/5 * * * * ?");yangJobData.setExecuteStrategy(JobExecuteStrategyEnum.WITH_FIXED_DELAY.getName());yangJobData.setExecuteClassPath("com.yang.task.TestJobExecutor");yangJobData.setExecuteParams("jobId:1;startIndex:1;endIndex:10;");return yangJobData;}

這里對于cron的解析，其實不是特別好，這里的思路是，獲取下一次執行的時間，和下下一次執行的時間，然后以此來計算initialDelay和period，但是如果這個cron表示的是某幾天、某幾個小時，比如說星期一、星期二、星期三執行，那么我們那種解析方式是有誤的，這個可以后續再好好斟酌一下，目前先這樣解析。
執行結果如下：