本文项目Github地址：https://github.com/zhouhuanghua/z-job

什么是任务调度平台呢？暂时不做解释，先来看一下定时器的发展历史吧！

首先，new Thread + while (true) + Thread.sleep的方式，虽然很low但是起码能够实现对吧——这种方式的问题是过于占用资源，定时任务一多就暴露出来了。

然后，就是利用一些框架，比如JDK提供的定时器工具Timer和线程池ScheduledThreadPoolExecutor，Netty中基于时间轮实现的HashedWheelTimer，还有大名鼎鼎的Quartz等，在代码里面写死了执行计划——显然，这种方式的问题是管理困难，有哪些定时任务在跑，执行计划是什么，要想知道得去刨代码。

接着，基于第二种方式，将执行计划和任务开关写到配置文件里面，达到了统一维护的目的——但是，修改执行计划、任务启用关闭，需要修改配置文件然后重新启动应用，相当麻烦（不过，对于一些小型项目这种方式已经够用，毕竟容易上手且成本低）。

最后，就是基于第二种方式，将定时任务以及相关的执行计划、开关都保存到数据库，然后提供一个可视化的界面，给用户在线修改。实现方式就是，以某种标准在代码里面写好业务代码，当用户操作时才将它们加入到定时任务里面，或者从定时任务里移除，应用启动时也会自动将数据库中启用状态的定时任务注册——这种方式已经很灵性了。

只是，伴随着系统的流量快速增长，大型网站对高并发高可用的要求逐渐提高，于是衍生出了集群这么一个东西。何谓"集群"，比如一个Tomcat的最大并发是500，但是网站的QPS是1000，这谁顶得住啊？于是需要再加一个Tomcat，通过负载均衡将流量分担，这两个Tomcat就形成了集群。注意与系统拆分的区别：一组集群里面的每个机器跑的应用是一样的。

那么，问题来了：假如我的应用里有定时任务，那岂不是集群里面每台机器都在跑，这肯定不行的！怎么解决呢？通过单独设置机器上面应用的配置，只让其中一台跑。这，当然也可以，不过相当麻烦，而且负载均衡下管理界面怎么解决（聪明的你们也许有办法）。还有另外一个，微服务：总不能每个子应用设置一个管理界面吧？

扯了一大推，好像有点废话了。那，什么是任务调度平台呢？

任务——调度——平台：通过一个平台，对所有机器上面的定时任务进行统一管理。不懂的话看图

(暂不提风靡全国、功能强大的轻量级分布式任务调度平台xxl-job了哈，只说我发明的轮子。手动滑稽)

其实，还是有必要说明一下滴：首先，需要在具体应用上面配置任务调度平台的地址信息和自己的信息，当应用启动时，将自己的应用名称、IP及端口发送给任务调度平台，后者保存到数据库。然后在任务调度平台手动添加任务信息，需要选择所属的应用并填写任务名称(具体应用里面实现了IJobHandler的类并注入到Spring容器后的beanName)、执行计划的CRON、告警邮件等信息。接着任务调度平台会根据这个任务的名称、所属应用名称、执行计划创建一个定时作业(Quartz的Job)，并将任务信息以及它的应用信息传递进去。最后任务调度平台的定时作业执行的时候，会拿出传递进来的任务信息和它的应用信息，根据应用信息的机器地址发送HTTP请求，请求参数为任务名称以及运行参数，具体的应用收到请求后执行对应的JobHandler并将结果返回，任务调度平台收到结果(如果返回执行失败或者异常，并且任务信息配了尝试次数、应用信息里还有其它机器地址的话，就会进行重试，始终不成功的话发送告警邮件)后，将这一个过程写入任务调度记录里面。

要不"镇楼图"先来一波！?

---首页

---应用列表

---任务列表。还有更多信息没展示，后期考虑做个详情页面

---任务调度日志。调度结果或者任务执行结果为失败时，鼠标放到提示的位置会展示失败原因

---发送的告警邮件。内容是任务调度的记录详情

下面我们就结合代码看一下开发步骤吧（基于Spring Boot的哦）

一、项目结构

一共分为三个模块(这里是为了方便开发，实际应用时需要分成不同的项目)。

z-job-core：核心模块。
z-job-admin：任务调度平台，依赖z-job-core模块。
z-job-example：使用z-job的一个示例应用，依赖z-job-core模块即可。

下面我们就逐一模块进行讲解。

二、z-job-core模块

这个模块是做什么的呢？你要使用一个框架，需要引入它的一些依赖，按照它的一些标准开发代码吧？z-job-core就是充当这么一个角色。它的作用有：

提供一个注解，达到Spring Boot自动配置、信息（当前应用的名称和描述、z-job-admin平台的IP和端口）写在属性上面的目的。
提供开发定时任务时需要遵守的标准，即实现给定的接口。
统一维护当前应用的所有定时任务。
与z-job-admin平台的交互(重点)：即应用启动时将地址信息注册到z-job-admin平台、接收来自z-job-admin平台的任务调用请求并返回任务执行结果。

1、约定每个任务需要遵守的标准

我们定义标准的是接口，因为这样可以确定方法的签名信息。此外，还需要这个类使用@Component注解将自己注入Spring容器，方便后续收集定时任务实例，并以beanName作为任务名称(自带唯一标识功能)。

入参：z-job-admin平台发送过来的任务执行参数。
返回：JobInvokeRsp，包含int类型的执行是否成功（code）和String类型的说明（msg）。

/*** 任务处理器接口** @author z_hh*/
public interface IJobHandler {JobInvokeRsp execute(String params) throws Exception;}

2、加载配置信息

为了方便和避免用户忘记，我们定义一个所有属性都非空的注解，用在Spring Boot启动类上（而且，仅当该注解存在时，才会开启z-job并加载相关数据）。

adminIp：z-job-admin平台的IP地址
adminPort：z-job-admin平台的端口
appName：当前应用的名称，在z-job-admin平台里面表示此应用的唯一标识
appDesc：当前应用的描述信息

