LTS(light-task-scheduler)主要用于解决分布式任务调度问题,支持实时任务,定时任务和Cron任务。有较好的伸缩性,扩展性,健壮稳定性而被多家公司使用,同时也希望开源爱好者一起贡献。
github地址: https://github.com/ltsopensource/light-task-scheduler
oschina地址: http://git.oschina.net/hugui/light-task-scheduler
这两个地址都会同步更新。感兴趣,请加QQ群:109500214 一起探讨、完善。越多人支持,就越有动力去更新,喜欢记得右上角star哈。
LTS 有主要有以下四种节点:
- JobClient:主要负责提交任务, 并接收任务执行反馈结果。
- JobTracker:负责接收并分配任务,任务调度。
- TaskTracker:负责执行任务,执行完反馈给JobTracker。
- LTS-Admin:(管理后台)主要负责节点管理,任务队列管理,监控管理等。
其中JobClient,JobTracker,TaskTracker节点都是无状态
的。
可以部署多个并动态的进行删减,来实现负载均衡,实现更大的负载量, 并且框架采用FailStore策略使LTS具有很好的容错能力。
LTS注册中心提供多种实现(Zookeeper,redis等),注册中心进行节点信息暴露,master选举。(Mongo or Mysql)存储任务队列和任务执行日志, netty or mina做底层通信, 并提供多种序列化方式fastjson, hessian2, java等。
LTS支持任务类型:
- 实时任务:提交了之后立即就要执行的任务。
- 定时任务:在指定时间点执行的任务,譬如 今天3点执行(单次)。
- Cron任务:CronExpression,和quartz类似(但是不是使用quartz实现的)譬如 0 0/1 * * * ?
支持动态修改任务参数,任务执行时间等设置,支持后台动态添加任务,支持Cron任务暂停,支持手动停止正在执行的任务(有条件),支持任务的监控统计,支持各个节点的任务执行监控,JVM监控等等.
###节点组
- 英文名称 NodeGroup,一个节点组等同于一个小的集群,同一个节点组中的各个节点是对等的,等效的,对外提供相同的服务。
- 每个节点组中都有一个master节点,这个master节点是由LTS动态选出来的,当一个master节点挂掉之后,LTS会立马选出另外一个master节点,框架提供API监听接口给用户。
###FailStore
- 顾名思义,这个主要是用于失败了存储的,主要用于节点容错,当远程数据交互失败之后,存储在本地,等待远程通信恢复的时候,再将数据提交。
- FailStore主要用户JobClient的任务提交,TaskTracker的任务反馈,TaskTracker的业务日志传输的场景下。
- FailStore目前提供几种实现:leveldb,rocksdb,berkeleydb,mapdb,ltsdb,用于可以自由选择使用哪种,用户也可以采用SPI扩展使用自己的实现。
下图是一个标准的实时任务执行流程。
目前后台带有由ztajy提供的一个简易的认证功能. 用户名密码在auth.cfg中,用户自行修改.
##特性
###1、Spring支持
LTS可以完全不用Spring框架,但是考虑到很用用户项目中都是用了Spring框架,所以LTS也提供了对Spring的支持,包括Xml和注解,引入lts-spring.jar
即可。
###2、业务日志记录器
在TaskTracker端提供了业务日志记录器,供应用程序使用,通过这个业务日志器,可以将业务日志提交到JobTracker,这些业务日志可以通过任务ID串联起来,可以在LTS-Admin中实时查看任务的执行进度。
###3、SPI扩展支持
SPI扩展可以达到零侵入,只需要实现相应的接口,并实现即可被LTS使用,目前开放出来的扩展接口有
- 对任务队列的扩展,用户可以不选择使用mysql或者mongo作为队列存储,也可以自己实现。
- 对业务日志记录器的扩展,目前主要支持console,mysql,mongo,用户也可以通过扩展选择往其他地方输送日志。
###4、故障转移
当正在执行任务的TaskTracker宕机之后,JobTracker会立马将分配在宕机的TaskTracker的所有任务再分配给其他正常的TaskTracker节点执行。
###5、节点监控
可以对JobTracker,TaskTracker节点进行资源监控,任务监控等,可以实时的在LTS-Admin管理后台查看,进而进行合理的资源调配。
###6、多样化任务执行结果支持
LTS框架提供四种执行结果支持,EXECUTE_SUCCESS
,EXECUTE_FAILED
,EXECUTE_LATER
,EXECUTE_EXCEPTION
,并对每种结果采取相应的处理机制,譬如重试。
- EXECUTE_SUCCESS: 执行成功,这种情况,直接反馈客户端(如果任务被设置了要反馈给客户端)。
- EXECUTE_FAILED:执行失败,这种情况,直接反馈给客户端,不进行重试。
- EXECUTE_LATER:稍后执行(需要重试),这种情况,不反馈客户端,重试策略采用1min,2min,3min的策略,默认最大重试次数为10次,用户可以通过参数设置修改这个重试次数。
- EXECUTE_EXCEPTION:执行异常, 这中情况也会重试(重试策略,同上)
###7、FailStore容错 采用FailStore机制来进行节点容错,Fail And Store,不会因为远程通信的不稳定性而影响当前应用的运行。具体FailStore说明,请参考概念说明中的FailStore说明。
##项目编译打包
项目主要采用maven进行构建,目前提供shell脚本的打包。
环境依赖:Java(jdk1.7)
Maven
用户使用一般分为两种: ###1、Maven构建 可以通过maven命令将lts的jar包上传到本地仓库中。在父pom.xml中添加相应的repository,并用deploy命令上传即可。具体引用方式可以参考lts中的例子即可。 ###2、直接Jar引用 需要将lts的各个模块打包成单独的jar包,并且将所有lts依赖包引入。具体引用哪些jar包可以参考lts中的例子即可。
##JobTracker和LTS-Admin部署
提供(cmd)windows
和(shell)linux
两种版本脚本来进行编译和部署:
-
运行根目录下的
sh build.sh
或build.cmd
脚本,会在dist
目录下生成lts-{version}-bin
文件夹 -
下面是其目录结构,其中bin目录主要是JobTracker和LTS-Admin的启动脚本。
jobtracker
中是 JobTracker的配置文件和需要使用到的jar包,lts-admin
是LTS-Admin相关的war包和配置文件。 lts-{version}-bin的文件结构
├── bin
│ ├── jobtracker.cmd
│ ├── jobtracker.sh
│ ├── lts-admin.cmd
│ └── lts-admin.sh
├── jobtracker
│ ├── conf
│ │ └── zoo
│ │ ├── jobtracker.cfg
│ │ └── log4j.properties
│ └── lib
│ └── *.jar
├── lts-admin
│ ├── conf
│ │ ├── log4j.properties
│ │ └── lts-admin.cfg
│ ├── lib
│ │ └── *.jar
│ └── lts-admin.war
└── tasktracker
├── bin
│ └── tasktracker.sh
├── conf
│ ├── log4j.properties
│ └── tasktracker.cfg
└── lib
└── *.jar
- JobTracker启动。如果你想启动一个节点,直接修改下
conf/zoo
下的配置文件,然后运行sh jobtracker.sh zoo start
即可,如果你想启动两个JobTracker节点,那么你需要拷贝一份zoo,譬如命名为zoo2
,修改下zoo2
下的配置文件,然后运行sh jobtracker.sh zoo2 start
即可。logs文件夹下生成jobtracker-zoo.out
日志。 - LTS-Admin启动.修改
lts-admin/conf
下的配置,然后运行bin
下的sh lts-admin.sh
或lts-admin.cmd
脚本即可。logs文件夹下会生成lts-admin.out
日志,启动成功在日志中会打印出访问地址,用户可以通过这个访问地址访问了。
##JobClient(部署)使用
需要引入lts的jar包有lts-jobclient-{version}.jar
,lts-core-{version}.jar
及其它第三方依赖jar。
###API方式启动
JobClient jobClient = new RetryJobClient();
jobClient.setNodeGroup("test_jobClient");
jobClient.setClusterName("test_cluster");
jobClient.setRegistryAddress("zookeeper://127.0.0.1:2181");
jobClient.start();
// 提交任务
Job job = new Job();
job.setTaskId("3213213123");
job.setParam("shopId", "11111");
job.setTaskTrackerNodeGroup("test_trade_TaskTracker");
// job.setCronExpression("0 0/1 * * * ?"); // 支持 cronExpression表达式
// job.setTriggerTime(new Date()); // 支持指定时间执行
Response response = jobClient.submitJob(job);
###Spring XML方式启动
<bean id="jobClient" class="com.lts.spring.JobClientFactoryBean">
<property name="clusterName" value="test_cluster"/>
<property name="registryAddress" value="zookeeper://127.0.0.1:2181"/>
<property name="nodeGroup" value="test_jobClient"/>
<property name="masterChangeListeners">
<list>
<bean class="com.lts.example.support.MasterChangeListenerImpl"/>
</list>
</property>
<property name="jobFinishedHandler">
<bean class="com.lts.example.support.JobFinishedHandlerImpl"/>
</property>
<property name="configs">
<props>
<!-- 参数 -->
<prop key="job.fail.store">leveldb</prop>
</props>
</property>
</bean>
###Spring 全注解方式
@Configuration
public class LTSSpringConfig {
@Bean(name = "jobClient")
public JobClient getJobClient() throws Exception {
JobClientFactoryBean factoryBean = new JobClientFactoryBean();
factoryBean.setClusterName("test_cluster");
factoryBean.setRegistryAddress("zookeeper://127.0.0.1:2181");
factoryBean.setNodeGroup("test_jobClient");
factoryBean.setMasterChangeListeners(new MasterChangeListener[]{
new MasterChangeListenerImpl()
});
Properties configs = new Properties();
configs.setProperty("job.fail.store", "leveldb");
factoryBean.setConfigs(configs);
factoryBean.afterPropertiesSet();
return factoryBean.getObject();
}
}
##TaskTracker(部署使用)
需要引入lts的jar包有lts-tasktracker-{version}.jar
,lts-core-{version}.jar
及其它第三方依赖jar。
###定义自己的任务执行类
public class MyJobRunner implements JobRunner {
private final static BizLogger bizLogger = LtsLoggerFactory.getBizLogger();
@Override
public Result run(Job job) throws Throwable {
try {
// TODO 业务逻辑
// 会发送到 LTS (JobTracker上)
bizLogger.info("测试,业务日志啊啊啊啊啊");
} catch (Exception e) {
return new Result(Action.EXECUTE_FAILED, e.getMessage());
}
return new Result(Action.EXECUTE_SUCCESS, "执行成功了,哈哈");
}
}
###API方式启动
TaskTracker taskTracker = new TaskTracker();
taskTracker.setJobRunnerClass(MyJobRunner.class);
taskTracker.setRegistryAddress("zookeeper://127.0.0.1:2181");
taskTracker.setNodeGroup("test_trade_TaskTracker");
taskTracker.setClusterName("test_cluster");
taskTracker.setWorkThreads(20);
taskTracker.start();
###Spring XML方式启动
<bean id="taskTracker" class="com.lts.spring.TaskTrackerAnnotationFactoryBean" init-method="start">
<property name="jobRunnerClass" value="com.lts.example.support.MyJobRunner"/>
<property name="bizLoggerLevel" value="INFO"/>
<property name="clusterName" value="test_cluster"/>
<property name="registryAddress" value="zookeeper://127.0.0.1:2181"/>
<property name="nodeGroup" value="test_trade_TaskTracker"/>
<property name="workThreads" value="20"/>
<property name="masterChangeListeners">
<list>
<bean class="com.lts.example.support.MasterChangeListenerImpl"/>
</list>
</property>
<property name="configs">
<props>
<prop key="job.fail.store">leveldb</prop>
</props>
</property>
</bean>
###Spring注解方式启动
@Configuration
public class LTSSpringConfig implements ApplicationContextAware {
private ApplicationContext applicationContext;
@Override
public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
this.applicationContext = applicationContext;
}
@Bean(name = "taskTracker")
public TaskTracker getTaskTracker() throws Exception {
TaskTrackerAnnotationFactoryBean factoryBean = new TaskTrackerAnnotationFactoryBean();
factoryBean.setApplicationContext(applicationContext);
factoryBean.setClusterName("test_cluster");
factoryBean.setJobRunnerClass(MyJobRunner.class);
factoryBean.setNodeGroup("test_trade_TaskTracker");
factoryBean.setBizLoggerLevel("INFO");
factoryBean.setRegistryAddress("zookeeper://127.0.0.1:2181");
factoryBean.setMasterChangeListeners(new MasterChangeListener[]{
new MasterChangeListenerImpl()
});
factoryBean.setWorkThreads(20);
Properties configs = new Properties();
configs.setProperty("job.fail.store", "leveldb");
factoryBean.setConfigs(configs);
factoryBean.afterPropertiesSet();
// factoryBean.start();
return factoryBean.getObject();
}
}
##参数说明
参数 | 是否必须 | 默认值 | 使用范围 | 设置方式 | 参数说明 |
---|---|---|---|---|---|
registryAddress | 必须 | 无 | JobClient,JobTracker,TaskTracker | setRegistryAddress("xxxx") | 注册中心,可以选用zk或者redis,参考值: zookeeper://127.0.0.1:2181 或 redis://127.0.0.1:6379 |
clusterName | 必须 | 无 | JobClient,JobTracker,TaskTracker | setClusterName("xxxx") | 集群名称,clusterName相同的所有节点才会组成整个LTS架构 |
listenPort | 必须 | 35001 | JobTracker | setListenPort(xxx) | JobTracker的远程监听端口 |
job.logger | 必须 | console | JobTracker | addConfig("job.logger","xxx") | LTS业务日志记录器,可选值console,mysql,mongo,或者自己实现SPI扩展 |
job.queue | 必须 | mongo | JobTracker | addConfig("job.queue", "xx") | LTS任务队列,可选值mongo,mysql,或者自己实现SPI扩展 |
jdbc.url | 可选 | 无 | JobTracker | addConfig("jdbc.url", "xxx") | mysql连接URL,当job.queue为mysql的时候起作用 |
jdbc.username | 可选 | 无 | JobTracker | addConfig("jdbc.username", "xxx") | mysql连接密码,当job.queue为mysql的时候起作用 |
jdbc.password | 可选 | 无 | JobTracker | addConfig("jdbc.password", "xxx") | mysql连接密码,当job.queue为mysql的时候起作用 |
mongo.addresses | 可选 | 无 | JobTracker | addConfig("mongo.addresses", "xxx") | mongo连接URL,当job.queue为mongo的时候起作用 |
mongo.database | 可选 | 无 | JobTracker | addConfig("mongo.database", "xxx") | mongo数据库名,当job.queue为mongo的时候起作用 |
zk.client | 可选 | zkclient | JobClient,JobTracker,TaskTracker | addConfig("zk.client", "xxx") | zookeeper客户端,可选值zkclient, curator |
job.pull.frequency | 可选 | 3 | TaskTracker | addConfig("job.pull.frequency", "xx") | TaskTracker去向JobTracker Pull任务的频率,针对不同的场景可以做相应的调整,单位秒 |
job.max.retry.times | 可选 | 10 | JobTracker | addConfig("job.max.retry.times", "xx") | 任务的最大重试次数 |
stop.working | 可选 | false | TaskTracker | addConfig("stop.working", "true") | 主要用于当TaskTracker与JobTracker出现网络隔离的时候,超过一定时间隔离之后,TaskTracker自动停止当前正在运行的任务 |
job.fail.store | 可选 | leveldb | JobClient,TaskTracker | addConfig("job.fail.store", "leveldb") | 可选值:leveldb(默认), rocksdb, berkeleydb, mapdb FailStore实现, leveldb有问题的同学,可以试试mapdb |
lazy.job.logger | 可选 | false | JobTracker | addConfig("lazy.job.logger", "true") | 可选值:ture,false, 是否延迟批量刷盘日志, 如果启用,采用队列的方式批量将日志刷盘(在应用关闭的时候,可能会造成日志丢失) |
dataPath | 可选 | user.home | JobClient,TaskTracker,JobTracker | setDataPath("xxxx") | FailStore文件存储路径及其它数据存储路径 |
lts.monitor.interval | 可选 | 1 | JobClient,TaskTracker,JobTracker | addConfig("lts.monitor.interval", "2") | 分钟,整数,建议1-5分钟 |
lts.remoting | 可选 | netty | JobClient,TaskTracker,JobTracker | addConfig("lts.remoting", "netty") | 底层通讯框架,可选值netty和mina,可以混用,譬如JobTracker是netty, JobClient采用mina |
lts.remoting.serializable.default | 可选 | fastjson | JobClient,TaskTracker,JobTracker | addConfig("lts.remoting.serializable.default", "fastjson") | 底层通讯默认序列化方式,可选值 fastjson, hessian2 ,java,底层会自动识别你请求的序列化方式,然后返回数据也是采用与请求的序列化方式返回,假设JobTracker设置的是fastjson,而JobClient是hessian2,那么JobClient提交任务的时候,序列化方式是hessian2,当JobTracker收到请求的时候采用hessian2解码,然后也会将响应数据采用hessian2编码返回给JobClient |
##使用建议 一般在一个JVM中只需要一个JobClient实例即可,不要为每种任务都新建一个JobClient实例,这样会大大的浪费资源,因为一个JobClient可以提交多种任务。相同的一个JVM一般也尽量保持只有一个TaskTracker实例即可,多了就可能造成资源浪费。当遇到一个TaskTracker要运行多种任务的时候,请参考下面的 "一个TaskTracker执行多种任务"。 ##一个TaskTracker执行多种任务 有的时候,业务场景需要执行多种任务,有些人会问,是不是要每种任务类型都要一个TaskTracker去执行。我的答案是否定的,如果在一个JVM中,最好使用一个TaskTracker去运行多种任务,因为一个JVM中使用多个TaskTracker实例比较浪费资源(当然当你某种任务量比较多的时候,可以将这个任务单独使用一个TaskTracker节点来执行)。那么怎么才能实现一个TaskTracker执行多种任务呢。下面是我给出来的参考例子。
/**
* 总入口,在 taskTracker.setJobRunnerClass(JobRunnerDispatcher.class)
* JobClient 提交 任务时指定 Job 类型 job.setParam("type", "aType")
*/
public class JobRunnerDispatcher implements JobRunner {
private static final ConcurrentHashMap<String/*type*/, JobRunner>
JOB_RUNNER_MAP = new ConcurrentHashMap<String, JobRunner>();
static {
JOB_RUNNER_MAP.put("aType", new JobRunnerA()); // 也可以从Spring中拿
JOB_RUNNER_MAP.put("bType", new JobRunnerB());
}
@Override
public Result run(Job job) throws Throwable {
String type = job.getParam("type");
return JOB_RUNNER_MAP.get(type).run(job);
}
}
class JobRunnerA implements JobRunner {
@Override
public Result run(Job job) throws Throwable {
// TODO A类型Job的逻辑
return null;
}
}
class JobRunnerB implements JobRunner {
@Override
public Result run(Job job) throws Throwable {
// TODO B类型Job的逻辑
return null;
}
}
##TaskTracker的JobRunner测试
一般在编写TaskTracker的时候,只需要测试JobRunner的实现逻辑是否正确,又不想启动LTS进行远程测试。为了方便测试,LTS提供了JobRunner的快捷测试方法。自己的测试类集成com.lts.tasktracker.runner.JobRunnerTester
即可,并实现initContext
和newJobRunner
方法即可。如lts-example
中的例子:
public class TestJobRunnerTester extends JobRunnerTester {
public static void main(String[] args) throws Throwable {
// 1. Mock Job 数据
Job job = new Job();
job.setTaskId("2313213");
// 2. 运行测试
TestJobRunnerTester tester = new TestJobRunnerTester();
Result result = tester.run(job);
System.out.println(JSONUtils.toJSONString(result));
}
@Override
protected void initContext() {
// TODO 初始化Spring容器等
}
@Override
protected JobRunner newJobRunner() {
return new TestJobRunner();
}
}
##Spring Quartz Cron任务无缝接入 对于Quartz的Cron任务只需要在Spring配置中增加一下代码就可以接入LTS平台
<bean class="com.lts.spring.quartz.QuartzLTSProxyBean">
<property name="clusterName" value="test_cluster"/>
<property name="registryAddress" value="zookeeper://127.0.0.1:2181"/>
<property name="nodeGroup" value="quartz_test_group"/>
</bean>
##多网卡选择问题 当机器有内网两个网卡的时候,有时候,用户想让LTS的流量走外网网卡,那么需要在host中,把主机名称的映射地址改为外网网卡地址即可,内网同理。
##打包成独立jar
请在lts-parent/lts
下install即可,会在 lts-parent/lts/target
下生成lts-{version}.jar
##关于节点标识问题
如果在节点启动的时候设置节点标识,LTS会默认设置一个UUID为节点标识,可读性会比较差,但是能保证每个节点的唯一性,如果用户能自己保证节点标识的唯一性,可以通过 setIdentity
来设置,譬如如果每个节点都是部署在一台机器(一个虚拟机)上,那么可以将identity设置为主机名称
##SPI扩展说明 支持JobLogger,JobQueue等等的SPI扩展
##和其它解决方案比较 ###和MQ比较 见docs/LTS业务场景说明.pdf ###和Quartz比较 见docs/LTS业务场景说明.pdf