optimize: seata-server timing tasks

common/src/main/java/io/seata/common/ConfigurationKeys.java

@@ -364,9 +364,24 @@ public interface ConfigurationKeys {
     String RECOVERY_PREFIX = SERVER_PREFIX + "recovery.";
 
     /**
-     * The constant HANDLE_ALL_SESSION_PERIOD.
+     * The constant COMMITING_RETRY_PERIOD.
      */
-    String HANDLE_ALL_SESSION_PERIOD = RECOVERY_PREFIX + "handleAllSessionPeriod";
+    String COMMITING_RETRY_PERIOD = RECOVERY_PREFIX + "committingRetryPeriod";
+
+    /**
+     * The constant ASYN_COMMITING_RETRY_PERIOD.
+     */
+    String ASYN_COMMITING_RETRY_PERIOD = RECOVERY_PREFIX + "asynCommittingRetryPeriod";
+
+    /**
+     * The constant ROLLBACKING_RETRY_PERIOD.
+     */
+    String ROLLBACKING_RETRY_PERIOD = RECOVERY_PREFIX + "rollbackingRetryPeriod";
+
+    /**
+     * The constant TIMEOUT_RETRY_PERIOD.
+     */
+    String TIMEOUT_RETRY_PERIOD = RECOVERY_PREFIX + "timeoutRetryPeriod";
 
     /**
      * The constant CLIENT_UNDO_PREFIX.
@@ -377,6 +392,7 @@ public interface ConfigurationKeys {
      * The constant TRANSACTION_UNDO_DATA_VALIDATION.
      */
     String TRANSACTION_UNDO_DATA_VALIDATION = CLIENT_UNDO_PREFIX + "dataValidation";
+
     /**
      * The constant TRANSACTION_UNDO_LOG_SERIALIZATION.
      */

server/src/main/java/io/seata/server/coordinator/DefaultCoordinator.java

@@ -78,8 +78,10 @@
 import org.slf4j.LoggerFactory;
 import org.slf4j.MDC;
 
-
-import static io.seata.common.Constants.HANDLE_ALL_SESSION;
+import static io.seata.common.Constants.ASYNC_COMMITTING;
+import static io.seata.common.Constants.RETRY_COMMITTING;
+import static io.seata.common.Constants.RETRY_ROLLBACKING;
+import static io.seata.common.Constants.TX_TIMEOUT_CHECK;
 import static io.seata.common.Constants.UNDOLOG_DELETE;
 
 /**
@@ -92,10 +94,27 @@ public class DefaultCoordinator extends AbstractTCInboundHandler implements Tran
     private static final int TIMED_TASK_SHUTDOWN_MAX_WAIT_MILLS = 5000;
 
     /**
-     * The constant HANDLE_ALL_SESSION_PERIOD.
+     * The constant COMMITTING_RETRY_PERIOD.
+     */
+    protected static final long COMMITTING_RETRY_PERIOD = CONFIG.getLong(ConfigurationKeys.COMMITING_RETRY_PERIOD,
+        1000L);
+
+    /**
+     * The constant ASYNC_COMMITTING_RETRY_PERIOD.
+     */
+    protected static final long ASYNC_COMMITTING_RETRY_PERIOD = CONFIG.getLong(
+        ConfigurationKeys.ASYN_COMMITING_RETRY_PERIOD, 1000L);
+
+    /**
+     * The constant ROLLBACKING_RETRY_PERIOD.
+     */
+    protected static final long ROLLBACKING_RETRY_PERIOD = CONFIG.getLong(ConfigurationKeys.ROLLBACKING_RETRY_PERIOD,
+        1000L);
+
+    /**
+     * The constant TIMEOUT_RETRY_PERIOD.
      */
-    protected static final long HANDLE_ALL_SESSION_PERIOD = CONFIG.getLong(ConfigurationKeys.HANDLE_ALL_SESSION_PERIOD,
-            1000L);
+    protected static final long TIMEOUT_RETRY_PERIOD = CONFIG.getLong(ConfigurationKeys.TIMEOUT_RETRY_PERIOD, 1000L);
 
     /**
      * The Transaction undo log delete period.
@@ -129,30 +148,26 @@ public class DefaultCoordinator extends AbstractTCInboundHandler implements Tran
     private static final boolean ROLLBACK_RETRY_TIMEOUT_UNLOCK_ENABLE = ConfigurationFactory.getInstance().getBoolean(
             ConfigurationKeys.ROLLBACK_RETRY_TIMEOUT_UNLOCK_ENABLE, false);
 
-    private final ExecutorService retryRollbacking = new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS,
-        new LinkedBlockingQueue<>(), new NamedThreadFactory("RetryRollbacking", 1));
-
-    private final ExecutorService retryCommitting = new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS,
-        new LinkedBlockingQueue<>(), new NamedThreadFactory("RetryCommitting", 1));
+    private final ScheduledThreadPoolExecutor retryRollbacking =
+        new ScheduledThreadPoolExecutor(1, new NamedThreadFactory("RetryRollbacking", 1));
 
-    private final ExecutorService asyncCommitting = new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS,
-        new LinkedBlockingQueue<>(), new NamedThreadFactory("AsyncCommitting", 1));
+    private final ScheduledThreadPoolExecutor retryCommitting =
+        new ScheduledThreadPoolExecutor(1, new NamedThreadFactory("RetryCommitting", 1));
 
-    private final ExecutorService timeoutCheck = new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS,
-        new LinkedBlockingQueue<>(), new NamedThreadFactory("TxTimeoutCheck", 1));
+    private final ScheduledThreadPoolExecutor asyncCommitting =
+        new ScheduledThreadPoolExecutor(1, new NamedThreadFactory("AsyncCommitting", 1));
 
-    private final ScheduledThreadPoolExecutor undoLogDelete = new ScheduledThreadPoolExecutor(1,
-            new NamedThreadFactory("UndoLogDelete", 1));
+    private final ScheduledThreadPoolExecutor timeoutCheck =
+        new ScheduledThreadPoolExecutor(1, new NamedThreadFactory("TxTimeoutCheck", 1));
 
-    private final ScheduledThreadPoolExecutor handleAllSession = new ScheduledThreadPoolExecutor(1,
-            new NamedThreadFactory("HandleAllSession", 1));
+    private final ScheduledThreadPoolExecutor undoLogDelete =
+        new ScheduledThreadPoolExecutor(1, new NamedThreadFactory("UndoLogDelete", 1));
 
-    private final List<GlobalStatus> rollbackingStatuses =
-        Collections.unmodifiableList(Arrays.asList(GlobalStatus.TimeoutRollbacking,
-            GlobalStatus.TimeoutRollbackRetrying, GlobalStatus.RollbackRetrying, GlobalStatus.Rollbacking));
+    private final GlobalStatus[] rollbackingStatuses =
+        new GlobalStatus[] {GlobalStatus.TimeoutRollbacking,
+            GlobalStatus.TimeoutRollbackRetrying, GlobalStatus.RollbackRetrying, GlobalStatus.Rollbacking};
 
-    private final List<GlobalStatus> retryCommittingStatuses = Collections.unmodifiableList(
-        Arrays.asList(GlobalStatus.Committing, GlobalStatus.CommitRetrying));
+    private final GlobalStatus[] retryCommittingStatuses = new GlobalStatus[]{GlobalStatus.Committing, GlobalStatus.CommitRetrying};
 
     private final ThreadPoolExecutor branchRemoveExecutor = new ThreadPoolExecutor(BRANCH_ASYNC_POOL_SIZE, BRANCH_ASYNC_POOL_SIZE,
             Integer.MAX_VALUE, TimeUnit.MILLISECONDS,
@@ -293,78 +308,14 @@ protected void doLockCheck(GlobalLockQueryRequest request, GlobalLockQueryRespon
                 core.lockQuery(request.getBranchType(), request.getResourceId(), request.getXid(), request.getLockKey()));
     }
 
-    /**
-     * Handle all session.
-     */
-    protected void handleAllSession() {
-        SessionCondition sessionCondition = new SessionCondition(GlobalStatus.values());
-        sessionCondition.setLazyLoadBranch(true);
-        Collection<GlobalSession> allSessions =
-            SessionHolder.getRootSessionManager().findGlobalSessions(sessionCondition);
-        if (CollectionUtils.isEmpty(allSessions)) {
-            return;
-        }
-        List<GlobalSession> retryRollbackingSessions = new ArrayList<>();
-        List<GlobalSession> beginGlobalSessions = new ArrayList<>();
-        List<GlobalSession> retryCommittingSessions = new ArrayList<>();
-        List<GlobalSession> asyncCommittingSessions = new ArrayList<>();
-        for (GlobalSession session : allSessions) {
-            if (rollbackingStatuses.contains(session.getStatus())) {
-                retryRollbackingSessions.add(session);
-            } else if (retryCommittingStatuses.contains(session.getStatus())) {
-                retryCommittingSessions.add(session);
-            } else if (GlobalStatus.AsyncCommitting.equals(session.getStatus())) {
-                asyncCommittingSessions.add(session);
-            } else if (GlobalStatus.Begin.equals(session.getStatus())) {
-                beginGlobalSessions.add(session);
-            }
-        }
-        List<CompletableFuture<Void>> futures = new ArrayList<>(4);
-        if (!retryRollbackingSessions.isEmpty()) {
-            futures.add(
-                CompletableFuture.runAsync(() -> handleRetryRollbacking(retryRollbackingSessions), retryRollbacking));
-        }
-        if (!beginGlobalSessions.isEmpty()) {
-            futures.add(CompletableFuture.runAsync(() -> timeoutCheck(beginGlobalSessions), timeoutCheck));
-        }
-        if (!retryCommittingSessions.isEmpty()) {
-            futures.add(
-                CompletableFuture.runAsync(() -> handleRetryCommitting(retryCommittingSessions), retryCommitting));
-        }
-        if (!asyncCommittingSessions.isEmpty()) {
-            futures.add(
-                CompletableFuture.runAsync(() -> handleAsyncCommitting(asyncCommittingSessions), asyncCommitting));
-        }
-        if (CollectionUtils.isNotEmpty(futures)) {
-            try {
-                CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).get();
-            } catch (InterruptedException e) {
-                LOGGER.error("transaction task thread ran abnormally: {}", e.getMessage(), e);
-            } catch (ExecutionException e) {
-                Throwable throwable = e.getCause() != null ? e.getCause() : e;
-                LOGGER.error("task execution exception: {}", throwable.getMessage(), throwable);
-            }
-        }
-    }
-
     /**
      * Timeout check.
      */
