feat: 2.6.0

1. 修复分布式锁异常问题

easy-retry-common/easy-retry-common-core/src/main/java/com/aizuda/easy/retry/common/core/enums/JobOperationReasonEnum.java

@@ -45,6 +45,6 @@ public enum JobOperationReasonEnum {
      */
     public static final List<Integer> WORKFLOW_SUCCESSOR_SKIP_EXECUTION = Arrays.asList(
         WORKFLOW_NODE_NO_REQUIRED.getReason(), WORKFLOW_DECISION_FAILED.getReason(),
-            WORKFLOW_CONDITION_NODE_EXECUTION_ERROR.reason);
+            WORKFLOW_CONDITION_NODE_EXECUTION_ERROR.getReason());
 
 }

easy-retry-server/easy-retry-server-common/src/main/java/com/aizuda/easy/retry/server/common/lock/AbstractLockProvider.java

@@ -1,9 +1,11 @@
 package com.aizuda.easy.retry.server.common.lock;
 
 import cn.hutool.core.lang.Assert;
+import com.aizuda.easy.retry.common.core.context.SpringContext;
 import com.aizuda.easy.retry.server.common.cache.CacheLockRecord;
 import com.aizuda.easy.retry.server.common.dto.LockConfig;
 import com.aizuda.easy.retry.server.common.exception.EasyRetryServerException;
+import org.springframework.transaction.support.TransactionTemplate;
 
 import java.time.Duration;
 