为了达到应用的启动类存在EnableJobAutoConfiguration注解时才加载z-job的目的，所以用Import注解的方式将两个类注册到Spring容器。

/*** 开启任务自动配置** @author z_hh*/
@Retention(RetentionPolicy.RUNTIME)
@Target({ElementType.TYPE})
@Documented
@Import({JobAutoConfigurationRegistrar.class, JobConfiguration.class})
public @interface EnableJobAutoConfiguration {String adminIp();int adminPort();String appName();String appDesc();
}

1）@EnableJobAutoConfiguration第一个Import的是JobAutoConfigurationRegistrar：该类用于读取注解上面的配置信息，并手动注册两个bean到Spring容器。这两个bean是

JobExecutor：这个类的作用是，应用启动时将地址信息注册到z-job-admin平台、统一维护应用的所有定时任务实例、运行指定定时任务。注册的同时将注解类的配置信息设置进去，并且执行init方法初始化。
JobInvokeServletRegistrar：这个类实际上是一个Servlet，用于接收来自z-job-admin平台的调用任务请求，并且返回任务执行结果。注册时使用newInstance工厂方法，并将jobExecutor这个bean注入。(为啥不直接定义一个Controller？因为控制不了当@EnableJobAutoConfiguration存在时才注册到Spring容器的目的，条件注解都没有用)。