-    @Deprecated
     protected void timeoutCheck() {
-        SessionCondition sessionCondition = new SessionCondition(new GlobalStatus[] {GlobalStatus.Begin});
+        SessionCondition sessionCondition = new SessionCondition(GlobalStatus.Begin);
         sessionCondition.setLazyLoadBranch(true);
-        Collection<GlobalSession> allSessions =
+        Collection<GlobalSession> beginGlobalsessions =
             SessionHolder.getRootSessionManager().findGlobalSessions(sessionCondition);
-        timeoutCheck(allSessions);
-    }
-
-    /**
-     * Timeout check.
-     *
-     * @param beginGlobalsessions
-     */
-    protected void timeoutCheck(Collection<GlobalSession> beginGlobalsessions) {
         if (CollectionUtils.isEmpty(beginGlobalsessions)) {
             return;
         }
@@ -403,24 +354,14 @@ protected void timeoutCheck(Collection<GlobalSession> beginGlobalsessions) {
 
     }
 
+
     /**
      * Handle retry rollbacking.
      */
-    @Deprecated
     protected void handleRetryRollbacking() {
-        SessionCondition sessionCondition =
-                new SessionCondition(rollbackingStatuses.toArray(new GlobalStatus[0]));
+        SessionCondition sessionCondition = new SessionCondition(rollbackingStatuses);
         Collection<GlobalSession> rollbackingSessions =
-                SessionHolder.getRetryRollbackingSessionManager().findGlobalSessions(sessionCondition);
-        handleRetryRollbacking(rollbackingSessions);
-    }
-
-    /**
-     * Handle retry rollbacking.
-     *
-     * @param rollbackingSessions
-     */
-    protected void handleRetryRollbacking(Collection<GlobalSession> rollbackingSessions) {
+            SessionHolder.getRetryRollbackingSessionManager().findGlobalSessions(sessionCondition);
         long now = System.currentTimeMillis();
         SessionHelper.forEach(rollbackingSessions, rollbackingSession -> {
             try {
@@ -454,10 +395,11 @@ protected void handleRetryRollbacking(Collection<GlobalSession> rollbackingSessi
 
     /**
      * Handle retry committing.
-     *
-     * @param committingSessions
      */
-    protected void handleRetryCommitting(Collection<GlobalSession> committingSessions) {
+    protected void handleRetryCommitting() {
+        SessionCondition retryCommittingSessionCondition = new SessionCondition(retryCommittingStatuses);
+        Collection<GlobalSession> committingSessions =
+            SessionHolder.getRootSessionManager().findGlobalSessions(retryCommittingSessionCondition);
         if (CollectionUtils.isEmpty(committingSessions)) {
             return;
         }
@@ -495,10 +437,11 @@ private boolean isRetryTimeout(long now, long timeout, long beginTime) {
 
     /**
      * Handle async committing.
-     * 
-     * @param asyncCommittingSessions
      */
-    protected void handleAsyncCommitting(Collection<GlobalSession> asyncCommittingSessions) {
+    protected void handleAsyncCommitting() {
+        SessionCondition asyncCommittingSessionCondition = new SessionCondition(GlobalStatus.AsyncCommitting);
+        Collection<GlobalSession> asyncCommittingSessions =
+            SessionHolder.getRootSessionManager().findGlobalSessions(asyncCommittingSessionCondition);
         if (CollectionUtils.isEmpty(asyncCommittingSessions)) {
             return;
         }
@@ -542,11 +485,25 @@ protected void undoLogDelete() {
      * Init.
      */
     public void init() {
-        handleAllSession.scheduleAtFixedRate(
-            () -> SessionHolder.distributedLockAndExecute(HANDLE_ALL_SESSION, this::handleAllSession), 0,
-                HANDLE_ALL_SESSION_PERIOD, TimeUnit.MILLISECONDS);
-        undoLogDelete.scheduleAtFixedRate(() -> SessionHolder.distributedLockAndExecute(UNDOLOG_DELETE, this::undoLogDelete),
-                UNDO_LOG_DELAY_DELETE_PERIOD, UNDO_LOG_DELETE_PERIOD, TimeUnit.MILLISECONDS);
+        retryRollbacking.scheduleAtFixedRate(
+            () -> SessionHolder.distributedLockAndExecute(RETRY_ROLLBACKING, this::handleRetryRollbacking), 0,
+            ROLLBACKING_RETRY_PERIOD, TimeUnit.MILLISECONDS);
+
+        retryCommitting.scheduleAtFixedRate(
+            () -> SessionHolder.distributedLockAndExecute(RETRY_COMMITTING, this::handleRetryCommitting), 0,
+            COMMITTING_RETRY_PERIOD, TimeUnit.MILLISECONDS);
+
+        asyncCommitting.scheduleAtFixedRate(
+            () -> SessionHolder.distributedLockAndExecute(ASYNC_COMMITTING, this::handleAsyncCommitting), 0,
+            ASYNC_COMMITTING_RETRY_PERIOD, TimeUnit.MILLISECONDS);
+
+        timeoutCheck.scheduleAtFixedRate(
+            () -> SessionHolder.distributedLockAndExecute(TX_TIMEOUT_CHECK, this::timeoutCheck), 0,
+            TIMEOUT_RETRY_PERIOD, TimeUnit.MILLISECONDS);
+
+        undoLogDelete.scheduleAtFixedRate(
+            () -> SessionHolder.distributedLockAndExecute(UNDOLOG_DELETE, this::undoLogDelete),
+            UNDO_LOG_DELAY_DELETE_PERIOD, UNDO_LOG_DELETE_PERIOD, TimeUnit.MILLISECONDS);
     }
 
     @Override
@@ -571,14 +528,12 @@ public void onResponse(AbstractResultMessage response, RpcContext context) {
     @Override
     public void destroy() {
         // 1. first shutdown timed task
-        handleAllSession.shutdown();
         retryRollbacking.shutdown();
         retryCommitting.shutdown();
         asyncCommitting.shutdown();
         timeoutCheck.shutdown();
         branchRemoveExecutor.shutdown();
         try {
-            handleAllSession.awaitTermination(TIMED_TASK_SHUTDOWN_MAX_WAIT_MILLS, TimeUnit.MILLISECONDS);
             retryRollbacking.awaitTermination(TIMED_TASK_SHUTDOWN_MAX_WAIT_MILLS, TimeUnit.MILLISECONDS);
             retryCommitting.awaitTermination(TIMED_TASK_SHUTDOWN_MAX_WAIT_MILLS, TimeUnit.MILLISECONDS);
             asyncCommitting.awaitTermination(TIMED_TASK_SHUTDOWN_MAX_WAIT_MILLS, TimeUnit.MILLISECONDS);

server/src/test/java/io/seata/server/coordinator/DefaultCoordinatorTest.java

@@ -158,7 +158,7 @@ public void test_handleRetryRollbacking() throws TransactionException, Interrupt
         Assertions.assertNotNull(branchId);
 
         Thread.sleep(100);
-        defaultCoordinator.handleAllSession();
+        defaultCoordinator.timeoutCheck();
         defaultCoordinator.handleRetryRollbacking();
 
         GlobalSession globalSession = SessionHolder.findGlobalSession(xid);