@@ -31,7 +33,6 @@ public abstract class AbstractLockProvider implements LockProvider {
 
         LockManager.setLockAtLeast(lockAtLeast);
         LockManager.setLockAtMost(lockAtMost);
-
         boolean tryToCreateLockRecord = !CacheLockRecord.lockRecordRecentlyCreated(lockName);
         if (tryToCreateLockRecord) {
             if (doLock(lockConfig)) {
@@ -42,7 +43,7 @@ public abstract class AbstractLockProvider implements LockProvider {
             CacheLockRecord.addLockRecord(lockName);
         }
 
-        boolean lock = false;
+        boolean lock;
         try {
             lock = doLockAfter(lockConfig);
         } catch (Exception e) {
@@ -51,10 +52,6 @@ public abstract class AbstractLockProvider implements LockProvider {
             }
 
             throw e;
-        } finally {
-            if (!lock) {
-                LockManager.clear();
-            }
         }
 
         return lock;

easy-retry-server/easy-retry-server-common/src/main/java/com/aizuda/easy/retry/server/common/lock/persistence/JdbcLockProvider.java

@@ -1,5 +1,6 @@
 package com.aizuda.easy.retry.server.common.lock.persistence;
 
+import com.aizuda.easy.retry.common.core.context.SpringContext;
 import com.aizuda.easy.retry.common.log.EasyRetryLog;
 import com.aizuda.easy.retry.server.common.Lifecycle;
 import com.aizuda.easy.retry.server.common.cache.CacheLockRecord;
@@ -18,7 +19,12 @@ import org.springframework.dao.DuplicateKeyException;
 import org.springframework.jdbc.BadSqlGrammarException;
 import org.springframework.jdbc.UncategorizedSQLException;
 import org.springframework.stereotype.Component;
+import org.springframework.transaction.TransactionDefinition;
+import org.springframework.transaction.TransactionStatus;
 import org.springframework.transaction.TransactionSystemException;
+import org.springframework.transaction.support.TransactionCallback;
+import org.springframework.transaction.support.TransactionCallbackWithoutResult;
+import org.springframework.transaction.support.TransactionTemplate;
 
 import java.time.LocalDateTime;
 import java.util.Arrays;
@@ -48,6 +54,9 @@ public class JdbcLockProvider implements LockStorage, Lifecycle {
 
     @Override
     public boolean createLock(LockConfig lockConfig) {
+        TransactionTemplate transactionTemplate = SpringContext.getBean(TransactionTemplate.class);
+        transactionTemplate.setPropagationBehavior(TransactionDefinition.PROPAGATION_NOT_SUPPORTED);
+        return Boolean.TRUE.equals(transactionTemplate.execute(status -> {
             try {
                 LocalDateTime now = lockConfig.getCreateDt();
                 DistributedLock distributedLock = new DistributedLock();
@@ -64,10 +73,15 @@ public class JdbcLockProvider implements LockStorage, Lifecycle {
                 EasyRetryLog.LOCAL.error("Unexpected exception. lockName:[{}]", lockConfig.getLockName(), e);
                 return false;
             }
+        }));
+
     }
 
     @Override
     public boolean renewal(LockConfig lockConfig) {
+        TransactionTemplate transactionTemplate = SpringContext.getBean(TransactionTemplate.class);
+        transactionTemplate.setPropagationBehavior(TransactionDefinition.PROPAGATION_NOT_SUPPORTED);
+        return Boolean.TRUE.equals(transactionTemplate.execute(status -> {
             LocalDateTime now = lockConfig.getCreateDt();
             DistributedLock distributedLock = new DistributedLock();
             distributedLock.setLockedBy(ServerRegister.CURRENT_CID);
@@ -82,29 +96,34 @@ public class JdbcLockProvider implements LockStorage, Lifecycle {
                      UncategorizedSQLException e) {
                 return false;
             }
+        }));
     }
 
     @Override
     public boolean releaseLockWithDelete(String lockName) {
-
+        TransactionTemplate transactionTemplate = SpringContext.getBean(TransactionTemplate.class);
+        transactionTemplate.setPropagationBehavior(TransactionDefinition.PROPAGATION_NOT_SUPPORTED);
+        return Boolean.TRUE.equals(transactionTemplate.execute(status -> {
             for (int i = 0; i < 10; i++) {
                 try {
-                CacheLockRecord.remove(lockName);
                     return distributedLockMapper.delete(new LambdaUpdateWrapper<DistributedLock>()
                             .eq(DistributedLock::getName, lockName)) > 0;
                 } catch (Exception e) {
                     EasyRetryLog.LOCAL.error("unlock error. retrying attempt [{}] ", i, e);
+                } finally {
+                    CacheLockRecord.remove(lockName);
                 }
             }
-
             return false;
+        }));
     }
 
     @Override
     public boolean releaseLockWithUpdate(String lockName, LocalDateTime lockAtLeast) {
-
+        TransactionTemplate transactionTemplate = SpringContext.getBean(TransactionTemplate.class);
+        transactionTemplate.setPropagationBehavior(TransactionDefinition.PROPAGATION_NOT_SUPPORTED);
         LocalDateTime now = LocalDateTime.now();
-
+        return Boolean.TRUE.equals(transactionTemplate.execute(status -> {
             for (int i = 0; i < 10; i++) {
                 try {
                     DistributedLock distributedLock = new DistributedLock();
@@ -118,6 +137,7 @@ public class JdbcLockProvider implements LockStorage, Lifecycle {
             }
 
             return false;
+        }));
     }
 
     @Override

easy-retry-server/easy-retry-server-common/src/main/java/com/aizuda/easy/retry/server/common/schedule/AbstractSchedule.java

@@ -7,6 +7,7 @@ import com.aizuda.easy.retry.server.common.Schedule;
 import com.aizuda.easy.retry.server.common.config.SystemProperties;
 import com.aizuda.easy.retry.server.common.exception.EasyRetryServerException;
 import com.aizuda.easy.retry.server.common.lock.LockBuilder;
+import com.aizuda.easy.retry.server.common.lock.LockManager;
 import com.aizuda.easy.retry.server.common.lock.LockProvider;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -52,6 +53,8 @@ public abstract class AbstractSchedule implements Schedule {
         } finally {
             if (lock) {
                 lockProvider.unlock();
+            } else {
+                LockManager.clear();
             }
         }
 

easy-retry-server/easy-retry-server-job-task/src/main/java/com/aizuda/easy/retry/server/job/task/support/dispatch/JobExecutorResultActor.java

@@ -73,11 +73,29 @@ public class JobExecutorResultActor extends AbstractActor {
 
                         Assert.isTrue(1 == jobTaskMapper.update(jobTask,
                                         new LambdaUpdateWrapper<JobTask>().eq(JobTask::getId, result.getTaskId())),
-                                ()-> new EasyRetryServerException("更新任务实例失败"));
+                                () -> new EasyRetryServerException("更新任务实例失败"));
-
+                        // 先尝试完成，若已完成则不需要通过获取分布式锁来完成
+                        boolean tryCompleteAndStop = tryCompleteAndStop(result);
+                        if (!tryCompleteAndStop) {
                             // 存在并发问题
                             distributedLockHandler.lockWithDisposableAndRetry(() -> {
-                            // 更新批次上的状态
+                                tryCompleteAndStop(result);
+                            }, MessageFormat.format(KEY, result.getTaskBatchId(),
+                                    result.getJobId()), Duration.ofSeconds(3), Duration.ofSeconds(1), 6);
+                        }
+                    }
+                });
+            } catch (Exception e) {
+                EasyRetryLog.LOCAL.error(" job executor result exception. [{}]", result, e);
+            } finally {
+                getContext().stop(getSelf());
+            }
+
+        }).build();
+
+    }
+
+    private boolean tryCompleteAndStop(JobExecutorResultDTO result) {
         CompleteJobBatchDTO completeJobBatchDTO = JobTaskConverter.INSTANCE.toCompleteJobBatchDTO(result);
         boolean complete = jobTaskBatchHandler.complete(completeJobBatchDTO);
         if (complete) {
@@ -91,16 +109,7 @@ public class JobExecutorResultActor extends AbstractActor {
                 instanceInterrupt.stop(stopJobContext);
             }
         }
-                        }, MessageFormat.format(KEY, result.getTaskBatchId(), result.getJobId()), Duration.ofSeconds(3), Duration.ofSeconds(1), 3);
 