/*** ImportBeanDefinitionRegistrar** @author z_hh*/
@Slf4j
public class JobAutoConfigurationRegistrar implements ImportBeanDefinitionRegistrar, EnvironmentAware {@Setterprivate Environment environment;@Overridepublic void registerBeanDefinitions(AnnotationMetadata annotationMetadata, BeanDefinitionRegistry beanDefinitionRegistry) {AnnotationAttributes annotationAttributes = AnnotationAttributes.fromMap(annotationMetadata.getAnnotationAttributes(EnableJobAutoConfiguration.class.getName()));if (Objects.isNull(annotationAttributes)) {log.error("【任务调度平台】EnableJobAutoConfiguration annotationAttributes is null!");return;}// 注册JobExecutorregisterJobExecutor(annotationAttributes, beanDefinitionRegistry);// 注册ServletregisterJobInvokeServletRegistrationBean(beanDefinitionRegistry);}private void registerJobExecutor(AnnotationAttributes annotationAttributes, BeanDefinitionRegistry beanDefinitionRegistry) {// 创建配置实例JobProperties jobProperties = new JobProperties();jobProperties.setAdminIp(annotationAttributes.getString("adminIp"));jobProperties.setAdminPort(annotationAttributes.getNumber("adminPort"));jobProperties.setAppName(annotationAttributes.getString("appName"));jobProperties.setAppDesc(annotationAttributes.getString("appDesc"));jobProperties.setIp(NetUtil.getIp());jobProperties.setPort(environment.getProperty("server.port", Integer.class, 8080));AbstractBeanDefinition beanDefinition = BeanDefinitionBuilder.genericBeanDefinition(JobExecutor.class).setInitMethodName("init").setDestroyMethodName("destroy").addPropertyValue("jobProperties", jobProperties).setAutowireMode(AbstractBeanDefinition.AUTOWIRE_BY_TYPE).getBeanDefinition();beanDefinitionRegistry.registerBeanDefinition("jobExecutor", beanDefinition);log.info("【任务调度平台】JobExecutor register success!");}private void registerJobInvokeServletRegistrationBean(BeanDefinitionRegistry beanDefinitionRegistry) {AbstractBeanDefinition beanDefinition = BeanDefinitionBuilder.genericBeanDefinition(JobInvokeServletRegistrar.class).setFactoryMethod("newInstance").addPropertyReference("jobExecutor", "jobExecutor").setAutowireMode(AbstractBeanDefinition.AUTOWIRE_NO).getBeanDefinition();beanDefinitionRegistry.registerBeanDefinition("JobInvokeServlet", beanDefinition);log.info("【任务调度平台】JobInvokeServletRegistrar register success!");}}

JobExecutor类的代码如下，init方法会做两件事

初始化所有JobHandler：将Spring容器里面所有实现了IJobHandler接口的bean取出来，然后放到一个Map里面，并以beanName为键。
将自己注册到调度中心：新起一个线程，使用注入的restTemplate，根据注解配置的z-job-admin平台地址信息，将当前应用信息（应用名称和IP端口等）发送到任务调度平台进行应用的自动注册。那边有专门的接口会做灵活处理（针对集群现象），后面会说。

/*** 任务执行器** @author z_hh*/
@Slf4j
public class JobExecutor {@Setterprivate JobProperties jobProperties;@Autowiredprivate RestTemplate restTemplate;@Autowiredprivate ApplicationContext applicationContext;private ConcurrentHashMap<String, IJobHandler> jobHandlerRepository = new ConcurrentHashMap();public IJobHandler registJobHandler(String name, IJobHandler jobHandler) {log.info("【任务调度平台】成功注册JobHandler >>>>>>>>>> name={}，handler={}", name, jobHandler.getClass().getName());return (IJobHandler)jobHandlerRepository.put(name, jobHandler);}public IJobHandler loadJobHandler(String name) {return (IJobHandler)jobHandlerRepository.get(name);}public void init() {log.info("【任务调度平台】JobExecutor init...");// 初始化所有JobHandlerinitJobHandler();// 将自己注册到调度中心new RegisterAppToAdminThread().start();}public void destroy() {log.info("【任务调度平台】JobExecutor destroy...");}public JobInvokeRsp jobInvoke(String name, String params) {IJobHandler jobHandler = jobHandlerRepository.get(name);if (Objects.isNull(jobHandler)) {return JobInvokeRsp.error("任务不存在！");}try {return jobHandler.execute(params);} catch (Exception e) {log.error("【任务调度平台】任务{}调用异常：{}", name, e);return JobInvokeRsp.error("任务调用异常！");}}private void initJobHandler() {String[] beanNames = applicationContext.getBeanNamesForType(IJobHandler.class);if (beanNames == null || beanNames.length == 0) {return;}Arrays.stream(beanNames).forEach(beanName -> {registJobHandler(beanName, (IJobHandler)applicationContext.getBean(beanName));});}private class RegisterAppToAdminThread extends Thread {private  RegisterAppToAdminThread() {super("AppToAdmin-T");}@Overridepublic void run() {log.info("【任务调度平台】开始往调度中心注册当前应用信息...");Map<String, Object> paramMap = new HashMap<>(4);paramMap.put("appName", jobProperties.getAppName());paramMap.put("appDesc", jobProperties.getAppDesc());paramMap.put("address", jobProperties.getIp() + ":" + jobProperties.getPort());try {restTemplate.postForObject("http://" + jobProperties.getAdminIp() + ":"+ jobProperties.getAdminPort() + "/api/job/app/auto_register",paramMap,Object.class);log.info("【任务调度平台】应用注册到调度中心成功！");} catch (Throwable t) {log.warn("【任务调度平台】应用注册到调度中心失败：{}", ThrowableUtils.getThrowableStackTrace(t));}}}}

JobInvokeServletRegistrar类的代码如下，继承ServletRegistrationBean类并提供newInstance的工厂方法用于创建Servlet实例注入到Spring容器。它将接收来自z-job-admin平台的任务调用请求，并根据参数的任务名称和执行参数利用JobExecutor去调用对应的定时任务执行，最后将结果写入相应。

/*** 任务调度Servlet** @author z_hh*/
@Slf4j
public class JobInvokeServletRegistrar<JobInvokeServlet> extends ServletRegistrationBean {@Setterprivate JobExecutor jobExecutor;public JobInvokeServletRegistrar() {super();}public static JobInvokeServletRegistrar newInstance() {JobInvokeServletRegistrar jobInvokeServletRegistrar = new JobInvokeServletRegistrar();jobInvokeServletRegistrar.setServlet(jobInvokeServletRegistrar.new JobInvokeServlet());jobInvokeServletRegistrar.setUrlMappings(Collections.singletonList("/api/job/invoke"));jobInvokeServletRegistrar.setLoadOnStartup(1);return jobInvokeServletRegistrar;}private class JobInvokeServlet extends HttpServlet {@Overrideprotected void service(HttpServletRequest req, HttpServletResponse resp) throws IOException {resp.setHeader("Content-Type", "application/json");try {Pair reqAndRsp = run(req, resp);log.info("【任务调度平台】执行作业：req={}，rsp={}", reqAndRsp.getKey(), reqAndRsp.getValue());} catch (Throwable t) {String msg = ThrowableUtils.getThrowableStackTrace(t);log.warn("任务调用异常：{}", msg);String rspStr = JsonUtils.writeValueAsString(JobInvokeRsp.error(msg));resp.getOutputStream().write(rspStr.getBytes(Charset.defaultCharset()));}}private Pair run(HttpServletRequest req, HttpServletResponse resp) throws Throwable {// 反序列化ServletInputStream inputStream = req.getInputStream();byte[] body = new byte[req.getContentLength()];inputStream.read(body);JobInvokeReq jobInvokeReq = (JobInvokeReq)SerializationUtils.deserialize(body);// 调用任务JobInvokeRsp jobInvokeRsp = JobInvokeServletRegistrar.this.jobExecutor.jobInvoke(jobInvokeReq.getName(), jobInvokeReq.getParams());// 响应结果String rspStr = JsonUtils.writeValueAsString(jobInvokeRsp);resp.getOutputStream().write(rspStr.getBytes(Charset.defaultCharset()));// 返回请求和响应提供日志记录return new Pair(jobInvokeReq, jobInvokeRsp);}}
}

2）@EnableJobAutoConfiguration第二个Import的是JobConfiguration：它目前只有一个作用，当Spring容器不存在RestTemplate实例时，就创建一个注册进去。

/*** 任务配置类** @author z_hh*/
@Slf4j
public class JobConfiguration {{log.info("【任务调度平台】Loading JobAutoConfiguration!");}@Bean@ConditionalOnMissingBeanpublic RestTemplate restTemplate() {return new RestTemplate();}
}

此外，前面没说到的类还有几个，其中获取当前机器IP地址的工具代码如下

/*** 网络工具类** @author z_hh*/
@Slf4j
public class NetUtil {private NetUtil() {}public static String getIp() {try {return InetAddress.getLocalHost().getHostAddress();} catch (UnknownHostException e) {log.error("获取IP地址异常：", ThrowableUtils.getThrowableStackTrace(e));throw new RuntimeException("获取IP地址异常！");}}}

其它的话，可以去看对应的源码。

三、z-job-admin模块

1、表结构，以及对应的增删改查

表定义了3张，开始曾经想过：应用的信息能否直接放在任务里面，后来觉得实在不妥，不太方便管理（参考过xxl-job的表哈，它们的也是这样的，嘻嘻）。所以，3张表就是：任务应用表、任务信息表、任务调度记录表。

1）任务应用表

CREATE TABLE `z_job_app` (`id` int(11) NOT NULL AUTO_INCREMENT,`app_name` varchar(64) NOT NULL COMMENT '应用名称',`app_desc` varchar(128) NOT NULL COMMENT '应用描述',`creator` varchar(32) NOT NULL COMMENT '创建人',`create_time` datetime NOT NULL COMMENT '创建时间',`create_way` tinyint(1) NOT NULL COMMENT '创建方式：1-自动，2-手工',`update_time` datetime DEFAULT NULL COMMENT '最后更新时间',`address_list` varchar(512) NOT NULL COMMENT '应用地址列表，多个逗号分隔',`enabled` tinyint(1) NOT NULL COMMENT '启用状态：1-启用，0-停用',`is_deleted` tinyint(1) NOT NULL COMMENT '是否删除：1-是，0-否',PRIMARY KEY (`id`),UNIQUE KEY `uk_app_name` (`app_name`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

2）任务信息表

CREATE TABLE `z_job_info` (`id` int(11) NOT NULL AUTO_INCREMENT,`job_app_id` int(11) NOT NULL COMMENT '任务所属应用id',`job_name` varchar(64) NOT NULL COMMENT '任务名称',`job_desc` varchar(512) DEFAULT '' COMMENT '任务描述',`alarm_email` varchar(512) DEFAULT '' COMMENT '报警邮件，多个逗号分隔',`creator` varchar(32) NOT NULL COMMENT '创建人',`create_time` datetime NOT NULL COMMENT '创建时间',`create_way` tinyint(1) NOT NULL COMMENT '创建方式：1-自动，2-手工',`update_time` datetime DEFAULT NULL COMMENT '最后更新时间',`run_cron` varchar(128) NOT NULL COMMENT '任务执行CRON',`run_strategy` tinyint(1) NOT NULL COMMENT '任务执行策略：1-随机，2-轮询',`run_param` varchar(512) DEFAULT '' COMMENT '任务执行参数',`run_timeout` smallint(3) NOT NULL COMMENT '任务执行超时时间，单位秒',`run_fail_retry_count` smallint(3) NOT NULL COMMENT '任务执行失败重试次数',`trigger_last_time` datetime DEFAULT NULL COMMENT '上次调度时间',`trigger_next_time` datetime DEFAULT NULL COMMENT '下次调度时间',`enabled` tinyint(1) NOT NULL COMMENT '启用状态：1-启用，0-停用',`is_deleted` tinyint(1) NOT NULL COMMENT '是否删除：1-是，0-否',PRIMARY KEY (`id`),UNIQUE KEY `uk_name_appid` (`job_name`,`job_app_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

3）任务调度记录表

CREATE TABLE `z_job_log` (`id` int(11) NOT NULL AUTO_INCREMENT,`job_id` int(11) NOT NULL COMMENT '任务ID',`run_address_list` varchar(512) NOT NULL COMMENT '本次运行的地址',`run_fail_retry_count` smallint(3) NOT NULL COMMENT '任务执行失败重试次数',`trigger_start_time` datetime NOT NULL COMMENT '调度开始时间',`trigger_end_time` datetime NOT NULL COMMENT '调度结束时间',`trigger_result` tinyint(1) NOT NULL COMMENT '调度结果：1-成功，0-失败',`trigger_msg` varchar(3000) DEFAULT '' COMMENT '调度日志',`job_run_result` tinyint(1) DEFAULT '0' COMMENT '任务执行结果：1-成功，0-失败',`job_run_msg` varchar(3000) DEFAULT '' COMMENT '任务执行日志',PRIMARY KEY (`id`),KEY `idx_job_id` (`job_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;

4）增删改查

使用spring-data-jpa，因为简单。该技术不在本文讨论范围内，你也可以使用Mybatis、甚至JDBC。

2、任务应用的自动注册

前面z-job-core模块说过，应用启动时会将自己的信息注册到任务调度平台，这边提供接口，并且会灵活处理。怎么个灵活法呢？君请看，会判断应用传过来的应用名称和地址信息

应用名称不存在，直接插入。
应用名称存在，手动新增的话提示重复，否则：如果地址也存在，忽略；如果地址不存在，合并地址。

PS：Controller层面需要加一个独占锁保证并发时线程安全。

public Result<JobApp> insert(JobApp jobApp) {if (Objects.nonNull(jobApp.getId())) {jobApp.setId(null);}// 根据appName查询数据JobApp exampleObj = new JobApp();exampleObj.setAppName(jobApp.getAppName());exampleObj.setIsDeleted(IsDeletedEnum.NO.getCode());Example<JobApp> example = Example.of(exampleObj);Optional<JobApp> JobAppOptional = dao.findOne(example);// 如果不存在，直接保存if (!JobAppOptional.isPresent()) {return Result.ok(save(jobApp));}// 如果是手动添加，提示重复if (Objects.equals(jobApp.getCreateWay(), CreateWayEnum.MANUAL.getCode())) {return Result.err("应用名称已存在！");}// 比较地址JobApp existsJobApp = JobAppOptional.get();List<String> addressList = Arrays.stream(existsJobApp.getAddressList().split(",")).filter(StringUtils::hasText).collect(Collectors.toList());// 存在，但是地址已经包含，忽略if (addressList.contains(jobApp.getAddressList())) {return Result.ok(existsJobApp);}// 存在，但是地址还没包含，合并地址addressList.add(jobApp.getAddressList());String newAddressList = addressList.stream().reduce((s1, s2) -> s1 + "," + s2).orElse("");existsJobApp.setAddressList(newAddressList);return Result.ok(save(existsJobApp));}

3、Quartz Job的注册以及移除(重点)

1）开始介绍z-job任务调度平台架构的时候说过，使用Quartz作为定时器的。在Spring Boot下怎么使用呢？

引入依赖并注册一个Scheduler的bean到Spring容器之后，就可以在需要的地方注入并使用了。

        <!--  作业调度框架Quartz  --><dependency><groupId>org.quartz-scheduler</groupId><artifactId>quartz</artifactId><version>${quartz-scheduler-version}</version></dependency><dependency><groupId>org.quartz-scheduler</groupId><artifactId>quartz-jobs</artifactId><version>${quartz-scheduler-version}</version></dependency>

@Beanpublic Scheduler scheduler() throws SchedulerException {return StdSchedulerFactory.getDefaultScheduler();}

2）新增任务或者将任务从停用改为启用时都需要注册一个定时作业。除此之外，任务调度平台启动的时候也需要将启用状态的任务进行注册。当任务停用的时候，需要将对应的定时作业移除。

定时作业：所有的Quartz Job都使用一个类相同的逻辑，执行的时候根据传递进来的任务信息和它的应用信息，通过JobInvoker调度器发起远程调用，如果失败了会根据任务的重试机制进行对应的处理，终究没成功的话会发送告警邮件，最后将整个调度过程插入到任务调度记录里，并修改任务的下一次调度时间。

/*** Quartz任务** @author z_hh*/
@Slf4j
public class QuartzJob implements Job {private static final byte SUCCESS = 1;private static final byte ERROR = 0;private JobInvoker jobInvoker;private MailSendService mailSendService;public QuartzJob() {this.jobInvoker = BeanUtils.getBean(JobInvoker.class);this.mailSendService = BeanUtils.getBean(MailSendService.class);}@Overridepublic void execute(JobExecutionContext jobExecutionContext) throws JobExecutionException {JobDataMap jobDataMap = jobExecutionContext.getJobDetail().getJobDataMap();JobApp jobApp = (JobApp) jobDataMap.get("JobApp");JobInfo jobInfo = (JobInfo) jobDataMap.get("jobInfo");log.info("任务{}开始调度...", jobInfo.getJobName());JobLog jobLog = new JobLog();jobLog.setJobId(jobInfo.getId());jobLog.setTriggerStartTime(new Date());try {jobRun(jobApp, jobInfo, jobLog);jobLog.setTriggerResult((byte)1);jobLog.setTriggerMsg("调度成功！");} catch (Throwable t) {String msg = ThrowableUtils.getThrowableStackTrace(t);log.warn("任务{}调度出现异常：{}", jobInfo.getJobName(), msg);jobLog.setTriggerResult((byte)0);jobLog.setTriggerMsg("调度异常：" + msg);}jobLog.setTriggerEndTime(new Date());// 记录任务的本次和下次调用时间jobInfo.setTriggerLastTime(jobExecutionContext.getFireTime());jobInfo.setTriggerNextTime(jobExecutionContext.getNextFireTime());// 插入日志jobLog.save();// 更新任务jobInfo.save();// 发送邮件sendMail(jobApp, jobInfo, jobLog);log.info("任务{}调度结束！", jobInfo.getJobName());}private void jobRun(JobApp jobApp, JobInfo jobInfo, JobLog jobLog) {List<String> addressList = Arrays.stream(jobApp.getAddressList().split(",")).filter(StringUtils::hasText).collect(Collectors.toList());Iterator<String> iterator = addressList.iterator();JobInvokeRsp jobInvokeRsp = null;List<String> hasInvokeAddress = new ArrayList<>();int failRetryCount = jobInfo.getRunFailRetryCount(),readyRetryCount = -1;while (iterator.hasNext() && ++readyRetryCount <= failRetryCount) {String address = iterator.next();hasInvokeAddress.add(address);try {jobInvokeRsp = jobInvoker.invoke(address, jobInfo.getJobName(), jobInfo.getRunParam());if (jobInvokeRsp.isOk()) {break;}log.warn("调用{}的{}任务失败：{}", address, jobInfo.getJobName(), jobInvokeRsp.getMsg());} catch (Throwable t) {String msg = ThrowableUtils.getThrowableStackTrace(t);log.warn("调用{}的{}任务时出现异常：{}", address, jobInfo.getJobName(), msg);jobInvokeRsp = JobInvokeRsp.error("任务调用异常：" + msg);}iterator.remove();}if (Objects.isNull(jobInvokeRsp)) {jobInvokeRsp = JobInvokeRsp.error("没有进行任务调用！");}jobLog.setJobRunResult(jobInvokeRsp.getCode());jobLog.setJobRunMsg(sub3000String(jobInvokeRsp.getMsg()));jobLog.setRunFailRetryCount(Objects.equals(readyRetryCount, -1) ? 0 : readyRetryCount);jobLog.setRunAddressList(hasInvokeAddress.stream().reduce((s1, s2) -> s1 + "," + s2).orElse(""));}private void sendMail(JobApp jobApp, JobInfo jobInfo, JobLog jobLog) {// 调度失败并且邮件不为空if ((Objects.equals(jobLog.getTriggerResult(), (byte)0)|| Objects.equals(jobLog.getJobRunResult(), (byte)0))&& StringUtils.hasText(jobInfo.getAlarmEmail())) {String alarmEmailStr = jobInfo.getAlarmEmail();List<String> mailList = null;if (alarmEmailStr.contains(",")) {mailList = Arrays.asList(alarmEmailStr.split(","));} else {mailList = Collections.singletonList(alarmEmailStr);}String subject = String.format("应用%s的%s任务调度失败！", jobApp.getAppName(), jobInfo.getJobName());try {String content = new ObjectMapper().writeValueAsString(jobLog);mailList.forEach(m -> {mailSendService.sendSimpleMail(m, subject, content);});} catch (JsonProcessingException e) {log.error("任务调度失败告警邮件发送失败：{}", ThrowableUtils.getThrowableStackTrace(e));}}}private String sub3000String(String str) {if (StringUtils.hasText(str) && str.length() > 3000) {return str.substring(0, 2888) + "...更多请查看日志记录！";}return str;}
}

任务调度器：使用RestTemplate将任务名称和执行参数发送给具体的应用。

/*** 任务调度器** @author z_hh*/
@Component
public class JobInvoker {@Autowiredprivate RestTemplate restTemplate;private static final String PREFIX = "http://";private static final String PATH = "/api/job/invoke";/*** 任务调度器** @param url 目标地址* @param jobHandler 任务名称* @param params 执行参数* @return 调用任务结果*/public JobInvokeRsp invoke(String url, String jobHandler, String params) {JobInvokeReq req = new JobInvokeReq();req.setName(jobHandler);req.setParams(params);byte[] dataBytes = SerializationUtils.serialize(req);return restTemplate.postForObject(PREFIX + url + PATH, dataBytes, JobInvokeRsp.class);}
}

发送邮件：引入依赖，配置发送方邮箱，编写发送服务。

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

  # 发送邮箱配置mail:# QQ邮箱主机host: smtp.qq.com# 用户名username: xxx000@qq.com# QQ邮箱开启SMTP的授权码password: 000xxx

/*** 邮件发送服务** @author z_hh*/
@Component
public class MailSendService {@Autowiredprivate JavaMailSender mailSender;@Value("${spring.mail.username}")private String from;/*** 发送普通邮件** @param to 收件人* @param subject 邮件主题* @param content 邮件内容* @throws MailException*/public void sendSimpleMail(String to, String subject, String content) throws MailException {SimpleMailMessage message = new SimpleMailMessage();message.setFrom(from);message.setTo(to);message.setSubject(subject);message.setText(content);mailSender.send(message);}
}

注册定时作业：首先根据任务的名称、描述以及所属应用名称使用QuartzJob类构建一个JobDetail实例，然后根据任务的执行计划构建Trigger实例，接着将任务信息、应用信息传递到JobDetail实例的JobDataMap里，最后使用scheduler注册到作业调度中。

public Result register(JobInfo jobInfo) {// 获取任务组信息Result<JobApp> JobAppResult = JobAppService.getById(jobInfo.getJobAppId());if (JobAppResult.isErr()) {return JobAppResult;}JobApp JobApp = JobAppResult.get();// 创建jobDetail实例JobDetail jobDetail = JobBuilder.newJob(QuartzJob.class).withIdentity(jobInfo.getJobName(), JobApp.getAppName()).withDescription(jobInfo.getJobDesc()).build();// 定义调度触发规则cornTrigger trigger = TriggerBuilder.newTrigger().withIdentity(jobInfo.getJobName(), JobApp.getAppName()).startAt(DateBuilder.futureDate(1, DateBuilder.IntervalUnit.SECOND)).withSchedule(CronScheduleBuilder.cronSchedule(jobInfo.getRunCron())).startNow().build();// 传递一些数据到任务里面JobDataMap jobDataMap = jobDetail.getJobDataMap();jobDataMap.put("JobApp", JobApp);jobDataMap.put("jobInfo", jobInfo);// 把作业和触发器注册到任务调度中try {scheduler.scheduleJob(jobDetail, trigger);} catch (SchedulerException e) {log.error("注册任务异常：{}", ThrowableUtils.getThrowableStackTrace(e));return Result.err("注册任务异常！");}// 启动try {if (!scheduler.isShutdown()) {scheduler.start();}} catch (SchedulerException e) {log.error("启动scheduler异常：{}", ThrowableUtils.getThrowableStackTrace(e));return Result.err("启动scheduler异常！");}return Result.ok();}

移除定时作业：很简单，根据任务的名称以及它的应用名称组成一个JobKey，然后使用scheduler进行删除即可。

public Result disable(Long id) {// 查询和校验Result<JobInfo> jobInfoResult = getById(id);if (jobInfoResult.isErr()) {return jobInfoResult;}JobInfo jobInfo = jobInfoResult.get();if (Objects.equals(jobInfo.getEnabled(), EnabledEnum.NO.getCode())) {return Result.err("任务已经处于停用状态！");}// 查询对应任务组Result<JobApp> JobAppResult = JobAppService.getById(jobInfo.getJobAppId());if (JobAppResult.isErr()) {return JobAppResult;}// 从scheduler移除任务JobApp jobApp = JobAppResult.get();JobKey jobKey = JobKey.jobKey(jobInfo.getJobName(), jobApp.getAppName());try {boolean deleteJobResult = scheduler.deleteJob(jobKey);if (!deleteJobResult) {return Result.err("停用定时任务失败！");}} catch (SchedulerException e) {log.error("停用定时任务异常：{}", ThrowableUtils.getThrowableStackTrace(e));return Result.err("停用定时任务异常！");}// 修改数据状态jobInfo.setEnabled(EnabledEnum.NO.getCode());jobInfo.setUpdateTime(new Date());save(jobInfo);return Result.ok("停用定时任务成功！");}

任务调度平台启动时将启用的任务注册到作业调度中：

/*** 应用启动后注册定时任务** @author z_hh*/
@Component
@Slf4j
public class RegisterJobOnAppStart implements ApplicationListener<ApplicationReadyEvent> {@Autowiredprivate JobInfoService jobInfoService;@Overridepublic void onApplicationEvent(ApplicationReadyEvent applicationReadyEvent) {jobInfoService.queryAll().stream().filter(jobInfo ->Objects.equals(jobInfo.getIsDeleted(), IsDeletedEnum.NO.getCode())&& Objects.equals(jobInfo.getEnabled(), EnabledEnum.YES.getCode())).forEach(jobInfo -> {Result registerResult = jobInfoService.register(jobInfo);if (registerResult.isErr()) {log.error("任务[id={}，jobName={}]注册失败：{}", jobInfo.getId(), jobInfo.getJobName(), registerResult.getMsg());}log.info("任务[id={}，jobName={}]注册成功！", jobInfo.getId(), jobInfo.getJobName());});}
}

好像没有特别重要的了，其它的话找源码看一下吧?

四、z-job-example模块

一切就绪之后，就可以编写一个示例爽一把了，这是最开心的时刻。

1、引入z-job-core的依赖

<dependency><groupId>cn.zhh</groupId><artifactId>z-job-core</artifactId><version>1.0-SNAPSHOT</version></dependency>

2、启动类加上EnableJobAutoConfiguration注解并配置任务调度平台的IP和端口，以及当前应用的名称和描述

@EnableJobAutoConfiguration(adminIp = "127.0.0.1",adminPort = 8888,appName = "example",appDesc = "示例应用")

3、编写一个定时任务类，实现IJobHandler接口并重写execute方法，加上Component注解

/*** 示例任务1** @author z_hh*/
@Component
public class JobExample1 implements IJobHandler {@Overridepublic JobInvokeRsp execute(String params) throws Exception {return JobInvokeRsp.success("我执行成功啦！收到参数：" + params);}
}

4、首先启动z-job-admin项目，然后启动项目，看到控制台输出。应用自动注册到任务调度平台

5、任务调度平台添加对应的任务信息。默认启用状态，它会自动注册到Quartz作业调度中

6、时间到了之后会执行任务调度

7、查看调度日志

至此，z-job整个核心流程已经开发完成了。

五、前端技术

前端使用的主要框架有Bootstrap、JQuery以及Vue。不是我擅长的领域，你们看一下代码就好。

六、项目涉及的其它技术

除了介绍核心代码以外，还有一些个人觉得是亮点的分享一下。

1、定义通用结果返回对象，并使用Aspect切面处理带来的事务问题。

一般是Controller层使用的。如果是Service层使用，就要考虑事务回滚的问题（因为Spring的Transactional是默认抛出RuntimeException时才触发事务回滚的），一般推荐抛出自定义业务异常并结合统一异常处理器进行转化处理。

/*** 通用结果返回对象** @author z_hh*/
@ToString
public class Result<T> implements Serializable {private static final long serialVersionUID = 6547662806723050209L;private static final int SUCCESS = 200;private static final int ERROR = 500;@Getterprivate Integer code;@Getterprivate String msg;@Getterprivate T content;private Result(Integer code, String msg, T content) {this.code = code;this.msg = msg;this.content = content;}public static <T> Result<T> ok() {return new Result<>(SUCCESS, null, (T)null);}public static <T> Result<T> ok(String msg) {return new Result<>(SUCCESS, msg, (T)null);}public static <T> Result<T> ok(T content) {return new Result<>(SUCCESS, null, content);}public static <T> Result<T> ok(String msg, T content) {return new Result<>(SUCCESS, msg, content);}public static <T> Result<T> err() {return new Result<>(ERROR, null, (T)null);}public static <T> Result<T> err(String msg) {return new Result<>(ERROR, msg, (T)null);}public boolean isOk() {return Objects.equals(this.code, SUCCESS);}public boolean isErr() {return Objects.equals(this.code, ERROR);}public T get() {if (isErr()) {throw new UnsupportedOperationException("result is error!");}return this.content;}
}

如果Service层使用，并需要返回Result.ok() == false时进行事务回滚，那么就需要结合Aspect切面进行手工回滚处理了。

/*** 对Spring的事务注解@Transactional做进一步处理，* 结合Service的返回值类型Result，做出是否启动事务回滚** @author z_hh*/
@Aspect
@Component
public class TransactionalAspect {@Around(value = "@annotation(org.springframework.transaction.annotation.Transactional)&&@annotation(transactional)")public Object verify(ProceedingJoinPoint pjp, Transactional transactional) throws Throwable {// 执行切面方法，获得返回值Object result = pjp.proceed();// 检测&强行回滚boolean requireRollback = requireRollback(result);if (requireRollback) {TransactionAspectSupport.currentTransactionStatus().setRollbackOnly();}// 返回切面方法运行结果return result;}private static boolean requireRollback(Object result) throws Exception {// 如果方法返回值不是Result对象，则不需要回滚if (!(result instanceof Result)) {return false;}// 如果result.isOk() == true，也不需要回滚Result r = (Result) result;if (!r.isOk()) {return false;}// 如果@Transactional启用了新事物(propagation = Propagation.REQUIRES_NEW)，需要回滚boolean isNewTransaction = TransactionAspectSupport.currentTransactionStatus().isNewTransaction();if (isNewTransaction) {return true;}// 如果方法没有被其它@Transactional注释的方法嵌套调用，说明该线程的事物已运行完毕，则需要回滚//  此处使用了较多的反射底层语法，强行访问Spring内部的private/protected 方法、字段，存在一定的风险Object currentTransactionInfo = executePrivateStaticMethod(TransactionAspectSupport.class, "currentTransactionInfo"),oldTransactionInfo = getPrivateFieldValue(currentTransactionInfo, "oldTransactionInfo");if (oldTransactionInfo == null) {return true;}// 其它情况，不回滚return false;}private static Object getPrivateFieldValue(Object target, String fieldName) throws NoSuchFieldException, SecurityException, IllegalArgumentException, IllegalAccessException {Field field = target.getClass().getDeclaredField(fieldName);field.setAccessible(true);return field.get(target);}private static Object executePrivateStaticMethod(Class<?> targetClass, String methodName) throws NoSuchMethodException, SecurityException, IllegalAccessException, IllegalArgumentException, InvocationTargetException {Method method = targetClass.getDeclaredMethod(methodName);method.setAccessible(true);return method.invoke(null);}
}

2、Swagger2的正确使用姿势

怎么使用我就不说了，就讲下怎么样可以让Controller看起来舒服点。处理方式就是，每个Controller定义一个Api接口，将注解信息写在接口上面。如

/*** 任务信息API** @author z_hh*/
@Api(tags = "任务信息API")
public interface JobInfoApi {@ApiOperation("分页查询任务")@GetMapping("/page_query")public Result<Page<JobInfoPageQueryRsp>> pageQuery(@RequestParam(required = false, defaultValue = "1") Integer pageNum,@RequestParam(required = false, defaultValue = "10") Integer pageSize);
}

/*** 任务信息控制器** @author z_hh*/
@RestController
@RequestMapping("/job/info")
@Slf4j
public class JobInfoController implements JobInfoApi {@Autowiredprivate JobInfoService jobInfoService;@Overridepublic Result<Page<JobInfoPageQueryRsp>> pageQuery(@RequestParam(required = false, defaultValue = "1") Integer pageNum,@RequestParam(required = false, defaultValue = "10") Integer pageSize) {Result<Page<JobInfoPageQueryRsp>> pageResult = jobInfoService.queryByPage(pageNum, pageSize);return pageResult;}
}

3、利用Aspect切面优雅对validation的请求参数进行校验

如果是在Controller层进行校验，那么可以直接在方法的参数前面加@Valid注解，然后定义全局统一异常处理器处理MethodArgumentNotValidException即可。这里是给假如要在Service层做校验的提供一种思路。

我的相关博客：https://blog.csdn.net/qq_31142553/article/details/89430100、https://blog.csdn.net/qq_31142553/article/details/86547201、https://blog.csdn.net/qq_31142553/article/details/85645957

1）定义一个标记接口（什么也没有，类似Serializable），让需要校验的类实现。

/*** 需要校验的请求对象** @author z_hh*/
public interface ValidateReq {
}

/*** 添加任务应用请求** @author z_hh*/
@ApiModel("添加任务应用请求")
@Data
public class JobAppAddReq implements ValidateReq {@ApiModelProperty(value = "应用名称", required = true)@NotBlank(message = "应用名称不能为空")private String appName;
}

2）定义一个Aspect切面，在控制层对这些请求对象进行校验。

这里可以使用@Around环绕切面，如果校验不通过，直接返回错误的Result结果，那么就没有第三步了。

/*** 校验请求对象切面** @author z_hh*/
@Aspect
@Component
public class ValidateReqAspect {private static final Validator validator = Validation.buildDefaultValidatorFactory().getValidator();@Before("execution(* cn.zhh.admin.controller.*.*(..))")public void validateReq(JoinPoint joinPoint) {Object[] args = joinPoint.getArgs();for(int i = 0, length = args.length; i < length; ++i) {Object obj = args[i];if (obj instanceof ValidateReq) {validate(obj);}}}private <T> void validate(T t) {// 校验对象Set<ConstraintViolation<T>> constraintViolations = validator.validate(t, new Class[0]);// 存在校验错误的话，拼接所有错误信息if (constraintViolations.size() > 0) {StringBuilder validateError = new StringBuilder();ConstraintViolation constraintViolation;for(Iterator iterator = constraintViolations.iterator(); iterator.hasNext(); validateError.append(constraintViolation.getMessage()).append(";")) {constraintViolation = (ConstraintViolation)iterator.next();}// 抛出异常，统一异常处理器将会处理throw new IllegalArgumentException(validateError.toString());}}}

3）定义全局统一异常处理器将IllegalArgumentException转化为错误的Result对象。

/*** 全局异常处理器** @author z_hh*/
@RestControllerAdvice
@Slf4j
public class GlobalExceptionHandler {/*** 处理Throwable异常* @param t 异常对象* @return 统一Result*/@ExceptionHandler(Throwable.class)public Result handleThrowable(Throwable t) {String msg = ThrowableUtils.getThrowableStackTrace(t);log.error("统一处理未知异常：{}", msg);return Result.err(msg);}/*** 处理非法参数异常* @param e 异常对象* @return 统一Result*/@ExceptionHandler(IllegalArgumentException.class)public Result handleIllegalArgumentException(IllegalArgumentException e) {log.error("统一处理非法参数异常：{}", ThrowableUtils.getThrowableStackTrace(e));return Result.err(e.getMessage());}
}

4、Spring Data Jpa的Active Record模式实现

我的相关博客：https://blog.csdn.net/qq_31142553/article/details/82959626

Active Record，即AR模式，就是让实体对象本身具备数据库操作(增删改查)的能力。违反了低耦合的设计原则，但有时候使用确实方便。

1）定义一个公共实体基类，子类继承时需要在泛型上面写入自己的类型以及主键类型、对应DAO的类型。

这里面会根据DAO泛型找到对应的Dao Bean实例以实现对数据库的操作。然后，子类可以选择重写获取主键值的方法，如果没有覆盖，就会通过反射找到有@ID注解的那个字段取值。

/*** AR模式的实体基类** @author z_hh*/
public class ActiveRecord<T extends ActiveRecord, ID, DAO> {private JpaRepository<T, ID> jpaRepository;/*** 达到延迟加载的效果** @return dao对象*/private JpaRepository<T, ID> dao() {return Optional.ofNullable(jpaRepository).orElseGet(() -> {Type type = this.getClass().getGenericSuperclass();Type[] parameter = ((ParameterizedType) type).getActualTypeArguments();Class<DAO> daoClazz = (Class<DAO>)parameter[2];if (daoClazz.isAnnotationPresent(Repository.class)) {Repository annotation = daoClazz.getAnnotation(Repository.class);return jpaRepository = (JpaRepository<T, ID>) BeanUtils.getBean(annotation.value());}String clazzName = daoClazz.getSimpleName();String beanName = clazzName.substring(0, 1).toLowerCase() + clazzName.substring(1);return jpaRepository = (JpaRepository<T, ID>) BeanUtils.getBean(beanName);});}/*** 保存this对象** @return 保存过的对象*/public T save() {return dao().save((T)this);}/*** 根据this的主键删除数据*/public void deleteById() {dao().deleteById(pkVal());}/*** 通过this构造Example进行查询** @return 结果列表*/public List<T> findAllByExample() {return dao().findAll(Example.of((T)this));}/*** 根据this的主键查询数据** @return Optional对象*/public Optional<T> findById() {return dao().findById(pkVal());}/*** 推荐子类重写该方法，返回主键的值** @return*/protected ID pkVal() {return Arrays.stream(this.getClass().getDeclaredFields()).filter(f -> f.isAnnotationPresent(Id.class)).map(f -> {f.setAccessible(true);return (ID) ReflectionUtils.getField(f, this);}).findAny().orElse((ID)null);}
}

2）子类继承基类，并设置上面的泛型参数。

/*** 任务应用** @author z_hh*/
@Data
@Entity(name = "z_job_app")
public class JobApp extends ActiveRecord<JobApp, Long, JobAppDao> implements Serializable {private static final long serialVersionUID = 1L;/*** id*/@Id@GeneratedValue(strategy = GenerationType.IDENTITY)private Long id;
}

3）这样，一个实体类就具备了增删改查的能力。

@RunWith(SpringJUnit4ClassRunner.class)
@SpringBootTest
public class ActiveRecordTest {@Testpublic void testSave() throws Exception {JobApp jobApp = new JobApp();jobApp.setAppName("example");jobApp.setAppDesc("实例应用");jobApp.setCreator("ZHH");jobApp.setCreateTime(new Date());jobApp.setCreateWay((byte) 0);jobApp.setAddressList("127.0.0.1:8080");jobApp.setEnabled((byte) 0);jobApp.setIsDeleted((byte) 0);jobApp = jobApp.save();Assert.assertNotNull(jobApp.getId());}@Testpublic void testFindAllByExample() {JobApp jobApp = new JobApp();jobApp.setId(1L);List<JobApp> jobAppList = jobApp.findAllByExample();Assert.assertFalse(jobAppList.isEmpty());}@Testpublic void findById() {JobApp jobApp = new JobApp();jobApp.setId(1L);Optional<JobApp> jobAppOptional = jobApp.findById();Assert.assertTrue(jobAppOptional.isPresent());}}

写了大半天终于搞完了?，有什么问题欢迎在评论区留言哦！