optimize: Optimize RM/TM client reconnect

changes/en-us/2.x.md

@@ -91,6 +91,7 @@ Add changes here for all PR submitted to the 2.x branch.
 - [[#7741](https://github.com/apache/incubator-seata/pull/7741)] supports publishing image based on JDK 25
 - [[#7743](https://github.com/seata/seata/pull/7743)] upgrade Apache Tomcat dependency from 9.0.108 to 9.0.109
 - [[#7740](https://github.com/apache/incubator-seata/pull/7740)] enhance HttpClient to support h2c
+- [[#7783](https://github.com/apache/incubator-seata/pull/7783)] Optimize RM/TM client reconnect
 - [[#7807](https://github.com/apache/incubator-seata/pull/7807)] support mariadb 3.x
 - [[#7781](https://github.com/apache/incubator-seata/pull/7781)] highlight pmd-check log
 - [[#7704](https://github.com/apache/incubator-seata/pull/7704)] fix frontend security vulnerabilities

changes/zh-cn/2.x.md

@@ -90,6 +90,7 @@
 - [[#7741](https://github.com/apache/incubator-seata/pull/7741)] 支持发布基于JDK 25的镜像
 - [[#7743](https://github.com/seata/seata/pull/7743)] 将 Apache Tomcat 依赖项从 9.0.108 升级到 9.0.109
 - [[#7740](https://github.com/apache/incubator-seata/pull/7740)] 优化http工具类使之支持h2c协议
+- [[#7783](https://github.com/apache/incubator-seata/pull/7783)] 优化 RM/TM 客户端重连
 - [[#7807](https://github.com/apache/incubator-seata/pull/7807)] 支持 mariadb 3.x
 - [[#7781](https://github.com/apache/incubator-seata/pull/7781)] 高亮 pmd 检查日志信息
 - [[#7704](https://github.com/apache/incubator-seata/pull/7704)] 修复前端依赖漏洞

core/src/main/java/org/apache/seata/core/rpc/netty/AbstractNettyRemotingClient.java

@@ -67,6 +67,7 @@
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.locks.Condition;
 import java.util.concurrent.locks.ReentrantLock;
 import java.util.function.Function;
@@ -96,6 +97,10 @@ public abstract class AbstractNettyRemotingClient extends AbstractNettyRemoting
     protected final Condition mergeCondition = mergeLock.newCondition();
     protected volatile boolean isSending = false;
 
+    private final Runnable reconnectTask;
+    private AtomicBoolean timerStarted = new AtomicBoolean(false);
+    private final ReentrantLock reconnectLock = new ReentrantLock();
+
     /**
      * When sending message type is {@link MergeMessage}, will be stored to mergeMsgMap.
      */
@@ -120,29 +125,30 @@ public abstract class AbstractNettyRemotingClient extends AbstractNettyRemoting
 
     @Override
     public void init() {
-        timerExecutor.scheduleAtFixedRate(
-                () -> {
-                    try {
-                        clientChannelManager.reconnect(getTransactionServiceGroup());
-                    } catch (Exception ex) {
-                        LOGGER.warn("reconnect server failed. {}", ex.getMessage());
-                    }
-                },
-                SCHEDULE_DELAY_MILLS,
-                SCHEDULE_INTERVAL_MILLS,
-                TimeUnit.MILLISECONDS);
-        if (this.isEnableClientBatchSendRequest()) {
-            mergeSendExecutorService = new ThreadPoolExecutor(
-                    MAX_MERGE_SEND_THREAD,
-                    MAX_MERGE_SEND_THREAD,
-                    KEEP_ALIVE_TIME,
-                    TimeUnit.MILLISECONDS,
-                    new LinkedBlockingQueue<>(),
-                    new NamedThreadFactory(getThreadPrefix(), MAX_MERGE_SEND_THREAD));
-            mergeSendExecutorService.submit(new MergedSendRunnable());
+        reconnectLock.lock();
+        try {
+            if (timerStarted.compareAndSet(false, true)) {
+                timerExecutor.scheduleAtFixedRate(
+                        reconnectTask, SCHEDULE_DELAY_MILLS, SCHEDULE_INTERVAL_MILLS, TimeUnit.MILLISECONDS);
+                LOGGER.info("Reconnect timer started (role: {})", transactionRole.name());
+            }
+
+            if (this.isEnableClientBatchSendRequest()) {
+                mergeSendExecutorService = new ThreadPoolExecutor(
+                        MAX_MERGE_SEND_THREAD,
+                        MAX_MERGE_SEND_THREAD,
+                        KEEP_ALIVE_TIME,
+                        TimeUnit.MILLISECONDS,
+                        new LinkedBlockingQueue<>(),
+                        new NamedThreadFactory(getThreadPrefix(), MAX_MERGE_SEND_THREAD));
+                mergeSendExecutorService.submit(new MergedSendRunnable());
+            }
+
+            super.init();
+            clientBootstrap.start();
+        } finally {
+            reconnectLock.unlock();
         }
-        super.init();
-        clientBootstrap.start();
     }
 
     public AbstractNettyRemotingClient(
@@ -155,6 +161,19 @@ public AbstractNettyRemotingClient(
         clientBootstrap.setChannelHandlers(new ClientHandler(), new ChannelEventHandler(this));
         clientChannelManager = new NettyClientChannelManager(
                 new NettyPoolableFactory(this, clientBootstrap), getPoolKeyFunction(), nettyClientConfig);
+        this.reconnectTask = () -> {
+            if (!timerStarted.get()) {
+                return;
+            }
+            try {
+                String serviceGroup = getTransactionServiceGroup();
+                if (StringUtils.isNotBlank(serviceGroup)) {
+                    clientChannelManager.reconnect(serviceGroup);
+                }
+            } catch (Throwable t) {
+                LOGGER.error("Reconnect task failed for role: {}", transactionRole.name(), t);
+            }
+        };
     }
 
     @Override
@@ -271,11 +290,19 @@ public void destroyChannel(String serverAddress, Channel channel) {
 
     @Override
     public void destroy() {
-        clientBootstrap.shutdown();
-        if (mergeSendExecutorService != null) {
-            mergeSendExecutorService.shutdown();
+        reconnectLock.lock();
+        try {
+            if (timerStarted.compareAndSet(true, false)) {
+                LOGGER.info("Reconnect timer stopped (role: {})", transactionRole.name());
+            }
+            clientBootstrap.shutdown();
+            if (mergeSendExecutorService != null) {
+                mergeSendExecutorService.shutdown();
+            }
+            super.destroy();
+        } finally {
+            reconnectLock.unlock();
         }
-        super.destroy();
     }
 
     public void setTransactionMessageHandler(TransactionMessageHandler transactionMessageHandler) {