---> 底部有招人帖

项目地址

1.7.2-SNAPSHOT(master)变更主要点

框架概况

架构图

概念说明

节点组

FailStore

流程图

LTS-Admin新版界面预览

特性

1、Spring支持

2、业务日志记录器

3、SPI扩展支持

4、故障转移

5、节点监控

6、多样化任务执行结果支持

7、FailStore容错

项目编译打包

1、Maven构建

2、直接Jar引用

JobTracker和LTS-Admin部署

--lts-${version}-bin    |--bin    |   |--jobtracker.cmd    |   |--jobtracker.sh    |   |--lts-admin.cmd    |   |--lts-admin.sh    |   |--lts-monitor.cmd    |   |--lts-monitor.sh    |   |--tasktracker.sh    |--conf    |   |--log4j.properties    |   |--lts-admin.cfg    |   |--lts-monitor.cfg    |   |--readme.txt    |   |--tasktracker.cfg    |   |--zoo    |       |--jobtracker.cfg    |       |--log4j.properties    |       |--lts-monitor.cfg    |--lib    |   |-- *.jar    |--war        |--jetty        |   |--lib        |       |-- *.jar        |--lts-admin.war

JobClient（部署）使用

API方式启动

JobClientjobClient =newRetryJobClient();jobClient.setNodeGroup("test_jobClient");jobClient.setClusterName("test_cluster");jobClient.setRegistryAddress("zookeeper://127.0.0.1:2181");jobClient.start();// 提交任务Jobjob =newJob();job.setTaskId("3213213123");job.setParam("shopId","11111");job.setTaskTrackerNodeGroup("test_trade_TaskTracker");// job.setCronExpression("0 0/1 * * * ?");  // 支持 cronExpression表达式// job.setTriggerTime(new Date()); // 支持指定时间执行Responseresponse =jobClient.submitJob(job);

Spring XML方式启动

<beanid="jobClient"class="com.github.ltsopensource.spring.JobClientFactoryBean">    <propertyname="clusterName"value="test_cluster"/>    <propertyname="registryAddress"value="zookeeper://127.0.0.1:2181"/>    <propertyname="nodeGroup"value="test_jobClient"/>    <propertyname="masterChangeListeners">        <list>            <beanclass="com.github.ltsopensource.example.support.MasterChangeListenerImpl"/>        </list>    </property>    <propertyname="jobFinishedHandler">        <beanclass="com.github.ltsopensource.example.support.JobFinishedHandlerImpl"/>    </property>    <propertyname="configs">        <props>            <!--参数 -->            <propkey="job.fail.store">leveldb</prop>        </props>    </property></bean>

Spring 全注解方式

@ConfigurationpublicclassLTSSpringConfig {@Bean(name ="jobClient")publicJobClientgetJobClient()throwsException {JobClientFactoryBeanfactoryBean =newJobClientFactoryBean();factoryBean.setClusterName("test_cluster");factoryBean.setRegistryAddress("zookeeper://127.0.0.1:2181");factoryBean.setNodeGroup("test_jobClient");factoryBean.setMasterChangeListeners(newMasterChangeListener[]{newMasterChangeListenerImpl()        });Propertiesconfigs =newProperties();configs.setProperty("job.fail.store","leveldb");factoryBean.setConfigs(configs);factoryBean.afterPropertiesSet();returnfactoryBean.getObject();    }}

TaskTracker(部署使用)

定义自己的任务执行类

