feat: optimize health schedule task

Author: caojiajun

lb-driver/src/main/java/com/netease/nim/lbd/ConnectionManager.java

@@ -2,11 +2,14 @@
 
 import com.netease.nim.lbd.util.AdjustCount;
 import com.netease.nim.lbd.util.AutoAdjustQueue;
+import com.netease.nim.lbd.util.CloseUtils;
 import com.netease.nim.lbd.util.NamedThreadFactory;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.sql.ResultSet;
 import java.sql.SQLException;
+import java.sql.Statement;
 import java.util.*;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.Executors;
@@ -316,8 +319,6 @@ private AtomicInteger getConnectErrorCount(SqlProxy sqlProxy) {
         return count;
     }
 
-
-
     //定时检查所有的sql-proxy是否可达
     private void checkReachable() {
         if (healthCheckStatus.compareAndSet(false, true)) {
@@ -331,7 +332,7 @@ private void checkReachable() {
                     if (!pool.isOnline()) {
                         continue;
                     }
-                    boolean reachable = checkReachable0(sqlProxy);
+                    boolean reachable = checkReachable0(sqlProxy, pool);
                     lock.lock();
                     try {
                         if (reachable && !pool.isReachable()) {
@@ -354,17 +355,44 @@ private void checkReachable() {
     }
 
     //检查sql proxy是否可达
-    private boolean checkReachable0(SqlProxy sqlProxy) {
-        RealConnection realConnection = new RealConnection(sqlProxy, LBDriverEnv.getRealDriver(), lbDriverUrl);
+    private boolean checkReachable0(SqlProxy sqlProxy, SqlProxyConnectionPool pool) {
+        RealConnection connection;
+        if (pool == null) {
+            connection = new RealConnection(sqlProxy, LBDriverEnv.getRealDriver(), lbDriverUrl);
+        } else {
+            connection = pool.checkReadableConnection;
+            if (connection == null) {
+                connection = new RealConnection(sqlProxy, LBDriverEnv.getRealDriver(), lbDriverUrl);
+                pool.checkReadableConnection = connection;
+            }
+        }
+        ResultSet rs = null;
+        Statement stmt = null;
+        boolean reachable;
+        boolean exception = false;
         try {
-            realConnection.syncCreating();
-            return true;
+            connection.syncCreating();
+            stmt = connection.getPhysicalConnection().createStatement();
+            rs = stmt.executeQuery(Constants.VALIDATION_QUERY);
+            reachable = rs.next();
         } catch (Exception e) {
+            exception = true;
+            reachable = false;
+        }
+        if (!reachable) {
             logger.warn("sql proxy = {} not reachable", sqlProxy);
-            return false;
-        } finally {
-            realConnection.close();
         }
+        CloseUtils.close(rs);
+        CloseUtils.close(stmt);
+        if (pool == null) {
+            CloseUtils.close(connection);
+        } else {
+            if (exception) {
+                CloseUtils.close(connection);
+                pool.checkReadableConnection = null;
+            }
+        }
+        return reachable;
     }
 
     //定时检查是否连接数均衡
@@ -474,21 +502,33 @@ private void checkSqlProxyListFromProvider() {
     private void removeOfflineSqlProxy() {
         try {
             Set<SqlProxy> set = new HashSet<>(poolMap.keySet());
+            Set<RealConnection> toClosedConnection = new HashSet<>();
+            Set<SqlProxy> offlinedSqlProxySet = new HashSet<>();
             for (SqlProxy sqlProxy : set) {
                 lock.lock();
                 try {
                     SqlProxyConnectionPool pool = poolMap.get(sqlProxy);
                     if (pool != null && !pool.isOnline() && pool.isConnectionZero()) {
                         poolMap.remove(sqlProxy);
                         connectErrorCountMap.remove(sqlProxy);
-                        logger.info("offline sql proxy = {} removed for connection 0", sqlProxy);
+                        RealConnection removed = pool.checkReadableConnection;
+                        if (removed != null) {
+                            toClosedConnection.add(removed);
+                        }
+                        offlinedSqlProxySet.add(sqlProxy);
                     }
                 } catch (Exception e) {
                     logger.error("remove offline sql proxy error, sql proxy = {}", sqlProxy, e);
                 } finally {
                     lock.unlock();
                 }
             }
+            for (SqlProxy sqlProxy : offlinedSqlProxySet) {
+                logger.info("offline sql proxy = {} removed for connection 0", sqlProxy);
+            }
+            for (RealConnection realConnection : toClosedConnection) {
+                CloseUtils.close(realConnection);
+            }
         } catch (Exception e) {
             logger.error("remove offline sql proxy error", e);
         }
@@ -500,7 +540,7 @@ private void addSqlProxy(List<SqlProxy> list) {
                 return;
             }
             for (SqlProxy sqlProxy : list) {
-                boolean reachable = checkReachable0(sqlProxy);
+                boolean reachable = checkReachable0(sqlProxy, null);
                 lock.lock();
                 try {
                     SqlProxyConnectionPool pool = poolMap.get(sqlProxy);
@@ -613,6 +653,9 @@ private class SqlProxyConnectionPool {
         private final AtomicLong closeCount = new AtomicLong(0);
         private final AtomicLong reuseCount = new AtomicLong(0);
 
+        //用于检测健康检查的链接
+        private RealConnection checkReadableConnection;
+
         //空闲的连接
         private final Deque<RealConnection> idleConnections = new ArrayDeque<>();
         //使用中的连接

lb-driver/src/main/java/com/netease/nim/lbd/Constants.java

@@ -14,6 +14,8 @@ public class Constants {
 
     public static final String MYSQL_DRIVER_URL_PREFIX = "jdbc:mysql://";
 
+    public static final String VALIDATION_QUERY = "/* ping */ SELECT 1";
+
     public static final UnsupportedMethodBehavior UNSUPPORTED_METHOD_BEHAVIOR = UnsupportedMethodBehavior.ThrowException;
     public static final boolean LOG_STATS = false;
     public static final int CHECK_BALANCE_INTERVAL_SECONDS = 10;

lb-driver/src/main/java/com/netease/nim/lbd/util/CloseUtils.java

@@ -0,0 +1,50 @@
+package com.netease.nim.lbd.util;
+
+import com.netease.nim.lbd.RealConnection;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.sql.ResultSet;
+import java.sql.Statement;
+
+/**
+ * Created by caojiajun on 2026/1/9
+ */
+public class CloseUtils {
+
+    private static final Logger logger = LoggerFactory.getLogger(CloseUtils.class);
+
+    public static void close(Statement statement) {
+        if (statement == null) {
+            return;
+        }
+        try {
+            statement.close();
+        } catch (Exception e) {
+            logger.debug("close statement error", e);
+        }
+    }
+
+    public static void close(ResultSet resultSet) {
+        if (resultSet == null) {
+            return;
+        }
+        try {
+            resultSet.close();
+        } catch (Exception e) {
+            logger.debug("close result set error", e);
+        }
+    }
+
+    public static void close(RealConnection connection) {
+        if (connection == null) {
+            return;
+        }
+        try {
+            connection.close();
+        } catch (Exception e) {
+            logger.debug("close real connection error", e);
+        }
+    }
+
+}