Merge pull request #36 from qq254963746/develop

Develop

Author: qq254963746

README.md

@@ -1,106 +1,83 @@
 LTS 轻量级分布式任务调度框架(Light Task Schedule)
 -----------------
 
-## 框架概况：
+###框架概况：
  LTS是一个轻量级分布式任务调度框架，参考hadoop的部分思想。有三种角色, JobClient, JobTracker, TaskTracker。各个节点都是无状态的，可以部署多个，来实现负载均衡，实现更大的负载量, 并且框架具有很好的容错能力。
  采用多种注册中心（Zookeeper，redis等）进行节点信息暴露，master选举。(Mongo or Mysql)存储任务队列和任务执行日志, netty做底层通信。
 * JobClient : 主要负责提交任务, 和 接收任务执行反馈结果。
 * JobTracker : 负责接收并分配任务，任务调度。
 * TaskTracker: 负责执行任务，执行完反馈给JobTracker。
 
-框架支持实时任务，也支持定时任务，同时也支持CronExpression, 有问题，请加QQ群：109500214 一起完善，探讨
+支持任务类型:
+* 实时任务 
+* 也支持定时任务 (如：3天之后执行)
+* CronExpression (如:0 0/1 * * * ?) 
 
-##架构图
+感兴趣，请加QQ群：109500214 一起探讨、完善。并且记得star一下哈，3Q
+
+###架构图
 ![Aaron Swartz](https://raw.githubusercontent.com/qq254963746/light-task-schedule/master/doc/LTS_architecture.png)
-##节点组:
+####节点组:
 * 1. 一个节点组等同于一个集群，同一个节点组中的各个节点是对等的，外界无论连接节点组中的任务一个节点都是可以的。
 * 2. 每个节点组中都有一个master节点(master宕机，会自动选举出新的master节点)，框架会提供接口API来监听master节点的变化，用户可以自己使用master节点做自己想做的事情。
 * 3. JobClient和TaskTracker都可以存在多个节点组。譬如 JobClient 可以存在多个节点组。 譬如：JobClient 节点组为 ‘lts_WEB’ 中的一个节点提交提交一个 只有节点组为’lts_TRADE’的 TaskTracker 才能执行的任务。
 * 4. (每个集群中)JobTacker只有一个节点组。
 * 5. 多个JobClient节点组和多个TaskTracker节点组再加上一个JobTacker节点组, 组成一个大的集群。
 
-## 工作流程:
-* 1. JobClient 提交一个 任务 给 JobTracker, 这里我提供了两种客户端API, 一种是如果JobTracker 不存在或者提交失败，直接返回提交失败。另一种客户端是重试客户端, 如果提交失败，先存储到本地leveldb(可以使用NFS来达到同个节点组共享leveldb文件的目的,多线程访问，做了文件锁处理)，返回给客户端提交成功的信息，待JobTracker可用的时候，再将任务提交。
-* 2. JobTracker 收到JobClient提交来的任务，先生成一个唯一的JobID。然后将任务储存在Mongo集群中。JobTracker 发现有（任务执行的）可用的TaskTracker节点(组) 之后，将优先级最大，最先提交的任务分发给TaskTracker。这里JobTracker会优先分配给比较空闲的TaskTracker节点，达到负载均衡。
-* 3. TaskTracker 收到JobTracker分发来的任务之后，执行。执行完毕之后，再反馈任务执行结果给JobTracker（成功or 失败[失败有失败错误信息]），如果发现JobTacker不可用，那么存储本地leveldb，等待TaskTracker可用的时候再反馈。反馈结果的同时，询问JobTacker有没有新的任务要执行。
-* 4. JobTacker收到TaskTracker节点的任务结果信息，生成并插入(mongo or mysql)任务执行日志。根据任务信息决定要不要反馈给客户端。不需要反馈的直接删除, 需要反馈的（同样JobClient不可用存储文件，等待可用重发）。
-* 5. JobClient 收到任务执行结果，进行自己想要的逻辑处理。
-
-## 特性
+###工作流程:
+* 1. JobClient 提交一个 任务 给 JobTracker, 这里我提供了两种客户端API, 一种是如果JobTracker 不存在或者提交失败，直接返回提交失败。另一种客户端是重试客户端, 如果提交失败，先存储到本地FailStore(可以使用NFS来达到同个节点组共享leveldb文件的目的,多线程访问,已经做了文件锁处理)，返回给客户端提交成功的信息，待JobTracker可用的时候，再将任务提交。
+* 2. JobTracker收到JobClient提交来的任务,将任务存入任务队列。JobTracker等待TaskTracker的Pull请求，然后将任务Push给TaskTracker去执行。
+* 3. TaskTracker收到JobTracker分发来的任务之后，然后从线程池中拿到一个线程去执行。执行完毕之后，再反馈任务执行结果给JobTracker（成功or 失败[失败有失败错误信息]），如果发现JobTacker不可用，那么存储本地FailStore，等待TaskTracker可用的时候再反馈。反馈结果的同时，询问JobTacker有没有新的任务要执行。
+* 4. JobTacker收到TaskTracker节点的任务结果信息。根据任务信息决定要不要反馈给客户端。不需要反馈的直接删除,需要反馈的,直接反馈,反馈失败进入FeedbackQueue, 等待重新反馈。
+* 5. JobClient收到任务执行结果，进行自己想要的逻辑处理。
 
+###特性
 * 负载均衡:
-     * JobClient 和 TaskTracker会随机连接JobTracker节点组中的一个节点，实现JobTracker负载均衡。当连接上后，将一直保持连接这个节点,保持连接通道，直到这个节点不可用，减少每次都重新连接一个节点带来的性能开销。
-     * JobTracker 分发任务时，是优先分配给最空闲的一个TaskTracker节点，实现TaskTracker节点的负载均衡。
+     * JobClient和TaskTracker可是根据自己设置的负载均衡策略来请求JobTracker节点组中的一个节点。当连接上后将一直保持连接这个节点,保持连接通道，直到这个节点不可用,减少每次都重新连接一个节点带来的性能开销。
 
 * 健壮性:
      * 当节点组中的一个节点当机之后，自动转到其他节点工作。当整个节点组当机之后，将会采用存储文件的方式，待节点组可用的时候进行重发。
-     * 当执行任务的TaskTracker节点当机之后，JobTracker 会将这个TaskTracker上的未完成的任务(死任务)，重新分配给节点组中其他节点执行。
+     * 当执行任务的TaskTracker节点当机之后，JobTracker会将这个TaskTracker上的未完成的任务(死任务)，重新分配给节点组中其他节点执行。
 
 * 伸缩性：
      * 因为各个节点都是无状态的，可以动态增加机器部署实例, 节点关注者会自动发现。
 * 扩展性:
      * 采用和dubbo一样的SPI扩展方式，可以实现任务队列扩展，日志记录器扩展等
 
-## 开发计划：
-* WEB后台管理
-* 框架优化
-
-## 调用示例
-* 安装 zookeeper(或redis) 和 mongo(或mysql) (后提供其他任务队列实现方式)
+###日志记录
+对于任务的分发，执行，还有用户通过 (BizLogger)  【LtsLoggerFactory.getBizLogger()】 输入的业务日志，LTS都有记录，用户可以在LTS Admin 后台界面查看某个任务的所有日志，可以实时查看这个任务的执行情况。 
 
-运行 job-example模块中的例子（包含API启动例子和Spring例子）
-分别执行 JobTrackerTest TaskTrackerTest JobClientTest
+###开发计划：
+* WEB后台管理：性能统计分析，预警等
+* 实现LTS的分布式队列存储
 
-这里给出的是java API(设置配置)方式启动, (spring启动和面添加)
+###LTS Admin
+![Aaron Swartz](https://raw.githubusercontent.com/qq254963746/light-task-schedule/master/doc/LTS_Admin.png)
+###调用示例
+下面提供的是最简单的配置方式。更多配置请查看 lts-example 模块下的 API 调用方式例子.
 
-## JobTracker 端
+####JobTracker 端
 ```java
     final JobTracker jobTracker = new JobTracker();
     // 节点信息配置
     jobTracker.setRegistryAddress("zookeeper://127.0.0.1:2181");
-    // jobTracker.setRegistryAddress("redis://127.0.0.1:6379");
-    jobTracker.setListenPort(35002); // 默认 35001
-    jobTracker.setClusterName("test_cluster");
-    jobTracker.addMasterChangeListener(new MasterChangeListenerImpl());
-    // 设置业务日志记录
-    //  jobTracker.addConfig("job.logger", "console");  // 默认
-    //  jobTracker.addConfig("job.logger", "mysql");
-    //  jobTracker.addConfig("job.logger", "mongo");
-    
     // 1. 任务队列用mongo
     jobTracker.addConfig("job.queue", "mongo");
     // mongo 配置
-    jobTracker.addConfig("mongo.addresses", "127.0.0.1:27017");     // 多个地址用逗号分割
-    jobTracker.addConfig("mongo.database", "job");
-    
-    // 2. 任务队里用mysql
-    // jobTracker.addConfig("job.queue", "mysql");
-    // mysql 配置
-    // jobTracker.addConfig("jdbc.url", "jdbc:mysql://127.0.0.1:3306/lts");
-    // jobTracker.addConfig("jdbc.username", "root");
-    // jobTracker.addConfig("jdbc.password", "root");
-    
+    jobTracker.addConfig("mongo.addresses", "127.0.0.1:27017"); 
+    jobTracker.addConfig("mongo.database", "lts");
     jobTracker.setOldDataHandler(new OldDataDeletePolicy());
-    // 设置 zk 客户端用哪个， 可选 zkclient, curator 默认是 zkclient
-    jobTracker.addConfig("zk.client", "zkclient");
     // 启动节点
     jobTracker.start();
 ```
 
-## TaskTracker端
+#### TaskTracker端
 ```java
     TaskTracker taskTracker = new TaskTracker();
     taskTracker.setJobRunnerClass(TestJobRunner.class);
-    // jobClient.setClusterName("lts");
     taskTracker.setRegistryAddress("zookeeper://127.0.0.1:2181");
-    // taskTracker.setRegistryAddress("redis://127.0.0.1:6379");
     taskTracker.setNodeGroup("test_trade_TaskTracker");
-    taskTracker.setClusterName("test_cluster");
     taskTracker.setWorkThreads(20);
-    // taskTracker.addConfig("job.fail.store", "leveldb");     // 默认
-    // taskTracker.addConfig("job.fail.store", "berkeleydb");
-    // taskTracker.addConfig("job.fail.store", "rocksdb");
-    // taskTracker.setLoadBalance("consistenthash");
     taskTracker.start();
     // 任务执行类
     public class TestJobRunner implements JobRunner {
@@ -117,18 +94,12 @@ LTS 轻量级分布式任务调度框架(Light Task Schedule)
     }
 ```
 
-## JobClient端
+#### JobClient端
 ```java
     JobClient jobClient = new RetryJobClient();
     // final JobClient jobClient = new JobClient();
     jobClient.setNodeGroup("test_jobClient");
-    jobClient.setClusterName("test_cluster");
     jobClient.setRegistryAddress("zookeeper://127.0.0.1:2181");
-    // jobClient.setRegistryAddress("redis://127.0.0.1:6379");
-    // jobClient.addConfig("job.fail.store", "leveldb");     // 默认
-    // jobClient.addConfig("job.fail.store", "berkeleydb");
-    // jobClient.addConfig("job.fail.store", "rocksdb");
-    // jobClient.setLoadBalance("consistenthash");
     jobClient.start();
 
     // 提交任务
@@ -137,7 +108,7 @@ LTS 轻量级分布式任务调度框架(Light Task Schedule)
     job.setParam("shopId", "11111");
     job.setTaskTrackerNodeGroup("test_trade_TaskTracker");
     // job.setCronExpression("0 0/1 * * * ?");  // 支持 cronExpression表达式
-    // job.setTriggerTime(new Date().getTime()); // 支持指定时间执行
+    // job.setTriggerTime(new Date()); // 支持指定时间执行
     Response response = jobClient.submitJob(job);
 ```
 

doc/LTS_Admin.png

@@ -7,6 +7,7 @@
 import com.lts.job.core.commons.utils.StringUtils;
 import com.lts.job.remoting.annotation.NotNull;
 
+import java.util.Date;
 import java.util.HashMap;
 import java.util.Map;
 
@@ -126,6 +127,12 @@ public boolean isSchedule() {
         return this.cronExpression != null && !"".equals(this.cronExpression.trim());
     }
 
+    public void setTriggerTime(Date date){
+        if(date != null){
+            this.triggerTime = date.getTime();
+        }
+    }
+
     public Long getTriggerTime() {
         return triggerTime;
     }

lts-core/src/main/java/com/lts/job/core/domain/Job.java

@@ -121,6 +121,9 @@ public void invokeAsync(RemotingCommand request, InvokeCallback invokeCallback)
             remotingClient.invokeAsync(jobTracker.getAddress(), request,
                     application.getConfig().getInvokeTimeoutMillis(), invokeCallback);
             this.serverEnable = true;
+            // publish msg
+            EventInfo eventInfo = new EventInfo(EcTopic.NO_JOB_TRACKER_AVAILABLE);
+            application.getEventCenter().publishAsync(eventInfo);
 
         } catch (Throwable e) {
             // 将这个JobTracker移除
@@ -152,6 +155,9 @@ public void invokeOneway(RemotingCommand request)
             remotingClient.invokeOneway(jobTracker.getAddress(), request,
                     application.getConfig().getInvokeTimeoutMillis());
             this.serverEnable = true;
+            // publish msg
+            EventInfo eventInfo = new EventInfo(EcTopic.NO_JOB_TRACKER_AVAILABLE);
+            application.getEventCenter().publishAsync(eventInfo);
 
         } catch (Throwable e) {
             // 将这个JobTracker移除

lts-core/src/main/java/com/lts/job/core/remoting/RemotingClientDelegate.java

@@ -40,8 +40,7 @@ public class JobPusher {
 
     public JobPusher(JobTrackerApplication application) {
         this.application = application;
-        executor = Executors.newFixedThreadPool(Constants.AVAILABLE_PROCESSOR * 5
-                , new NamedThreadFactory(JobPusher.class.getSimpleName()));
+        executor = Executors.newFixedThreadPool(Constants.AVAILABLE_PROCESSOR * 5, new NamedThreadFactory(JobPusher.class.getSimpleName()));
     }
 
     public void push(final RemotingServerDelegate remotingServer, final JobPullRequest request) {
@@ -69,9 +68,10 @@ private void pushJob(RemotingServerDelegate remotingServer, JobPullRequest reque
         String nodeGroup = request.getNodeGroup();
         String identity = request.getIdentity();
         // 更新TaskTracker的可用线程数
-        application.getTaskTrackerManager().updateTaskTrackerAvailableThreads(nodeGroup, identity, request.getAvailableThreads(), request.getTimestamp());
-
-        TaskTrackerNode taskTrackerNode = application.getTaskTrackerManager().getTaskTrackerNode(nodeGroup, identity);
+        application.getTaskTrackerManager().updateTaskTrackerAvailableThreads(nodeGroup,
+                identity, request.getAvailableThreads(), request.getTimestamp());
+        TaskTrackerNode taskTrackerNode = application.getTaskTrackerManager().
+                getTaskTrackerNode(nodeGroup, identity);
 
         if (taskTrackerNode == null) {
             return;
@@ -81,24 +81,20 @@ private void pushJob(RemotingServerDelegate remotingServer, JobPullRequest reque
 
         while (availableThreads > 0) {
             // 推送任务
-            int code = sendJob(remotingServer, taskTrackerNode);
-            if (code == NO_JOB) {
-                // 没有可以执行的任务, 直接停止
-                break;
-            }
-            if (code == PUSH_FAILED) {
+            PushResult result = sendJob(remotingServer, taskTrackerNode);
+            if (result == PushResult.SUCCESS) {
+                availableThreads = taskTrackerNode.getAvailableThread().decrementAndGet();
+            } else {
                 break;
             }
-            availableThreads = taskTrackerNode.getAvailableThread().get();
         }
     }
 
-    // 没有任务可执行
-    private final int NO_JOB = 1;
-    // 推送成功
-    private final int PUSH_SUCCESS = 2;
-    // 推送失败
-    private final int PUSH_FAILED = 3;
+    private enum PushResult {
+        NO_JOB, // 没有任务可执行
+        SUCCESS, //推送成功
+        FAILED      //推送失败
+    }
 
     /**
      * 是否推送成功
@@ -107,7 +103,7 @@ private void pushJob(RemotingServerDelegate remotingServer, JobPullRequest reque
      * @param taskTrackerNode
      * @return
      */
-    private int sendJob(RemotingServerDelegate remotingServer, TaskTrackerNode taskTrackerNode) {
+    private PushResult sendJob(RemotingServerDelegate remotingServer, TaskTrackerNode taskTrackerNode) {
 
         String nodeGroup = taskTrackerNode.getNodeGroup();
         String identity = taskTrackerNode.getIdentity();
@@ -116,7 +112,7 @@ private int sendJob(RemotingServerDelegate remotingServer, TaskTrackerNode taskT
         JobPo jobPo = application.getExecutableJobQueue().take(nodeGroup, identity);
 
         if (jobPo == null) {
-            return NO_JOB;
+            return PushResult.NO_JOB;
         }
 
         JobPushRequest body = application.getCommandBodyWrapper().wrapper(new JobPushRequest());
@@ -162,7 +158,7 @@ public void operationComplete(ResponseFuture responseFuture) {
                 LOGGER.debug("Job push failed! nodeGroup=" + nodeGroup + ", identity=" + identity + ", job=" + job);
             }
             application.getExecutableJobQueue().resume(jobPo);
-            return PUSH_FAILED;
+            return PushResult.FAILED;
         }
         try {
             application.getExecutingJobQueue().add(jobPo);
@@ -178,6 +174,6 @@ public void operationComplete(ResponseFuture responseFuture) {
         jobLogPo.setLevel(Level.INFO);
         application.getJobLogger().log(jobLogPo);
 
-        return PUSH_SUCCESS;
+        return PushResult.SUCCESS;
     }
 }