-                    }
+        return complete;
-                });
-            } catch (Exception e) {
-                EasyRetryLog.LOCAL.error(" job executor result exception. [{}]", result, e);
-            } finally {
-                getContext().stop(getSelf());
-            }
-
-        }).build();
     }
 }

easy-retry-server/easy-retry-server-job-task/src/main/java/com/aizuda/easy/retry/server/job/task/support/dispatch/JobTaskPrepareActor.java

@@ -61,6 +61,7 @@ public class JobTaskPrepareActor extends AbstractActor {
                 prepare.getTaskExecutorScene());
         if (TaskTypeEnum.WORKFLOW.getType().equals(jobTaskExecutorSceneEnum.getTaskType().getType())) {
             queryWrapper.eq(JobTaskBatch::getWorkflowNodeId, prepare.getWorkflowNodeId());
+            queryWrapper.eq(JobTaskBatch::getWorkflowTaskBatchId, prepare.getWorkflowTaskBatchId());
             queryWrapper.eq(JobTaskBatch::getTaskType, TaskTypeEnum.WORKFLOW.getType());
         } else {
             queryWrapper.eq(JobTaskBatch::getTaskType, TaskTypeEnum.JOB.getType());

easy-retry-server/easy-retry-server-job-task/src/main/java/com/aizuda/easy/retry/server/job/task/support/dispatch/WorkflowExecutorActor.java

@@ -89,7 +89,8 @@ public class WorkflowExecutorActor extends AbstractActor {
 
         // 添加父节点，为了判断父节点的处理状态
         List<JobTaskBatch> allJobTaskBatchList = jobTaskBatchMapper.selectList(new LambdaQueryWrapper<JobTaskBatch>()
-                .select(JobTaskBatch::getWorkflowTaskBatchId, JobTaskBatch::getWorkflowNodeId, JobTaskBatch::getTaskBatchStatus)
+                .select(JobTaskBatch::getWorkflowTaskBatchId, JobTaskBatch::getWorkflowNodeId,
+                        JobTaskBatch::getTaskBatchStatus, JobTaskBatch::getOperationReason)
                 .eq(JobTaskBatch::getWorkflowTaskBatchId, workflowTaskBatch.getId())
                 .in(JobTaskBatch::getWorkflowNodeId, Sets.union(successors, Sets.newHashSet(taskExecute.getParentId())))
         );

easy-retry-server/easy-retry-server-job-task/src/main/java/com/aizuda/easy/retry/server/job/task/support/executor/workflow/AbstractWorkflowExecutor.java

@@ -60,7 +60,6 @@ public abstract class AbstractWorkflowExecutor implements WorkflowExecutor, Init
     private TransactionTemplate transactionTemplate;
 
     @Override
-    @Transactional
     public void execute(WorkflowExecutorContext context) {
         distributedLockHandler.lockWithDisposableAndRetry(
                 () -> {
@@ -111,7 +110,7 @@ public abstract class AbstractWorkflowExecutor implements WorkflowExecutor, Init
                         }
                     });
                 }, MessageFormat.format(KEY, context.getWorkflowTaskBatchId(), context.getWorkflowNodeId()),
-                Duration.ofSeconds(5), Duration.ofSeconds(1), 3);
+                Duration.ofSeconds(6), Duration.ofSeconds(2), 12);
 
     }
 

easy-retry-server/easy-retry-server-job-task/src/main/java/com/aizuda/easy/retry/server/job/task/support/handler/DistributedLockHandler.java

@@ -2,6 +2,7 @@ package com.aizuda.easy.retry.server.job.task.support.handler;
 
 import com.aizuda.easy.retry.common.log.EasyRetryLog;
 import com.aizuda.easy.retry.server.common.lock.LockBuilder;
