Auto-Job任务调度框架
| 阿里云国内75折 回扣 微信号：monov8 |
| 阿里云国际，腾讯云国际，低至75折。AWS 93折 免费开户实名账号 代冲值 优惠多多 微信号：monov8 飞机：@monov6 |
Auto-Job 任务调度框架
生活中，业务上我们会碰到很多有关作业调度的场景，如每周五十二点发放优惠券、或者每天凌晨进行缓存预热、亦或每月定期从第三方系统抽数等等，Spring和java目前也有原生的定时任务支持，但是其都存在一些弊病，如下：
- 不支持集群，未避免任务重复执行的问题
- 不支持生命周期的统一管理
- 不支持分片任务：处理有序数据时，多机器分片执行任务处理不同数据
- 不支持失败重试：出现异常任务终结，不能根据执行状态控制任务重新执行
- 不能很好的和企业系统集成，如不能很好的和企业系统前端集成以及不能很好的嵌入到后端服务
- 不支持动态调整：不重启服务情况下不能修改任务参数
- 无报警机制：任务失败之后没有报警通知（邮箱、短信）
- 无良好的执行日志和调度日志跟踪
基于原生定时任务的这些弊病，AutoJob就由此诞生，AutoJob为解决分布式作业调度提供了新的思路和解决方案。
二、特性
简单： 简单包括集成简单、开发简单和使用简单。
集成简单：框架能非常简单的集成到Spring项目和非Spring项目，得益于AutoJob不依赖于Spring容器环境和MyBatis环境，你无需为了使用该框架还得搭建一套Spring应用。
开发简单：AutoJob开发初衷就希望具有低代码侵入性和快速开发的特点，如下在任意一个类中，你只需要在某个需要调度的任务上加上注解，该任务就会被框架进行动态调度：
@AutoJob(attributes = "{'我爱你，心连心',12.5,12,true}", cronExpression = "5/7 * * * * ?")
public void formatAttributes(String string, Double decimal, Integer num, Boolean flag) {
//参数注入
AutoJobLogHelper logger = new AutoJobLogHelper();//使用框架内置的日志类
logger.setSlf4jProxy(log);//对Slf4j的log进行代理，日志输出将会使用Slf4j输出
logger.info("string={}", string);
logger.warn("decimal={}", decimal);
logger.debug("num={}", num);
logger.error("flag={}", flag);
//使用mapper
mapper.selectById(21312L);
//...
}
使用简单：使用该框架你无需关注太多的配置，整个框架的启动只需要一行代码,如下：
//配置任务扫描包路径
@AutoJobScan({"com.yourpackage"})
//处理器自动扫描
@AutoJobProcessorScan({"com.yourpackage"})
public class AutoJobMainApplication {
public static void main(String[] args) {
//框架启动
new AutoJobBootstrap(AutoJobMainApplication.class)
.build()
.run();
System.out.println("==================================>系统创建完成");
}
}
得益于良好的系统架构和编码设计，你的应用启动无需过多配置，只需要一行代码
拓展： 框架源码采用多种合理设计模式设计，具有良好的可拓展性和可维护性。
动态： 框架提供API，支持任务的动态CURD操作，即时生效。
多数据库支持： 提供多类型数据库支持，目前支持MySQL和PostgreSQL。
任务依赖： 支持配置子任务，当父任务执行结束且执行成功后将会主动触发一次子任务的执行。
一致性： 框架使用DB乐观锁实现任务的一致性，在集群模式下，调度器在调度任务前都会尝试获取锁，获取锁成功后才会进行该任务的调度。
HA（开发中）： 该框架支持去中心化的集群部署，集群节点通过RPC加密通信。集群节点之间会自动进行故障转移和负载均衡，
弹性增缩容（开发中）： 支持节点的动态上下线，同时节点支持开启保护模式，防止恶劣的网络环境下节点脱离集群。
任务失败重试： 支持任务失败重试，并且可设置重试间隔。
完整的生命周期： 框架提供任务完整的生命周期事件，业务可捕捉并做对应的处理。
动态调度线程池： 框架使用自研的动态线程池，可灵活根据任务流量动态调整线程池核心线程和最大线程参数，节省系统线程资源，并且提供了默认的拒绝处理器，防止任务被missFire。
异步非阻塞的日志处理： 日志采用生产者消费者模型，基于自研的内存消息队列，任务方法作为日志的生产者，生产日志放入消息队列，框架启动对应的日志消费线程进行日志处理。
实时日志： 日志将会实时的进行保存，便于跟踪。
任务白名单： 提供任务白名单功能，只有在白名单中的任务才允许被注册和调度，保证系统安全。
可拓展的日志存储策略： 日志支持多种策略保存，如内存Cache、数据库等，可根据项目需要灵活增加保存策略，如Redis、文件等。
丰富的调度机制： 支持Cron like表达式，repeat-cycle调度、子任务触发、延迟触发等，得益于良好的编码设计，用户可非常简单的新增自定义调度器，如下：
/**
* 你的自定义调度器
* @Author Huang Yongxiang
* @Date 2022/08/18 14:56
*/
public class YourScheduler extends AbstractScheduler{
public YourScheduler(AutoJobTaskExecutorPool executorPool, IAutoJobRegister register, AutoJobConfigHolder configHolder) {
super(executorPool, register, configHolder);
}
//...调度逻辑
}
@AutoJobScan("com.example.autojob.job")
@AutoJobProcessorScan("com.example.autojob")
public class AutoJobMainApplication {
public static void main(String[] args) {
new AutoJobLauncherBuilder(AutoJobMainApplication.class)
.withAutoScanProcessor()
//配置你的调度器
.addScheduler(YourScheduler.class)
.build()
.run();
System.out.println("==================================>系统创建完成");
}
}
任务报警： 框架支持邮件报警，目前原生支持QQ邮箱、163邮箱、GMail等，同时也支持自定义的邮箱smtp服务器。
目前系统提供：任务失败报警、任务被拒报警、节点开启保护模式报警、节点关闭保护模式报警，当然用户也可非常简单的进行邮件报警的拓展。
丰富的任务入参： 框架支持基础的数据类型和对象类型的任务入参，如Boolean,String,Long,Integer,Double等类型，对于对象入参，框架默认使用JSON进行序列化入参。
良好的前端集成性： 框架提供相关API，用户可以灵活开发Restful接口接入到企业项目，无需额外占用一个进程或机器来单独运行调度中心。
内存任务： 框架提供DB任务和内存任务两种类型，DB任务持久化到数据库，声明周期在数据库内记录，内存任务除了日志，整个生命周期都在内存中完成，相比DB任务具有无锁、调度快速的特点。
脚本任务： 提供脚本任务的执行，如Python、Shell，SQL等。
动态分片（开发中）： 集群模式下框架支持任务分片，多机运行。
全异步： 任务调度流程采用全异步实现，如异步调度、异步执行、异步日志等，有效对密集调度进行流量削峰，理论上支持任意时长任务的运行。
三、快速使用
1、项目导入
该框架不依赖于Spring容器环境和MyBatis等持久层框架，你可以将其作为一个Maven模块导入到你的项目中，你可以去码云上下载：https://gitee.com/hyxl-520/auto-job.git
项目分为两个模块：auto-job-framework和auto-job-spring，前者是框架的核心部分，后者是与Spring集成的使用，后续可能会基于Spring web开发相关控制台。
2、项目配置
项目配置主要为框架配置和数据源配置。框架配置默认读取类路径下的auto-job.yml和auto-job.properties文件，具体配置项内容见“所有配置”；数据源配置，框架默认使用Druid作为连接池，你只需要在druid.properties文件中配置数据源就行了，当然你可以自定义数据源，具体方法在AutoJobBootstrap里。相关建表脚本可以在db目录下找到。框架默认使用MySQL数据库，理论上支持SQL标准的其他数据库
3、任务开发
3.1、基于注解
开发一个基于注解的任务非常简单，除了日志输出使用框架内置的日志辅助类AutoJobLogHelper输出外，其他你就只需要关心你的业务。当然，AutoJobLogHelper使用起来和slf4j几乎没有区别，它提供四种级别的日志输出：debug、info、warn、error，而且你可以使用AutoJobLogHelper对你的slf4j进行代理，这样这些任务执行中输出的日志将会直接使用slf4j进行输出。如下，是一个简单演示：
@AutoJob(attributes = "{'我爱你，心连心',12.5,12,true}", cronExpression = "5/7 * * * * ?", id = 2, alias = "参数测试任务")
public void formatAttributes(String string, Double decimal, Integer num, Boolean flag) {
AutoJobLogHelper logger=new AutoJobLogHelper();
//log是org.slf4j.Logger对象，这里对其进行代理
logger.setSlf4jProxy(log);
logger.info("string={}", string);
logger.warn("decimal={}", decimal);
logger.debug("num={}", num);
logger.error("flag={}", flag);
}
在你开发的任务上加上@AutoJob注解，配置一些东西，这个任务就开发完成了。@AutoJob是用来标识一个方法是一个AutoJob任务，当然还有其他注解，这里暂不做阐述。细心的同学会发现这个任务是有参数的，没错，AutoJob框架支持参数，更多参数的配置后文会详细讲解。
3.2、基于构建
手动创建任务相比注解来说更为灵活，框架提供了创建任务的构建者对象，如AutoJobMethodTaskBuilder和AutoJobScriptTaskBuilder对象，前者用于构建方法型任务，后者用于构建脚本型任务。
MethodTask task = new AutoJobMethodTaskBuilder(Jobs.class, "hello") //方法型任务需要指定方法所在的类以及方法名
.setTaskId(IdGenerator.getNextIdAsLong())
.setTaskAlias("测试任务") //任务别名
.setParams("{'我爱你，心连心',12.5,12,true}") //任务参数，支持simple参数
.setTaskType(AutoJobTask.TaskType.MEMORY_TASk)
.setMethodObjectFactory(new DefaultMethodObjectFactory()) //方法运行对象工厂，用于创建方法运行的对象上下文
.addACronExpressionTrigger("* 5 7 * * * ?", -1) //添加一个cron-like触发器
.build();
AutoJobApplication
.getInstance()
.getMemoryTaskAPI() //获取全局的内存任务的API
.registerTask(new AutoJobMethodTaskAttributes(task)); //注册任务
4、框架启动
得益于良好的设计，该框架你可以在任何一个main方法启动，如下是示列的一个启动
import com.example.autojob.skeleton.annotation.AutoJobProcessorScan;
import com.example.autojob.skeleton.annotation.AutoJobScan;
import com.example.autojob.skeleton.framework.boot.AutoJobLauncherBuilder;
@AutoJobScan("com.example.autojob.job")
@AutoJobProcessorScan("com.example.autojob")
public class AutoJobMainApplication {
public static void main(String[] args) {
new AutoJobBootstrap(AutoJobSpringApplication.class)
.withAutoScanProcessor()
.build()
.run();
System.out.println("==================================>AutoJob应用已启动完成");
}
}
第5行是用于配置任务扫描的类路径，支持子包扫描，不配置时会扫描整个项目，用时较长。
第6行是处理器扫描，处理器主要是在框架启动前和框架启动后进行一些处理，默认是扫描整个项目，注意该注解只有设置了withAutoScanProcessor才能生效，如代码第10行，框架自己的处理器为自动加载，无需配置。
第9-12行是框架的启动代码，AutoJobBootstrap是应用引导构建程序，通过它你能增加很多自定义的配置。在第11行后，AutoJob应用即创建完成，第12行调用run方法启动整个应用。
5、动态修改
框架本身不是一个Web应用，没有提供对应修改的Rest接口，但是框架提供了很多操作任务的API，你可以在AutoJobAPI和AutoJobLogAPI里找到。你可以你可以参考auto-job-spring模块里提供的实例开发对应Rest接口，随着版本更替，autojob将会在未来支持控制台。
四、任务类型
按照功能分类
任务按照功能可以分为方法型任务和脚本型任务。
方法型任务对应Java中的一个方法，该方法可以有返回值，允许有参数，参数的注入可以见“任务参数”。方法内部的日志输出必须使用AutoJobLogHelper来输出，否则日志可能无法保存。
脚本型任务对应一个磁盘上的脚本文件或一段cmd命令。具体使用可见章节：“高级用法-脚本任务”。
按照调度方式分类
任务按照调度方式可以分为内存型任务和DB型任务。
内存型任务的生命周期都在内存中完成，具有调度迅速、无锁、随调随动的特点，适合短周期、有限次、临时性的任务。
DB型任务将会保存到数据库，每一次调度都会更新数据库相关状态。DB型任务采用乐观锁，每次执行前都需要获得锁才能执行，具有长期性、易维护、易修改等特点，适合于定期数据同步、定时缓存预热等在长期内都会用到的任务。
五、任务参数
方法型任务
方法型任务支持两种参数格式，一种是FULL型参数，一种是SIMPLE参数，具体区别可见如下示列：
void exampleMethod1(String str, Integer num, Double decimal, Boolean flag);
void exampleMethod2(String str, Integer num, Double decimal, Boolean flag, Long count, Param param);
class param{
private int id;
private String num;
//...
}
如上方法：exampleMethod1，使用SIMPLE型参数：
MethodTask task = new AutoJobMethodTaskBuilder(Jobs.class, "hello")
.setTaskId(IdGenerator.getNextIdAsLong())
.setTaskAlias("测试任务")
.setParams("{'我是字符串参数',12,12.5,true}")
.setTaskType(AutoJobTask.TaskType.MEMORY_TASk)
.setMethodObjectFactory(new DefaultMethodObjectFactory())
.build();
//{'我是字符串参数',12,12.5,true}
使用FULL型参数
MethodTask task = new AutoJobMethodTaskBuilder(Jobs.class, "hello")
.setTaskId(IdGenerator.getNextIdAsLong())
.setTaskAlias("测试任务")
.setParams("[{\"values\":{\"value\":\"字符串参数\"},\"type\":\"string\"},{\"values\":{\"value\":12},\"type\":\"integer\"},{\"values\":{\"value\":12.5},\"type\":\"decimal\"},{\"values\":{\"value\":false},\"type\":\"boolean\"}]")
.setTaskType(AutoJobTask.TaskType.MEMORY_TASk)
.setMethodObjectFactory(new DefaultMethodObjectFactory())
.build();
/*
[
{
"values": {
"value": "字符串参数"
},
"type": "string"
},
{
"values": {
"value": 12
},
"type": "integer"
},
{
"values": {
"value": 12.5
},
"type": "decimal"
},
{
"values": {
"value": false
},
"type": "boolean"
}
]
*/
我们可以发现SIMPLE参数十分简单，"{a1,a2,a3,...}"，参数表达式本身是一个字符串，大引号包裹，参数顺序按照从左到右依次匹配。SIMPLE参数支持四类参数
'字符串参数'，单引号包裹，对应类型String；
12：整数型参数，对应类型：Integer包装类型，如果数值超过整形范围，则会自动匹配Long类型。
12.5：小数型参数，对应类型：Double包装类型。
true|false：布尔型参数，对应类型：Boolean包装类型。
FULL型参数相比就要复杂的多了，本身是一个JSON数组字符串，每一个JSON对象代表一个参数，每个对象有type和values两个属性，字面意思，类型和值，FULL类型除了支持SIMPLE型的四种类型参数外还支持对象型，对象型的参数使用JSON来进行序列化和反序列化。由于FULL型参数过于复杂，因此框架提供了AttributesBuilder对象，可以非常简单的生成FULL型参数，以exampleMethod2为例：
Param param = new Param();
param.setId(1);
param.setNum("12");
System.out.println(new AttributesBuilder()
.addParams(AttributesBuilder.AttributesType.STRING, "字符串参数")
.addParams(AttributesBuilder.AttributesType.INTEGER, 12)
.addParams(AttributesBuilder.AttributesType.DECIMAL, 12.5)
.addParams(AttributesBuilder.AttributesType.BOOLEAN, false)
.addParams(Param.class, param)
.getAttributesString());
/*
[
{
"values": {
"value": "字符串参数"
},
"type": "string"
},
{
"values": {
"value": 12
},
"type": "integer"
},
{
"values": {
"value": 12.5
},
"type": "decimal"
},
{
"values": {
"value": false
},
"type": "boolean"
},
{
"values": {
"id": 1,
"num": "12"
},
"type": "com.example.autojob.job.Param"
}
]
*/
一般来说，基于注解的任务开发我们更倾向于推荐使用SIMPLE型参数，简单、明了；基于构建的任务开发我们更钟意于FULL型参数，类型丰富。
脚本型任务
脚本型任务的参数是通过启动命令给出的，如python /script.test.py -a 12 -b，其中-a 12和-b就是两个参数，因此脚本型任务只支持字符串型参数。
七、任务运行对象工厂
任务运行对象工厂是方法型任务才有的属性，因为方法型任务对应的是Java某个类中的方法，因此方法的执行可能依赖于对象实例的上下文，特别是当该框架与Spring集成时很可能会使用Spring容器中的Bean，因此可以指定创建方法依赖的对象的工厂：IMethodObjectFactory，框架默认使用类的无参构造方法创建对象实例，当然你可以创建自定义的工厂：
public class SpringMethodObjectFactory implements IMethodObjectFactory {
public Object createMethodObject(Class<?> methodClass) {
// SpringUtil持有Spring的容器，获取Spring容器中的Bean
return SpringUtil.getBean(JobBean.class);
}
}
那么怎么让我们的任务运行对象工厂生效呢，见如下示列：
// 基于注解的任务开发只需要指定methodObjectFactory属性即可，框架将会调用指定工厂的无参构造方法创建一个工厂实例
@AutoJob
(
id = 1
, attributes = "{'hello autoJob'}"
, defaultStartTime = StartTime.NOW
, repeatTimes = -1, cycle = 5
, methodObjectFactory = SpringMethodObjectFactory.class
)
public void hello(String str) {
logHelper.info(str);
}
//基于构建的任务开发时将工厂实例配置进去即可
public static void main(String[] args) {
MethodTask methodTask = new AutoJobMethodTaskBuilder(Jobs.class, "hello")
.setMethodObjectFactory(new SpringMethodObjectFactory())
.build();
AutoJobApplication
.getInstance()
.getMemoryTaskAPI() //获取全局的内存任务的API
.registerTask(new AutoJobMethodTaskAttributes(task)); //注册任务
}
八、任务日志
作为一款任务调度框架，详细的日志一定是必不可少的。框架提供三种类型日志记录：调度日志、执行日志、运行日志
调度日志
任务的每一次启动到完成被任务是一次调度，调度日志详细记录了调度任务的基础信息、调度时间、运行状态、执行时长、以及任务结果（任务结果对应方法型任务是返回值，由JSON序列化，脚本型任务是脚本返回值）。调度日志对应数据库表aj_scheduling_record，其ID关联到本次调度中产生的运行日志和执行日志。
运行日志
运行日志为任务在运行期间内部输出的日志，方法型任务为使用AutoJobLogHelper输出的日志，脚本型任务为脚本或cmd命令在控制台的输出。运行日志对应数据库表aj_job_logs。
执行日志
执行日志记录了某次调度任务的执行情况，如何时启动、何时完成、是否运行成功、任务结果、任务异常等。执行日志对应库表aj_run_logs。
任务日志都是实时更新的，如果你使用的是框架的默认日志保存策略（数据库存储），你可以通过AutoJobLogDBAPI获取到日志。运行日志和执行日志都绑定了调度ID，通过调度ID即可找到本次调度所产生的运行日志和执行日志。
九、框架架构
框架架构图的左部分的组件是框架的核心组件。
任务容器模块
任务容器模块包含DB任务容器和内存任务容器，分别用于存放DB型的任务和内存型的任务。
调度模块
调度模块由调度器、任务调度队列、注册器、时间轮调度器以及时间轮构成。内存任务调度器AutoJobMemoryTaskScheduler和DB任务调度器AutoJobDBScheduler负责从任务容器调度出即将执行的任务（<=5秒）放到任务调度队列缓存AutoJobTaskQueue。时间轮调度器AutoJobTimeWheelScheduler通过注册器AutoJobRegister调度任务调度队列中的任务进入时间轮，准备执行。时间轮按秒滚动，将执行的任务提交进任务执行器池进行执行。运行成功调度器AutoJobRunSuccessScheduler执行运行成功后的相关操作，比如更新状态、更新下次触发时间等等，运行失败调度器AutoJobRunErrorScheduler执行运行失败后的相关操作，比如更新状态、根据配置的重试策略更新触发时间、故障转移等等。
任务执行器池模块
任务执行器池包含两个动态线程池，分别为快池（fast-pool）和慢池（slow-pool），任务默认第一次执行提交进快池，第二次执行会根据上次执行时长决定是否降级处理。动态线程池是具有根据流量动态调节的线程池，具体的配置可以见“十、所有配置:执行器池配置”。
日志模块
日志模块和核心调度模块是完全解耦的，运行日志由任务执行时产生并且发布到内存消息队列，日志模块监听消息发布事件并且取出消息放入消息buffer，单独由日志处理线程定期、定量保存日志。运行日志通过监听任务事件来进行保存。日志模块的设计都是异步化的，尽最大可能减小日志IO对调度的影响。
除了以上的核心组件外，框架还有部分功能拓展组件。
生命周期处理器
生命周期处理器也可以理解成生命周期钩子，具体来说是一个任务的生命周期钩子，具体看下面的生命周期事件图
要使用一个生命周期钩子也十分简单，下面来看一个示列：
//方式一（子事件处理器）
public class TaskBeforeRunHandle implements ITaskEventHandler<TaskBeforeRunEvent> {
@Override
public void doHandle(TaskBeforeRunEvent event) {
System.out.println("任务：" + event
.getTask()
.getAlias() + "即将开始运行");
}
@Override
public int getHandlerLevel() {
return 0;
}
}
以上示列表示一个在任务执行前在控制台输出：“任务：{任务别名}即将开始运行”，要实现一个事件处理器只需要实现ITaskEventHandler接口即可，泛型代表你需要处理的事件。当然还可以通过如下方式来实现同上面示列一样的功能
//方式二（父事件处理器）
public class TaskBeforeRunHandle implements ITaskEventHandler<TaskEvent> {
@Override
public void doHandle(TaskEvent event) {
if (event instanceof TaskBeforeRunEvent) {
System.out.println("任务：" + event
.getTask()
.getAlias() + "即将开始运行");
}
}
@Override
public int getHandlerLevel() {
//数字越大，级别越高
return 0;
}
}
TaskEvent是所有任务事件的父类，实现其父类事件的处理器时所有的任务相关事件都会执行该处理器，可以判断事件类型来完成相关操作，当一个处理器需要处理多种事件类型时可以如上使用。每个事件处理器可以通过重写getHandlerLevel方法指定级别，数字越大，级别越高，执行越会被优先执行。父事件处理器高级别>父事件处理器低级别>子事件处理器高级别>子事件处理器低级别。当然，只声明处理器不将其添加到应用也不会生效的，下面介绍如何使得事件处理器生效。
public class TaskEventHandlerLoader implements IAutoJobLoader {
@Override
public void load() {
//方式一（子事件处理器）
TaskEventHandlerDelegate
.getInstance()
.addHandler(TaskBeforeRunEvent.class, new TaskBeforeRunHandle());
//方式二（父事件处理器）
TaskEventHandlerDelegate
.getInstance()
.addHandler(TaskEvent.class, new TaskBeforeRunHandle());
}
}
//将启动处理器添加进应用上下文
public static void main(String[] args) {
new AutoJobBootstrap(AutoJobMainApplication.class)
.addProcessor(new TaskEventHandlerLoader()) //添加到上下文
.build()
.run();
}
上面的代码演示了如何添加处理器到上下文。在AutoJob中，在框架启动前和框架关闭前执行某些操作的处理器成为Processor，框架启动前执行的处理器为IAutoJobLoader，框架关闭前执行的处理器为IAutoJobEnd，上面代码中，通过启动处理器将事件处理器添加到“事件委派者”：TaskEventHandlerDelegate，再在应用构建时手动将启动处理器添加到应用上下文中。当然如果你的Processor非常多，可以通过注解@AutoJobProcessorScan来自动扫描Processor，可以指定扫描的包，支持子包扫描，不指定时默认全项目扫描。扫描后通过调用Processor的无参构造方法创建实例后自动注入上下文。如下示列：
@AutoJobProcessorScan("com.example.autojob")
public class AutoJobMainApplication {
public static void main(String[] args) {
new AutoJobBootstrap(AutoJobMainApplication.class) //指定入口类
.withAutoScanProcessor() //手动开启处理器自动扫描，默认是关闭的，以防全项目扫描耗时较长
.build()
.run();
System.out.println("==================================>系统创建完成");
}
十、所有配置
框架提供了丰富的配置，这些配置默认是从auto-job.yml或者auto-job.properties文件中加载，当然你可以从数据库动态加载实现动态配置，全部配置如下：
# 动态任务调度框架配置
autoJob:
context:
schedulingQueue:
length: 100 # 调度队列长度，调度队列用于存放即将执行的任务
memoryContainer: # 内存型任务容器，存放内存型任务
length: 200 # 容器容量
cleanStrategy: CLEAN_FINISHED # 清理策略，CLEAN_FINISHED-定期清理已经执行完成的任务 KEEP_FINISHED-保留执行完成的任务，会将其移入一个内存Cache，不会占用容器容量
annotation:
enable: true # 是否启用注解扫描，扫描被@AutoJob @FactoryJob的方法并将其包装成可执行任务对象
defaultDelayTime: 30 # 在未给注解的任务配置调度信息的情况下，默认的任务延迟执行时间：min
database:
type: mysql # 数据库类型，目前支持，MySQL和PostgreSQL
executor: # 执行器池，分为快池和慢池
fastPool: # 快池相关配置，慢池相同
update: # 执行器池支持根据流量动态调整线程数目
enable: true # 是否开启
trafficUpdateCycle: 5 # 流量监控周期：秒
adjustedThreshold: 0.05 # 如果流量变化相比最大线程数超过此比例（0-1），则进行调整
coreThread: # 核心线程数
initial: 5 # 初始值
min: 5 # 允许变化到的最小值
max: 50 # 允许变化到的最大值
keepAliveTime: 60 # 当线程数大于当前核心线程数时，线程保持生存的时间：秒
maxThread: # 最大线程数
initial: 10
min: 10
max: 50
slowPool: # 慢池
update:
enable: false
trafficUpdateCycel: 5
adjustedThreshold: 0.05
coreThread:
initial: 10
min: 5
max: 50
keepAliveTime: 60
maxThread:
initial: 20
min: 10
max: 50
relegation:
threshold: 3 # 降级阈值，当任务的上次执行时长超过该阈值（分钟）时，下次将会降级到slow pool运行
register:
filter: # 注册过滤器用于防止某些不安全的任务被执行
enable: true
classPath: "**.job.**" # 只有在这些类路径下的任务才允许被注册和执行
scheduler:
finished:
error:
retry: # 失败重试相关配置，该配置是全局的
enable: true
retryCount: 3
interval: 1 # 两次重试的间隔：min
emailAlert: # 全局邮件报警相关配置
enable: true
auth:
sender: "XXX@163.com" # 发送方，唯一
receiver: "XXX@qq.com" # 接收方，多个逗号分割
token: "LXZYE214123CEWASU" # smtp密码
type: 163Mail # 邮件类型，目前支持：QQMail、163Mail、gMail（google）、outLookMail、customize（自定义）
customize: # 自定义下的smtp服务器的相关配置
smtpAddress:
smtpPort:
config: # 提供部分事件报警（开关）
taskRunError: true # 任务运行出错（优先使用任务私有邮件客户端，不存在使用全局客户端）
taskRefuseHandle: true # 任务被拒绝执行（优先使用任务私有邮件客户端，不存在使用全局客户端）
clusterOpenProtectedMode: true # 集群节点开启保护模式（集群模式下有效）
clusterCloseProtectedMode: true # 集群节点关闭保护模式（集群模式下有效）
logging: # 日志的相关配置
taskLog: # 任务内部通过调用logger输出的日志
memory: # 日志默认是数据库保存，框架额外提供了内存Cache保存，内存Cache一般仅做测试，该配置一般情况下无需更改
enable: false
length: 100
defaultExpireTime: 3 # 分钟
runLog: # 任务的调度日志
memory:
enable: false
length: 100
defaultExpireTime: 3
cluster: # 集群相关配置，目前版本暂无需考虑
enable: false # 集群开关，目前版本开启后会启动PRC服务器
port: 8080 # TCP端口
auth: # RPC通信身份验证
enable: true # 是否启动身份验证
publicKey: "autoJob!@#=123.?" # 通信加密公钥，16位字符串
token: "hello" # token，相同token的两个AutoJob应用才能通信
client:
nodeUrl: "localhost:8086"
pool: # RPC会话池相关配置
size: 10
getTimeout: 3
getDataTimeout: 10
connectTimeout: 10
keepAliveTimeout: 10
allowMaxJetLag: 3
nodeSync:
cycle: 5
offLineThreshold: 3
config:
annotations:
enable: false
protectedMode:
enable: true
threshold: 0.2
当然上面配置并不是都需要你配置，框架基本所有配置都设置了默认值，能保证常规场景下的调度。
十一、高级用法
1、脚本任务
框架支持脚本任务，原生支持：Python、Shell、PHP、NodeJs以及PowerShell，提供其他脚本类型拓展。脚本任务对应的对象为ScriptTask。脚本作为一个服务器上的脚本文件保存在磁盘上，要构建一个脚本任务非常简单，框架提供AutoJobScriptTaskBuilder来辅助构建一个完整的脚本任务，下面看几个示列：
ScriptTask task = new AutoJobScriptTaskBuilder()
.setTaskId(IdGenerator.getNextIdAsLong()) //设置任务ID，任务ID作为区分任务的键，不指定时将会随机分配
.setTaskAlias("测试脚本任务1") //任务别名
.setTaskType(AutoJobTask.TaskType.MEMORY_TASk) //任务类型，有内存型任务和DB型任务，内存型任务的所有生命周期都在内存完成，除了日志外不会保留到数据库
.setBelongTo(1L) //保留拓展字段，用于说明该任务所属
.addACronExpressionTrigger("* 15 7 * * * ?", -1) //添加一个cron-like触发器，两个参数分别是：cron-like表达式、重复次数。不指定触发器时将会在默认延迟后执行一次，-1表示该任务为永久执行，如果只需执行n次，重复次数为n-1
.createNewWithContent(ScriptType.PYTHON, "print('hello auto-job')"); // 使用脚本类型和脚本内容构建一个脚本任务对象
ScriptTask task1 = new AutoJobScriptTaskBuilder()
.setTaskId(IdGenerator.getNextIdAsLong())
.setTaskAlias("测试脚本任务2")
.setTaskType(AutoJobTask.TaskType.MEMORY_TASk)
.setBelongTo(1L)
.addASimpleTrigger(SystemClock.now(), 3, 10, TimeUnit.SECONDS) //添加一个简单触发器，四个参数分别是：启动时间、重复次数、周期、周期时间单位，该触发器表示立即执行，并且重复执行三次，总共执行四次，周期为10秒
.createNew("python", "/script", "test", "py"); // 使用给定路径的脚本文件创建一个脚本任务，四个参数分别是：启动命令、脚本路径、脚本文件名、脚本后缀，该方法能够创建除框架原生脚本类型以外的脚本任务
ScriptTask task2 = new AutoJobScriptTaskBuilder()
.setTaskId(IdGenerator.getNextIdAsLong())
.setTaskAlias("测试脚本任务3")
.setTaskType(AutoJobTask.TaskType.MEMORY_TASk)
.setBelongTo(1L)
.addAChildTaskTrigger() // 添加一个子任务触发器，该任务不会自动触发，只有当有任务主动关联该任务作为其子任务且父任务完成一次调度时才会触发该任务
.createNewWithCmd("ping www.baidu.com"); // 创建一个cmd脚本任务
ScriptTask task3 = new AutoJobScriptTaskBuilder()
.setTaskId(IdGenerator.getNextIdAsLong())
.setTaskAlias("测试脚本任务4")
.setTaskType(AutoJobTask.TaskType.MEMORY_TASk)
.setBelongTo(1L)
.addADelayTrigger(3, TimeUnit.MINUTES) // 添加一个延迟触发器，任务将在给定延迟后自动触发一次，默认使用该类型触发器，延迟时间可以在框架配置中配置
.createNewWithExistScriptFile(ScriptType.PYTHON, "/script", "test"); // 使用已存在的脚本创建一个脚本任务，三个参数分别是：脚本类型、脚本路径、脚本文件名
以上示列除了演示了如何创建一个脚本任务，也介绍了触发器。框架提供了四种触发器，分别是cron-like触发器、simple触发器、父-子任务触发器、延迟触发器，具体触发器的介绍上面代码注释基本讲解了这里就不作冗述。
2、自定义调度器
调度器的概念在第九节：框架架构里已经说明，那么怎么来自定义一个自己的调度器呢，下面做一个简单示列：
/**
* 你的自定义调度器
*
* @Author Huang Yongxiang
* @Date 2022/08/18 14:56
*/
public class YourScheduler extends AbstractScheduler{
//调度器默认构造方法
public YourScheduler(AutoJobTaskExecutorPool executorPool, IAutoJobRegister register, AutoJobConfigHolder configHolder) {
super(executorPool, register, configHolder);
}
//...调度逻辑
}
@AutoJobScan("com.example.autojob.job")
@AutoJobProcessorScan("com.example.autojob")
public class AutoJobMainApplication {
public static void main(String[] args) {
new AutoJobLauncherBuilder(AutoJobMainApplication.class)
.withAutoScanProcessor()
//配置你的调度器，如果你的调度器支持默认构造方法可以只指定类型
.addScheduler(YourScheduler.class)
//.addScheduler(new YourScheduler()) 如果不支持默认构造方法就需要添加一个实例
.build()
.run();
System.out.println("==================================>系统创建完成");
}
}
可能你希望框架只通过你的调度器来进行调度，而不再需要内存任务调度器或DB任务调度器，你可以在应用启动时选择性关闭：
@AutoJobScan("com.example.autojob.job")
@AutoJobProcessorScan("com.example.autojob")
public class AutoJobMainApplication {
public static void main(String[] args) {
new AutoJobBootstrap(AutoJobMainApplication.class)
.withAutoScanProcessor()
.closeDBTaskScheduler() // 关闭DB任务调度器
.closeMemoryTaskScheduler() // 关闭内存任务调度器
.build()
.run();
System.out.println("==================================>系统创建完成");
}
}
注意！！！如果你没有指定自己的调度器而关闭了框架原生的内存任务调度器或DB任务调度器，则框架会丧失该类型任务的调度功能，如果都关闭了则框架不再具有任何任务的调度功能。
3、自定义邮件报警
AutoJob中的邮件报警也是事件驱动的，框架发布相关报警事件->对应处理器创建邮件对象->发送，因此要实现自定义的邮件报警，只需要实现：自定义的报警事件、何时发布事件、报警事件处理器（模板的创建）。
所有的报警事件都继承于AlertEvent，下面我们看一下框架的任务运行错误报警的实现方式：
//定义报警事件
@Getter
@Setter
public class TaskRunErrorAlertEvent extends AlertEvent {
public TaskRunErrorAlertEvent(String title, String content, AutoJobTask errorTask) {
super(title, AlertEventLevel.WARN, content);
this.errorTask = errorTask;
}
private AutoJobTask errorTask;
private String stackTrace;
}
//报警事件的邮件模板创建
public static AlertMail newRunErrorAlertMail(TaskRunErrorAlertEvent event) {
AlertMailBuilder builder = AlertMailBuilder.newInstance();
AutoJobTask errorTask = event.getErrorTask();
return builder
.setMailClient(errorTask.getMailClient())
.setTitle(event.getTitle())
.setLevel(AlertEventLevel.WARN)
.addContentTitle(String.format("任务：\"%d:%s\"执行失败", errorTask.getId(), errorTask.getAlias()), 1)
.addBr()
.addBold("报警时间：" + DateUtils.formatDateTime(event.getPublishTime()))
.addBr()
.addBold(String.format("报警机器：%s:%s", event
.getNode()
.getHost(), event
.getNode()
.getPort()))
.addBr()
.addBold("任务路径：" + errorTask.getReference())
.addBr()
.addParagraph("堆栈信息如下：")
.addParagraph(event
.getStackTrace()
.replace("\n", "</br>"))
.addError("请及时处理")
.getAlertMail();
}
//事件处理器
@Slf4j
public class TaskRunErrorAlertEventHandler implements IAlertEventHandler<TaskRunErrorAlertEvent> {
@Override
public void doHandle(TaskRunErrorAlertEvent event) {
AutoJobConfig config = AutoJobApplication.getInstance().getConfigHolder().getAutoJobConfig();
if (!config.getTaskRunErrorAlert()) {
return;
}
AlertMail alertMail = AlertMailFactory.newRunErrorAlertMail(event);
if (alertMail != null) {
if (alertMail.send()) {
log.info("发送报警邮件成功");
} else {
log.error("发送报警邮件失败");
}
}
}
}
//事件处理器添加进上下文
public class AlertEventHandlerLoader implements IAutoJobLoader {
@Override
public void load() {
TaskEventHandlerDelegate
.getInstance()
.addHandler(TaskRunErrorEvent.class, new TaskRunErrorEventHandler());
}
}
//事件发布
public class TaskRunErrorEventHandler implements ITaskEventHandler<TaskRunErrorEvent> {
@Override
public void doHandle(TaskRunErrorEvent event) {
AlertEventHandlerDelegate
.getInstance()
.doHandle(AlertEventFactory.newTaskRunErrorAlertEvent(event.getTask(), event.getErrorStack()));
}
}
上面的代码大家需要关注几个地方：AlertMailBuilder是一个邮件模板构建类，可以构建一个邮件对象；报警事件处理器和任务事件处理器一样需要通过Processor添加进上下文。
4、自定义日志存储
框架默认日志的存储位置是数据库，你可以自己定义相关的存储策略和存储策略委派者，来实现日志在其他地方的存储。下面来简单演示：
//定义日志存储策略
/**
* 运行日志文件保存策略
*
* @Date 2022/11/21 9:15
*/
public class AutoJobLogFileStrategy implements IAutoJobLogSaveStrategy<AutoJobLog> {
@Override
public void doHandle(String taskPath, List<AutoJobLog> logList) {
try {
FileWriter fileWriter = new FileWriter(new File(taskPath));
//...
} catch (IOException e) {
e.printStackTrace();
}
}
}
/**
* 执行日志文件保存策略
*
* @Date 2022/11/21 9:15
*/
public class AutoJobRunLogFileStrategy implements IAutoJobLogSaveStrategy<AutoJobRunLog> {
@Override
public void doHandle(String taskPath, List<AutoJobRunLog> logList) {
try {
FileWriter fileWriter = new FileWriter(new File(taskPath));
//...
} catch (IOException e) {
e.printStackTrace();
}
}
}
//设置策略委派
public class AutoJobLogFileDelegate implements ILogSaveStrategyDelegate<AutoJobLog> {
@Override
public IAutoJobLogSaveStrategy<AutoJobLog> doDelegate(AutoJobConfigHolder configHolder, Class<AutoJobLog> type) {
//默认使用File保存策略
return new AutoJobLogFileStrategy();
}
}
public class AutoJobRunLogFileDelegate implements ILogSaveStrategyDelegate<AutoJobRunLog> {
@Override
public IAutoJobLogSaveStrategy<AutoJobRunLog> doDelegate(AutoJobConfigHolder configHolder, Class<AutoJobRunLog> type) {
//默认使用File保存策略
return new AutoJobRunLogFileStrategy();
}
}
以上代码定义好了文件的存储策略，那么如何使得我们的策略生效呢，这就需要我们再创建任务时把我们的策略委派给添加进上下文
public static void main(String[] args) {
new AutoJobBootstrap(AutoJobMainApplication.class)
.setLogSaveStrategyDelegate(new AutoJobLogFileDelegate()) //设置运行日志存储策略委派者
.setRunLogSaveStrategyDelegate(new AutoJobRunLogFileDelegate()) //设置执行日志存储策略委派者
.build()
.run();
}
将我们的日志存储策略委派者设置进去后，原有的存储策略就会被覆盖，当然如果你的委派者逻辑里面返回了AutoJobLogDBStrategy等原生的保存策略除外。
5、自定义任务过滤器
任务过滤器是一个过滤器链，在任务注册进任务调度队列前执行，主要功能是用于过滤某些不安全的任务的执行，框架提供了基于类路径过滤的任务过滤器ClassPathFilter，但是白名单只能在配置文件配置，因此很可能你希望实现一个动态的白名单配置，比如从数据库比对等等，这时你就需要继承AbstractRegisterFilter，如下示列：
//package com.example.spring.job
public class TestFilter extends AbstractRegisterFilter {
@Override
public void doHandle(AutoJobTask task) {
if(/*...*/){
//当该任务不允许注册时直接设置成不允许注册
task.setIsAllowRegister(false);
}
}
}
@SpringBootApplication
@AutoJobScan("com.example.spring.job")
@AutoJobRegisterPreProcessorScan("com.example.spring") //指定扫描包
public class AutoJobSpringApplication {
public static void main(String[] args) {
SpringApplication.run(AutoJobSpringApplication.class, args);
System.out.println("==================================>Spring应用已启动完成");
new AutoJobBootstrap(AutoJobSpringApplication.class)
.withAutoScanProcessor()
.build()
.run();
System.out.println("==================================>AutoJob应用已启动完成");
}
}
在创建应用时还需要在入口类上配置@AutoJobRegisterPreProcessorScan，指定注册前置处理器的扫描包路径，否则该过滤器不会被扫描到。
注意：子任务不会被该类过滤器处理。
6、注解任务开发的高级应用
在第三章节-第三小节-基于注解中，简单演示了注解@AutoJob的用法，AutoJob框架还提供了其他注解，如@FactoryAutoJob、@Conditional等，下面一一讲解。
@AutoJob注解
@Autojob注解是框架中使用最多的一个注解，将其标注在一个方法上，配置好调度信息，该方法就会在应用启动时将其包装成一个方法型任务放到对应的任务容器，可以参考下下面的示列。
@Slf4j
public class Jobs {
private static final AutoJobLogHelper logHelper = new AutoJobLogHelper();
static {
logHelper.setSlf4jProxy(log);
}
//立即启动，重复无限次，周期为5秒，使用自定义方法运行对象工厂，参数为"hello autoJob"
@AutoJob(id = 1, attributes = "{'hello autoJob'}", defaultStartTime = StartTime.NOW, repeatTimes = -1, cycle = 5, methodObjectFactory = SpringMethodObjectFactory.class)
public void hello(String str) {
logHelper.info(str);
}
//2022-11-21 12:00:00启动，重复3次，总共执行4次，周期为10秒，作为DB任务调度，最长允许运行时长5秒
@AutoJob(id = 2, startTime = "2022-11-21 12:00:00", repeatTimes = 3, cycle = 10, asType = AutoJobTask.TaskType.DB_TASK, maximumExecutionTime = 5000)
public void longTask() {
logHelper.info("long task start");
SyncHelper.sleepQuietly(10, TimeUnit.SECONDS);
logHelper.info("long task end");
}
//作为子任务调度
@AutoJob(id = 3, schedulingStrategy = SchedulingStrategy.AS_CHILD_TASK)
public void childTask() {
logHelper.info("child task start");
SyncHelper.sleepQuietly(3, TimeUnit.SECONDS);
logHelper.info("child task end");
}
//按照cron like表达式调度，重复无限次，子任务为3
@AutoJob(id = 4, alias = "获取随机字符串", cronExpression = "* * 0/5 17 * * ?", repeatTimes = -1, childTasksId = "3")
public String getRandomString() {
return StringUtils.getRandomStr(16);
}
//仅保存到数据库
@AutoJob(id = 4, schedulingStrategy = SchedulingStrategy.ONLY_SAVE)
public void error() {
String str = null;
str.length();
}
}
@FactoryAutoJob注解
由于@AutoJob的配置都是固定的，可能你希望能够动态配置任务的某些属性，因此@FactoryAutoJob就为了解决此类场景而出现的，当然你也可以使用基于构建的方式开发任务来实现动态，下面来看一个示列：
@FactoryAutoJob(RandomStringMethodFactory.class)
public String getRandomString() {
return StringUtils.getRandomStr(16);
}
public class RandomStringMethodFactory implements IMethodTaskFactory {
@Override
public MethodTask newTask(AutoJobConfigHolder configHolder, Method method) {
return new AutoJobMethodTaskBuilder(method.getDeclaringClass(), method.getName())
.setTaskId(IdGenerator.getNextIdAsLong())
//...
.build();
}
}
如上示列，getRandomString的包装将由RandomStringMethodFactory来进行。
@Conditional注解
相信经常使用Spring的小可耐们对此注解应该熟悉，在Spring中，该注解用于实现条件注入，即符合条件时该Bean才会注入到容器。在AutoJob中，功能类似，只有符合该注解指定条件的方法才能被包装成一个任务。
7、使用内置RPC框架
AutoJob的目标是一款分布式的任务调度框架，因此内部开发了通信框架：RPC， 这里只做简单介绍，后期会基于该RPC开发分布式的AutoJob。每一个AutoJob都有服务端和客户端，服务端的开启可以通过在配置文件里cluster.enable=true开启，要使用RPC框架首先需要开发服务提供类，如框架自带的API：
@AutoJobRPCService("MemoryTaskAPI") //通过该注解声明该类是一个RPC服务提供方
@Slf4j
public class MemoryTaskAPI implements AutoJobAPI {
//...细节省略
@Override
@RPCMethod("count") //声明该方法对外提供的方法名
//服务方法返回值和参数都得是包装类型
public Integer count() {
//...
}
}
其他AutoJob节点如何调用该服务呢，也非常简单，如下示列：
@AutoJobRPCClient("MemoryTaskAPI") //声明该接口是一个RPC客户端
public class MemoryTaskAPIClient{
//方法名同服务对外提供方法名相同
Integer count();
}
RPCClientProxy<MemoryTaskAPIClient> proxy = new RPCClientProxy<>("localhost", 7777, MemoryTaskAPIClient.class); //创建接口代理
MemoryTaskAPIClient client = proxy.clientProxy(); //获取代理实例
System.out.println(client.count()); //像本地方法一样使用
内嵌RPC框架基于netty开发，使用JSON进行序列化和反序列化。基础数据类型仅支持包装类型，即如int需要使用Integer。集合支持Map和List，支持泛型。目前RPC仅供学习使用。
8、使用基于时间动态调整的线程池封装
框架的执行池AutoJobTaskExecutorPool是任务执行的地方，其包含一个快池和一个慢池，分别用于执行运行时间短和运行时间长的任务。框架任务执行原生使用的是两个基于流量动态更新的线程池FlowThreadPoolExecutorHelper，为了更加适应业务需求，提供基于时间动态调整的线程池TimerThreadPoolExecutorPool。
TimerThreadPoolExecutorHelper.TimerEntry entry = new TimerThreadPoolExecutorHelper.TimerEntry("0 0 7 * * ?", 10, 20, 60, TimeUnit.SECONDS);//配置调整项，<0的项不作调整
//添加一个触发监听器
entry.setTriggerListener((cronExpression, threadPoolExecutor) -> {
System.out.println("日间线程池调整");
});
TimerThreadPoolExecutorHelper fastPool = TimerThreadPoolExecutorHelper
.builder()
.setInitialCoreTreadCount(3)
.setInitialMaximizeTreadCount(5)
.setTaskQueueCapacity(100)
.addTimerEntry("0 0 22 * * ?", 0, 1, -1, null)
.addTimerEntry(entry)
.build();
new AutoJobBootstrap(AutoJobSpringApplication.class)
.withAutoScanProcessor()
//自定义执行池
.setExecutorPool(new AutoJobTaskExecutorPool(null, fastPool, FlowThreadPoolExecutorHelper
.builder()
.build()))
.build()
.run();
System.out.println("==================================>AutoJob应用已启动完成");
lExecutorPool`。
TimerThreadPoolExecutorHelper.TimerEntry entry = new TimerThreadPoolExecutorHelper.TimerEntry("0 0 7 * * ?", 10, 20, 60, TimeUnit.SECONDS);//配置调整项，<0的项不作调整
//添加一个触发监听器
entry.setTriggerListener((cronExpression, threadPoolExecutor) -> {
System.out.println("日间线程池调整");
});
TimerThreadPoolExecutorHelper fastPool = TimerThreadPoolExecutorHelper
.builder()
.setInitialCoreTreadCount(3)
.setInitialMaximizeTreadCount(5)
.setTaskQueueCapacity(100)
.addTimerEntry("0 0 22 * * ?", 0, 1, -1, null)
.addTimerEntry(entry)
.build();
new AutoJobBootstrap(AutoJobSpringApplication.class)
.withAutoScanProcessor()
//自定义执行池
.setExecutorPool(new AutoJobTaskExecutorPool(null, fastPool, FlowThreadPoolExecutorHelper
.builder()
.build()))
.build()
.run();
System.out.println("==================================>AutoJob应用已启动完成");
如上示列，快池使用基于时间动态调整的线程池封装，其会在每天早上七点将线程池扩容到核心10线程，最大20线程，核心空闲时长更新为60秒，在每晚十点将线程池缩容到核心0线程，最大1线程并且添加了一个触发监听器；慢池使用基于流量调整线程池封装。
如果对你有帮助，谢谢点赞、收藏、Star ₍˄·͈༝·͈˄*₎◞ ̑̑
| 阿里云国内75折 回扣 微信号：monov8 |
| 阿里云国际，腾讯云国际，低至75折。AWS 93折 免费开户实名账号 代冲值 优惠多多 微信号：monov8 飞机：@monov6 |