Merge pull request #31 from qq254963746/develop

定时任务在LTS Admin 中的添加和删除

Author: qq254963746

lts-core/src/main/java/com/lts/job/core/cluster/AbstractJobNode.java

@@ -58,10 +58,10 @@ final public void start() {
 
             registry.register(node);
 
-            LOGGER.info("started!");
+            LOGGER.info("Start success!");
 
         } catch (Throwable e) {
-            LOGGER.error("start failed!", e);
+            LOGGER.error("Start failed!", e);
         }
     }
 
@@ -72,18 +72,19 @@ final public void stop() {
             innerStop();
             remotingStop();
 
-            LOGGER.info("stop success!");
+            LOGGER.info("Stop success!");
         } catch (Throwable e) {
-            LOGGER.error("stop failed!", e);
+            LOGGER.error("Stop failed!", e);
         }
     }
 
     @Override
     public void destroy() {
         try {
             registry.destroy();
+            LOGGER.info("Destroy success!");
         } catch (Throwable e) {
-            LOGGER.error("destroy failed!", e);
+            LOGGER.error("Destroy failed!", e);
         }
     }
 

lts-core/src/main/java/com/lts/job/remoting/common/RemotingHelper.java

@@ -11,7 +11,7 @@
  */
 public class RemotingHelper {
 
-    public static final String RemotingLogName = "LTSRemoting";
+    public static final String RemotingLogName = "LTS.Remoting";
 
     public static String exceptionSimpleDesc(final Exception e) {
         StringBuffer sb = new StringBuffer();

lts-job-client/src/main/java/com/lts/job/client/support/JobSubmitProtector.java

@@ -36,9 +36,10 @@ public JobSubmitProtector(int concurrentSize) {
     }
 
     public Response execute(final List<Job> jobs, final JobSubmitExecutor<Response> jobSubmitExecutor) throws JobSubmitException {
+        boolean acquire = false;
         try {
             try {
-                boolean acquire = semaphore.tryAcquire(timeout, TimeUnit.MILLISECONDS);
+                acquire = semaphore.tryAcquire(timeout, TimeUnit.MILLISECONDS);
                 if (!acquire) {
                     throw new JobSubmitProtectException(concurrentSize, errorMsg);
                 }
@@ -47,7 +48,9 @@ public Response execute(final List<Job> jobs, final JobSubmitExecutor<Response>
             }
             return jobSubmitExecutor.execute(jobs);
         } finally {
-            semaphore.release();
+            if (acquire) {
+                semaphore.release();
+            }
         }
     }
 

lts-queue/lts-queue-mongo/src/main/java/com/lts/job/queue/mongo/MongoExecutableJobQueue.java

@@ -76,8 +76,9 @@ public boolean add(JobPo jobPo) {
 
     @Override
     public JobPo take(String taskTrackerNodeGroup, String taskTrackerIdentity) {
+        boolean acquire = false;
         try {
-            boolean acquire = semaphore.tryAcquire(acquireTimeout, TimeUnit.MILLISECONDS);
+            acquire = semaphore.tryAcquire(acquireTimeout, TimeUnit.MILLISECONDS);
             if (!acquire) {
                 // 直接返回null
                 return null;
@@ -99,7 +100,9 @@ public JobPo take(String taskTrackerNodeGroup, String taskTrackerIdentity) {
                             .set("gmtModified", System.currentTimeMillis());
             return template.findAndModify(query, operations, false);
         } finally {
-            semaphore.release();
+            if (acquire) {
+                semaphore.release();
+            }
         }
     }
 

lts-queue/lts-queue-mysql/src/main/java/com/lts/job/queue/mysql/MysqlExecutableJobQueue.java

@@ -110,8 +110,9 @@ public boolean add(JobPo jobPo) {
 
     @Override
     public JobPo take(final String taskTrackerNodeGroup, final String taskTrackerIdentity) {
+        boolean acquire = false;
         try {
-            boolean acquire = semaphore.tryAcquire(acquireTimeout, TimeUnit.MILLISECONDS);
+            acquire = semaphore.tryAcquire(acquireTimeout, TimeUnit.MILLISECONDS);
             if (!acquire) {
                 // 直接返回null
                 return null;
@@ -157,7 +158,9 @@ public JobPo take(final String taskTrackerNodeGroup, final String taskTrackerIde
         } catch (SQLException e) {
             throw new JobQueueException(e);
         } finally {
-            semaphore.release();
+            if(acquire){
+                semaphore.release();
+            }
         }
     }