Merge pull request #10 from qq254963746/develop

添加 mysql JobLogger 实现

Author: qq254963746

job-core/src/main/java/com/lts/job/core/file/FileUtils.java

@@ -1,7 +1,6 @@
 package com.lts.job.core.file;
 
-import java.io.File;
-import java.io.IOException;
+import java.io.*;
 
 /**
  * @author Robert HG ([email protected]) on 3/6/15.
@@ -32,5 +31,13 @@ public static File createDirIfNotExist(String path) {
         return file;
     }
 
-
+    public static String read(InputStream is) throws IOException {
+        BufferedReader br = new BufferedReader(new InputStreamReader(is));
+        StringBuilder createTableSql = new StringBuilder();
+        String data = null;
+        while ((data = br.readLine()) != null) {
+            createTableSql.append(data);
+        }
+        return createTableSql.toString();
+    }
 }

job-core/src/main/java/com/lts/job/core/util/JSONUtils.java

@@ -13,6 +13,9 @@
 public class JSONUtils {
 
     public static <T> T parse(String json, Type type) {
+        if (StringUtils.isEmpty(json)) {
+            return null;
+        }
         return (T) JSONObject.parseObject(json, type);
     }
 
@@ -24,18 +27,30 @@ public static <T> T parse(String json, TypeReference<T> type) {
     }
 
     public static String toJSONString(Object obj) {
+        if (obj == null) {
+            return null;
+        }
         return JSONObject.toJSONString(obj);
     }
 
     public static JSONObject toJSON(Object obj) {
+        if (obj == null) {
+            return null;
+        }
         return (JSONObject) JSONObject.toJSON(obj);
     }
 
     public static JSONArray parseArray(String obj) {
+        if (obj == null) {
+            return null;
+        }
         return JSON.parseArray(obj);
     }
 
     public static JSONObject parseObject(String obj) {
+        if (obj == null) {
+            return null;
+        }
         return JSON.parseObject(obj);
     }
 

job-example/pom.xml

@@ -37,6 +37,11 @@
             <artifactId>job-logger-console</artifactId>
             <version>${project.version}</version>
         </dependency>
+        <dependency>
+            <groupId>com.lts</groupId>
+            <artifactId>job-logger-mysql</artifactId>
+            <version>${project.version}</version>
+        </dependency>
         <dependency>
             <groupId>com.lts</groupId>
             <artifactId>job-queue-mongo</artifactId>

job-example/src/main/java/com/lts/job/example/api/JobClientTest.java

@@ -34,7 +34,7 @@ public static void console() throws IOException {
 //        jobClient.setJobInfoSavePath(Constants.USER_HOME);
         jobClient.setJobFinishedHandler(new JobFinishedHandlerImpl());
         jobClient.addMasterChangeListener(new MasterChangeListenerImpl());
-        jobClient.setLoadBalance("consistenthash");
+//        jobClient.setLoadBalance("consistenthash");
         jobClient.start();
 
         JobClientTest jobClientTest = new JobClientTest();

job-example/src/main/java/com/lts/job/example/api/JobTrackerTest.java

@@ -53,7 +53,7 @@ public void run() {
     }
 
     /**
-     * 使用mongo做任务队列
+     * 使用mysql做任务队列
      */
     public static void testMysqlQueue() {
         final JobTracker jobTracker = new JobTracker();
@@ -65,8 +65,8 @@ public static void testMysqlQueue() {
 
         jobTracker.addMasterChangeListener(new MasterChangeListenerImpl());
 
-        // 设置业务日志记录
-//        jobTracker.addConfig("job.logger", "mongo");
+        // 设置业务日志记录 mysql
+        jobTracker.addConfig("job.logger", "mysql");
         // 任务队列用mysql
         jobTracker.addConfig("job.queue", "mysql");
         // mysql 配置

job-logger/job-logger-api/src/main/java/com/lts/job/biz/logger/JobLogException.java

@@ -0,0 +1,27 @@
+package com.lts.job.biz.logger;
+
+/**
+ * @author Robert HG ([email protected]) on 5/21/15.
+ */
+public class JobLogException extends RuntimeException {
+
+    public JobLogException() {
+        super();
+    }
+
+    public JobLogException(String message) {
+        super(message);
+    }
+
+    public JobLogException(String message, Throwable cause) {
+        super(message, cause);
+    }
+
+    public JobLogException(Throwable cause) {
+        super(cause);
+    }
+
+    protected JobLogException(String message, Throwable cause, boolean enableSuppression, boolean writableStackTrace) {
+        super(message, cause, enableSuppression, writableStackTrace);
+    }
+}

job-logger/job-logger-api/src/main/java/com/lts/job/biz/logger/JobLogUtils.java

@@ -0,0 +1,27 @@
+package com.lts.job.biz.logger;
+
+import com.lts.job.biz.logger.domain.BizLogPo;
+import com.lts.job.biz.logger.domain.JobLogPo;
+import com.lts.job.biz.logger.domain.LogType;
+
+/**
+ * @author Robert HG ([email protected]) on 5/21/15.
+ */
+public class JobLogUtils {
+
+    public static JobLogPo bizConvert(BizLogPo bizLogPo) {
+        if (bizLogPo == null) {
+            return null;
+        }
+        JobLogPo jobLogPo = new JobLogPo();
+        jobLogPo.setTimestamp(bizLogPo.getTimestamp());
+        jobLogPo.setTaskTrackerNodeGroup(bizLogPo.getTaskTrackerNodeGroup());
+        jobLogPo.setTaskTrackerIdentity(bizLogPo.getTaskTrackerIdentity());
+        jobLogPo.setJobId(bizLogPo.getJobId());
+        jobLogPo.setMsg(bizLogPo.getMsg());
+        jobLogPo.setLevel(bizLogPo.getLevel());
+        jobLogPo.setLogType(LogType.BIZ);
+        return jobLogPo;
+    }
+
+}

job-logger/job-logger-api/src/main/java/com/lts/job/biz/logger/domain/LogType.java

@@ -7,6 +7,7 @@ public enum LogType {
 
     FINISHED,        // 任务执行完成
     RESEND,          // TaskTracker 重新发送的任务执行结果
-    FIXED_DEAD       // 修复死掉的任务
+    FIXED_DEAD,       // 修复死掉的任务
+    BIZ             // 业务日志
     ;
 }

job-logger/job-logger-mongo/src/main/java/com/lts/job/biz/logger/mongo/MongoJobLogger.java

@@ -1,9 +1,11 @@
 package com.lts.job.biz.logger.mongo;
 
 
+import com.lts.job.biz.logger.JobLogUtils;
 import com.lts.job.biz.logger.JobLogger;
 import com.lts.job.biz.logger.domain.BizLogPo;
 import com.lts.job.biz.logger.domain.JobLogPo;
+import com.lts.job.biz.logger.domain.LogType;
 import com.lts.job.core.cluster.Config;
 import com.lts.job.store.mongo.AbstractMongoRepository;
 
@@ -23,13 +25,7 @@ public void log(JobLogPo jobLogPo) {
 
     @Override
     public void log(BizLogPo bizLogPo) {
-        JobLogPo jobLogPo = new JobLogPo();
-        jobLogPo.setTimestamp(bizLogPo.getTimestamp());
-        jobLogPo.setTaskTrackerNodeGroup(bizLogPo.getTaskTrackerNodeGroup());
-        jobLogPo.setTaskTrackerIdentity(bizLogPo.getTaskTrackerIdentity());
-        jobLogPo.setJobId(bizLogPo.getJobId());
-        jobLogPo.setMsg(bizLogPo.getMsg());
-        ds.save(jobLogPo);
+        ds.save(JobLogUtils.bizConvert(bizLogPo));
     }
 
 }

job-logger/job-logger-mysql/pom.xml

@@ -0,0 +1,30 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <parent>
+        <artifactId>job-logger</artifactId>
+        <groupId>com.lts</groupId>
+        <version>1.4.0-SNAPSHOT</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>job-logger-mysql</artifactId>
+
+    <dependencies>
+        <dependency>
+            <groupId>com.lts</groupId>
+            <artifactId>job-logger-api</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>commons-dbutils</groupId>
+            <artifactId>commons-dbutils</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>mysql</groupId>
+            <artifactId>mysql-connector-java</artifactId>
+        </dependency>
+    </dependencies>
+
+</project>