publicclassMyJobRunnerimplementsJobRunner {@OverridepublicResultrun(JobContextjobContext)throwsThrowable {try {// TODO 业务逻辑// 会发送到 LTS (JobTracker上)jobContext.getBizLogger().info("测试，业务日志啊啊啊啊啊");        }catch (Exceptione) {returnnewResult(Action.EXECUTE_FAILED,e.getMessage());        }returnnewResult(Action.EXECUTE_SUCCESS,"执行成功了，哈哈");    }}

API方式启动

TaskTrackertaskTracker =newTaskTracker();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方式启动

<beanid="taskTracker"class="com.github.ltsopensource.spring.TaskTrackerAnnotationFactoryBean"init-method="start">    <propertyname="jobRunnerClass"value="com.github.ltsopensource.example.support.MyJobRunner"/>    <propertyname="bizLoggerLevel"value="INFO"/>    <propertyname="clusterName"value="test_cluster"/>    <propertyname="registryAddress"value="zookeeper://127.0.0.1:2181"/>    <propertyname="nodeGroup"value="test_trade_TaskTracker"/>    <propertyname="workThreads"value="20"/>    <propertyname="masterChangeListeners">        <list>            <beanclass="com.github.ltsopensource.example.support.MasterChangeListenerImpl"/>        </list>    </property>    <propertyname="configs">        <props>            <propkey="job.fail.store">leveldb</prop>        </props>    </property></bean>

Spring注解方式启动

@ConfigurationpublicclassLTSSpringConfigimplementsApplicationContextAware {privateApplicationContextapplicationContext;@OverridepublicvoidsetApplicationContext(ApplicationContextapplicationContext)throwsBeansException {this.applicationContext =applicationContext;    }@Bean(name ="taskTracker")publicTaskTrackergetTaskTracker()throwsException {TaskTrackerAnnotationFactoryBeanfactoryBean =newTaskTrackerAnnotationFactoryBean();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(newMasterChangeListener[]{newMasterChangeListenerImpl()        });factoryBean.setWorkThreads(20);Propertiesconfigs =newProperties();configs.setProperty("job.fail.store","leveldb");factoryBean.setConfigs(configs);factoryBean.afterPropertiesSet();//        factoryBean.start();returnfactoryBean.getObject();    }}

参数说明

使用建议

一个TaskTracker执行多种任务

/** * 总入口，在 taskTracker.setJobRunnerClass(JobRunnerDispatcher.class) * JobClient 提交 任务时指定 Job 类型  job.setParam("type", "aType") */publicclassJobRunnerDispatcherimplementsJobRunner {privatestaticfinalConcurrentHashMap<String/*type*/,JobRunner>JOB_RUNNER_MAP =newConcurrentHashMap<String,JobRunner>();static {JOB_RUNNER_MAP.put("aType",newJobRunnerA());// 也可以从Spring中拿JOB_RUNNER_MAP.put("bType",newJobRunnerB());    }@OverridepublicResultrun(JobContextjobContext)throwsThrowable {Jobjob =jobContext.getJob();Stringtype =job.getParam("type");returnJOB_RUNNER_MAP.get(type).run(job);    }}classJobRunnerAimplementsJobRunner {@OverridepublicResultrun(JobContextjobContext)throwsThrowable {//  TODO A类型Job的逻辑returnnull;    }}classJobRunnerBimplementsJobRunner {@OverridepublicResultrun(JobContextjobContext)throwsThrowable {// TODO B类型Job的逻辑returnnull;    }}

TaskTracker的JobRunner测试

publicclassTestJobRunnerTesterextendsJobRunnerTester {publicstaticvoidmain(String[]args)throwsThrowable {//  Mock Job 数据Jobjob =newJob();job.setTaskId("2313213");JobContextjobContext =newJobContext();jobContext.setJob(job);JobExtInfojobExtInfo =newJobExtInfo();jobExtInfo.setRetry(false);jobContext.setJobExtInfo(jobExtInfo);// 运行测试TestJobRunnerTestertester =newTestJobRunnerTester();Resultresult =tester.run(jobContext);System.out.println(JSON.toJSONString(result));    }@OverrideprotectedvoidinitContext() {// TODO 初始化Spring容器    }@OverrideprotectedJobRunnernewJobRunner() {returnnewTestJobRunner();    }}

Spring Quartz Cron任务无缝接入

<beanclass="com.github.ltsopensource.spring.quartz.QuartzLTSProxyBean">    <propertyname="clusterName"value="test_cluster"/>    <propertyname="registryAddress"value="zookeeper://127.0.0.1:2181"/>    <propertyname="nodeGroup"value="quartz_test_group"/></bean>

Spring Boot 支持

@SpringBootApplication@EnableJobTracker// 启动JobTracker@EnableJobClient// 启动JobClient@EnableTaskTracker// 启动TaskTracker@EnableMonitor// 启动MonitorpublicclassApplication {publicstaticvoidmain(String[]args) {SpringApplication.run(Application.class,args);    }}

多网卡选择问题

关于节点标识问题

SPI扩展说明

和其它解决方案比较

LTS-Admin使用jetty启动(默认)，不定期挂掉解决方案

招人！！！