performance sth

Author: 胡贵

lts-admin/src/test/java/com/lts/biz/logger/mysql/MysqlJobLoggerTest.java

@@ -14,7 +14,7 @@
 import static org.junit.Assert.*;
 
 /**
- * Created by hugui on 6/12/15.
+ * @author Robert HG ([email protected]) on 6/12/15.
  */
 public class MysqlJobLoggerTest {
 

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

@@ -1,10 +1,10 @@
 package com.lts.core.domain;
 
 
-import com.lts.core.exception.JobSubmitException;
-import com.lts.core.support.CronExpression;
 import com.lts.core.commons.utils.JSONUtils;
 import com.lts.core.commons.utils.StringUtils;
+import com.lts.core.exception.JobSubmitException;
+import com.lts.core.support.CronExpression;
 import com.lts.remoting.annotation.NotNull;
 
 import java.util.Date;
@@ -16,7 +16,6 @@
  */
 public class Job {
 
-    protected String jobId;
     @NotNull
     protected String taskId;
     /**
@@ -85,14 +84,6 @@ public void setNeedFeedback(boolean needFeedback) {
         this.needFeedback = needFeedback;
     }
 
-    public String getJobId() {
-        return jobId;
-    }
-
-    public void setJobId(String jobId) {
-        this.jobId = jobId;
-    }
-
     public Map<String, String> getExtParams() {
         return extParams;
     }

lts-core/src/main/java/com/lts/core/domain/JobResult.java

@@ -3,8 +3,8 @@
 import com.lts.core.commons.utils.JSONUtils;
 
 /**
- * @author Robert HG ([email protected]) on 8/19/14.
- * 任务执行结果
+ * @author Robert HG ([email protected]) on 6/13/15.
+ * 发送给客户端的 任务执行结果
  */
 public class JobResult {
 

lts-core/src/main/java/com/lts/core/domain/JobWrapper.java

@@ -0,0 +1,43 @@
+package com.lts.core.domain;
+
+/**
+ * @author Robert HG ([email protected]) on 6/13/15.
+ */
+public class JobWrapper {
+
+    private String jobId;
+
+    private Job job;
+
+    public JobWrapper(String jobId, Job job) {
+        this.jobId = jobId;
+        this.job = job;
+    }
+
+    public JobWrapper() {
+    }
+
+    public String getJobId() {
+        return jobId;
+    }
+
+    public void setJobId(String jobId) {
+        this.jobId = jobId;
+    }
+
+    public Job getJob() {
+        return job;
+    }
+
+    public void setJob(Job job) {
+        this.job = job;
+    }
+
+    @Override
+    public String toString() {
+        return "JobWrapper{" +
+                "jobId='" + jobId + '\'' +
+                ", job=" + job +
+                '}';
+    }
+}

lts-core/src/main/java/com/lts/core/domain/TaskTrackerJobResult.java

@@ -0,0 +1,56 @@
+package com.lts.core.domain;
+
+import com.lts.core.commons.utils.JSONUtils;
+
+/**
+ * @author Robert HG ([email protected]) on 8/19/14.
+ * TaskTracker 任务执行结果
+ */
+public class TaskTrackerJobResult {
+
+    private JobWrapper jobWrapper;
+
+    // 执行成功还是失败
+    private boolean success;
+
+    private String msg;
+    // 任务完成时间
+    private Long time;
+
+    public JobWrapper getJobWrapper() {
+        return jobWrapper;
+    }
+
+    public void setJobWrapper(JobWrapper jobWrapper) {
+        this.jobWrapper = jobWrapper;
+    }
+
+    public boolean isSuccess() {
+        return success;
+    }
+
+    public void setSuccess(boolean success) {
+        this.success = success;
+    }
+
+    public String getMsg() {
+        return msg;
+    }
+
+    public void setMsg(String msg) {
+        this.msg = msg;
+    }
+
+    public Long getTime() {
+        return time;
+    }
+
+    public void setTime(Long time) {
+        this.time = time;
+    }
+
+    @Override
+    public String toString() {
+        return JSONUtils.toJSONString(this);
+    }
+}

lts-core/src/main/java/com/lts/core/failstore/berkeleydb/BerkeleydbFailStore.java

@@ -1,6 +1,5 @@
 package com.lts.core.failstore.berkeleydb;
 
-import com.lts.core.cluster.Config;
 import com.lts.core.commons.file.FileLock;
 import com.lts.core.commons.file.FileUtils;
 import com.lts.core.commons.utils.CollectionUtils;

lts-core/src/main/java/com/lts/core/failstore/leveldb/LeveldbFailStore.java

@@ -39,6 +39,7 @@ public LeveldbFailStore(String failStorePath) {
         dbPath = FileUtils.createDirIfNotExist(failStorePath);
         options = new Options();
         options.createIfMissing(true);
+//        options.compressionType(CompressionType.NONE);
         options.cacheSize(100 * 1024 * 1024);   // 100M
 //        options.logger(new Logger() {
 //            @Override
@@ -107,10 +108,13 @@ public void delete(List<String> keys) throws FailStoreException {
 
     @Override
     public <T> List<KVPair<String, T>> fetchTop(int size, Type type) throws FailStoreException {
+        Snapshot snapshot = db.getSnapshot();
         DBIterator iterator = null;
         try {
             List<KVPair<String, T>> list = new ArrayList<KVPair<String, T>>(size);
-            iterator = db.iterator();
+            ReadOptions options=new ReadOptions();
+            options.snapshot(snapshot);
+            iterator = db.iterator(options);
             for (iterator.seekToFirst(); iterator.hasNext(); iterator.next()) {
                 Map.Entry<byte[], byte[]> entry = iterator.peekNext();
                 String key = new String(entry.getKey(), "UTF-8");
@@ -132,6 +136,11 @@ public <T> List<KVPair<String, T>> fetchTop(int size, Type type) throws FailStor
                     throw new FailStoreException(e);
                 }
             }
+            try {
+                snapshot.close();
+            } catch (IOException e) {
+                throw new FailStoreException(e);
+            }
         }
     }
 

lts-core/src/main/java/com/lts/core/protocol/command/JobPushRequest.java

@@ -1,6 +1,6 @@
 package com.lts.core.protocol.command;
 
-import com.lts.core.domain.Job;
+import com.lts.core.domain.JobWrapper;
 import com.lts.remoting.annotation.NotNull;
 
 /**
@@ -9,13 +9,13 @@
 public class JobPushRequest extends AbstractCommandBody{
 
     @NotNull
-    private Job job;
+    private JobWrapper jobWrapper;
 
-    public Job getJob() {
-        return job;
+    public JobWrapper getJobWrapper() {
+        return jobWrapper;
     }
 
-    public void setJob(Job job) {
-        this.job = job;
+    public void setJobWrapper(JobWrapper jobWrapper) {
+        this.jobWrapper = jobWrapper;
     }
 }

lts-core/src/main/java/com/lts/core/protocol/command/TtJobFinishedRequest.java

@@ -0,0 +1,55 @@
+package com.lts.core.protocol.command;
+
+import com.lts.core.domain.TaskTrackerJobResult;
+import com.lts.remoting.annotation.NotNull;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * @author Robert HG ([email protected]) on 8/16/14.
+ * TaskTracker Job finished request command body
+ */
+public class TtJobFinishedRequest extends AbstractCommandBody {
+    /**
+     * 是否接受新任务
+     */
+    private boolean receiveNewJob = false;
+
+    @NotNull
+    private List<TaskTrackerJobResult> taskTrackerJobResults;
+
+    // 是否是重发(重发是批量发)
+    private boolean reSend = false;
+
+    public boolean isReSend() {
+        return reSend;
+    }
+
+    public void setReSend(boolean reSend) {
+        this.reSend = reSend;
+    }
+
+    public boolean isReceiveNewJob() {
+        return receiveNewJob;
+    }
+
+    public void setReceiveNewJob(boolean receiveNewJob) {
+        this.receiveNewJob = receiveNewJob;
+    }
+
+    public List<TaskTrackerJobResult> getTaskTrackerJobResults() {
+        return taskTrackerJobResults;
+    }
+
+    public void setTaskTrackerJobResults(List<TaskTrackerJobResult> taskTrackerJobResults) {
+        this.taskTrackerJobResults = taskTrackerJobResults;
+    }
+
+    public void addJobResult(TaskTrackerJobResult taskTrackerJobResult) {
+        if (taskTrackerJobResults == null) {
+            taskTrackerJobResults = new ArrayList<TaskTrackerJobResult>();
+        }
+        taskTrackerJobResults.add(taskTrackerJobResult);
+    }
+}

lts-core/src/main/java/com/lts/core/support/RetryScheduler.java

@@ -93,9 +93,6 @@ public void stop() {
      */
     private class CheckRunner implements Runnable {
 
-        // 一次最多提交maxSentSize个, 保证文件所也能被其他线程拿到
-        private int maxSentSize = 100;
-
         @Override
         public void run() {
             try {
@@ -104,8 +101,6 @@ public void run() {
                     return;
                 }
 
-                int sentSize = 0;
-
                 List<KVPair<String, T>> kvPairs = null;
                 do {
                     try {
@@ -124,20 +119,11 @@ public void run() {
                             values.add(kvPair.getValue());
                         }
                         if (retry(values)) {
-                            LOGGER.info("{} local files send success, {}", name, JSONUtils.toJSONString(values));
+                            LOGGER.info("{} local files send success, size: {}, {}", name, values.size(), JSONUtils.toJSONString(values));
                             failStore.delete(keys);
                         } else {
                             break;
                         }
-                        sentSize += kvPairs.size();
-                        if (sentSize >= maxSentSize) {
-                            // 一次最多提交maxSentSize个, 保证文件所也能被其他线程拿到
-                            try {
-                                Thread.sleep(1000L);
-                            } catch (InterruptedException e1) {
-                                LOGGER.warn(e1.getMessage(), e1);
-                            }
-                        }
                     } finally {
                         failStore.close();
                     }
@@ -155,6 +141,7 @@ public void inSchedule(String key, T value) {
             try {
                 failStore.open();
                 failStore.put(key, value);
+                LOGGER.info("{}  local files save success, {}", name, JSONUtils.toJSONString(value));
             } finally {
                 failStore.close();
             }