+import com.aizuda.easy.retry.server.common.lock.LockManager;
 import com.aizuda.easy.retry.server.common.lock.LockProvider;
 import com.aizuda.easy.retry.server.job.task.support.LockExecutor;
 import com.github.rholder.retry.Attempt;
@@ -13,9 +14,10 @@ import com.github.rholder.retry.StopStrategies;
 import com.github.rholder.retry.WaitStrategies;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Component;
+import org.springframework.transaction.annotation.Propagation;
+import org.springframework.transaction.annotation.Transactional;
 
 import java.time.Duration;
-import java.util.concurrent.Callable;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
 
@@ -54,11 +56,17 @@ public class DistributedLockHandler {
             .withRetryListener(new RetryListener() {
                 @Override
                 public <V> void onRetry(final Attempt<V> attempt) {
-                    if (!attempt.hasResult()) {
+                    Object result = null;
-                        EasyRetryLog.LOCAL.warn("第【{}】次尝试获取锁. lockName:[{}]",
+                    if (attempt.hasResult()) {
-                            attempt.getAttemptNumber(), lockName);
+                        try {
+                            result = attempt.get();
+                        } catch (ExecutionException ignored) {
                         }
                     }
+
+                    EasyRetryLog.LOCAL.info("第【{}】次尝试获取锁. lockName:[{}] result:[{}] treadName:[{}]",
+                            attempt.getAttemptNumber(), lockName, result, Thread.currentThread().getName());
+                }
             }).build();
 
         boolean lock = false;
@@ -71,13 +79,20 @@ public class DistributedLockHandler {
             Throwable throwable = e;
             if (e.getClass().isAssignableFrom(RetryException.class)) {
                 RetryException re = (RetryException) e;
-                throwable = re.getLastFailedAttempt().getExceptionCause();
+                Attempt<?> lastFailedAttempt = re.getLastFailedAttempt();
+                if (lastFailedAttempt.hasException()) {
+                    throwable = lastFailedAttempt.getExceptionCause();
+                }
             }
 
             EasyRetryLog.LOCAL.error("lock execute error. lockName:[{}]", lockName, throwable);
         } finally {
             if (lock) {
+                EasyRetryLog.LOCAL.info("[{}] 锁已释放", lockName);
                 lockProvider.unlock();
+            } else {
+                // 未获取到锁直接清除线程中存储的锁信息
+                LockManager.clear();
             }
         }
 
@@ -90,6 +105,7 @@ public class DistributedLockHandler {
      * @param lockName 锁名称
      * @param lockAtMost 锁超时时间
      */
+    @Transactional(propagation = Propagation.NOT_SUPPORTED)
     public void lockWithDisposable(String lockName, Duration lockAtMost, LockExecutor lockExecutor) {
 
         LockProvider lockProvider = LockBuilder.newBuilder()
@@ -107,6 +123,9 @@ public class DistributedLockHandler {
         } finally {
             if (lock) {
                 lockProvider.unlock();
+            } else {
+                // 未获取到锁直接清除线程中存储的锁信息
+                LockManager.clear();
             }
         }
     }

frontend/public/lib/index.html

@@ -5,8 +5,8 @@
   <meta charset="UTF-8" />
   <meta name="viewport" content="width=device-width, initial-scale=1.0" />
   <title>Easy Retry</title>
-  <script type="module" crossorigin src="./assets/1RK3CeRh.js"></script>
+  <script type="module" crossorigin src="./assets/06nP_ETa.js"></script>
-  <link rel="stylesheet" crossorigin href="./assets/Z7fZzXo1.css">
+  <link rel="stylesheet" crossorigin href="./assets/dSfAux_s.css">
 </head>
 
 <body>

frontend/src/views/job/JobBatchLog.vue

@@ -135,6 +135,7 @@ export default {
             this.fromIndex = res.data.fromIndex
             if (res.data.message) {
               this.logList.push(...res.data.message)
+              this.logList.sort((a, b) => a.time_stamp - b.time_stamp)
             }
           })
           .catch(() => {
@@ -153,7 +154,7 @@ export default {
       const hours = date.getHours()
       const minutes = date.getMinutes().toString().length === 1 ? '0' + date.getMinutes() : date.getMinutes().toString()
       const seconds = date.getSeconds().toString().length === 1 ? '0' + date.getSeconds() : date.getSeconds().toString()
-      return `${year}-${month}-${day} ${hours}:${minutes}:${seconds}`
+      return `${year}-${month}-${day} ${hours}:${minutes}:${seconds}.${date.getMilliseconds()}`
     }
   }
 }