Use unified VtThreadFactoryBuilder

jd-opensource · Sep 15, 2023 · 0dabb7f · 0dabb7f
1 parent a205ef7
commit 0dabb7f
Show file tree

Hide file tree

Showing 5 changed files with 28 additions and 9 deletions.
diff --git a/src/main/java/com/jd/jdbc/discovery/TopologyWatcherManager.java b/src/main/java/com/jd/jdbc/discovery/TopologyWatcherManager.java
@@ -18,7 +18,6 @@
 
 package com.jd.jdbc.discovery;
 
-import com.google.common.util.concurrent.ThreadFactoryBuilder;
 import com.jd.jdbc.common.util.CollectionUtils;
 import com.jd.jdbc.context.IContext;
 import com.jd.jdbc.sqlparser.support.logging.Log;
@@ -32,7 +31,6 @@
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ScheduledThreadPoolExecutor;
-import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
@@ -57,7 +55,7 @@ public enum TopologyWatcherManager {
         cellTopologyWatcherMap = new ConcurrentHashMap<>(16);
         globalKeyspacesMap = new ConcurrentHashMap<>(16);
 
-        scheduledExecutor = new ScheduledThreadPoolExecutor(1, VtThreadFactoryBuilder.build("reload-cell-schedule"));
+        scheduledExecutor = new ScheduledThreadPoolExecutor(1, new VtThreadFactoryBuilder.DefaultThreadFactory("reload-cell-schedule", true));
         scheduledExecutor.setExecuteExistingDelayedTasksAfterShutdownPolicy(false);
         scheduledExecutor.setRemoveOnCancelPolicy(true);
     }
@@ -104,8 +102,9 @@ public void close() {
     public void resetScheduledExecutor() {
         closeScheduledExecutor();
 
-        ThreadFactory threadFactory = new ThreadFactoryBuilder().setNameFormat("reload-cell-schedule").setDaemon(true).build();
-        scheduledExecutor = new ScheduledThreadPoolExecutor(1, threadFactory);
+        scheduledExecutor = new ScheduledThreadPoolExecutor(1, new VtThreadFactoryBuilder.DefaultThreadFactory("reload-cell-schedule", true));
+        scheduledExecutor.setExecuteExistingDelayedTasksAfterShutdownPolicy(false);
+        scheduledExecutor.setRemoveOnCancelPolicy(true);
     }
 
     public void closeScheduledExecutor() {

diff --git a/src/main/java/com/jd/jdbc/pool/HikariUtil.java b/src/main/java/com/jd/jdbc/pool/HikariUtil.java
@@ -23,13 +23,13 @@
 import com.jd.jdbc.sqlparser.support.logging.LogFactory;
 import com.jd.jdbc.topo.topoproto.TopoProto;
 import com.jd.jdbc.util.KeyspaceUtil;
+import com.jd.jdbc.util.threadpool.VtThreadFactoryBuilder;
 import com.mysql.cj.jdbc.ConnectionImpl;
 import com.zaxxer.hikari.HikariConfig;
 import com.zaxxer.hikari.metrics.MetricsTrackerFactory;
 import com.zaxxer.hikari.metrics.prometheus.PrometheusHistogramMetricsTrackerFactory;
 import com.zaxxer.hikari.pool.HikariProxyConnection;
 import com.zaxxer.hikari.pool.ProxyConnection;
-import com.zaxxer.hikari.util.UtilityElf;
 import io.vitess.proto.Topodata;
 import java.sql.Connection;
 import java.sql.SQLException;
@@ -46,7 +46,7 @@ public class HikariUtil {
     private static final MetricsTrackerFactory METRICS_TRACKER_FACTORY;
 
     static {
-        ThreadFactory threadFactory = new UtilityElf.DefaultThreadFactory("housekeeper", true);
+        ThreadFactory threadFactory = new VtThreadFactoryBuilder.DefaultThreadFactory("housekeeper", true);
         HOUSEKEEPER_EXECUTOR = new ScheduledThreadPoolExecutor(1, threadFactory, new ThreadPoolExecutor.DiscardPolicy());
         HOUSEKEEPER_EXECUTOR.setExecuteExistingDelayedTasksAfterShutdownPolicy(false);
         HOUSEKEEPER_EXECUTOR.setRemoveOnCancelPolicy(true);

diff --git a/src/main/java/com/jd/jdbc/srvtopo/ResilientServer.java b/src/main/java/com/jd/jdbc/srvtopo/ResilientServer.java
@@ -18,7 +18,6 @@
 
 package com.jd.jdbc.srvtopo;
 
-import com.google.common.util.concurrent.ThreadFactoryBuilder;
 import com.jd.jdbc.context.IContext;
 import com.jd.jdbc.context.VtContext;
 import com.jd.jdbc.monitor.SrvKeyspaceCollector;
@@ -27,6 +26,7 @@
 import com.jd.jdbc.topo.TopoException;
 import com.jd.jdbc.topo.TopoExceptionCode;
 import com.jd.jdbc.topo.TopoServer;
+import com.jd.jdbc.util.threadpool.VtThreadFactoryBuilder;
 import io.vitess.proto.Topodata;
 import java.util.Objects;
 import java.util.Set;
@@ -44,7 +44,7 @@ public class ResilientServer implements SrvTopoServer {
     private static final ScheduledThreadPoolExecutor srvKeyspaceTimer;
 
     static {
-        ThreadFactory threadFactory = new ThreadFactoryBuilder().setNameFormat("srvKeyspace-Timer").build();
+        ThreadFactory threadFactory = new VtThreadFactoryBuilder.DefaultThreadFactory("srvKeyspace-Timer", true);
         srvKeyspaceTimer = new ScheduledThreadPoolExecutor(1, threadFactory, new ThreadPoolExecutor.DiscardPolicy());
         srvKeyspaceTimer.setExecuteExistingDelayedTasksAfterShutdownPolicy(false);
         srvKeyspaceTimer.setRemoveOnCancelPolicy(true);

diff --git a/src/main/java/com/jd/jdbc/util/NetUtil.java b/src/main/java/com/jd/jdbc/util/NetUtil.java
@@ -77,6 +77,8 @@ public static String sendPostQuery(String url, String json) {
             // 发送POST请求必须设置如下两行
             conn.setDoOutput(true);
             conn.setDoInput(true);
+            conn.setConnectTimeout(5);
+            conn.setReadTimeout(5);
             conn.connect();
 
             out = new PrintWriter(conn.getOutputStream());

diff --git a/src/main/java/com/jd/jdbc/util/threadpool/VtThreadFactoryBuilder.java b/src/main/java/com/jd/jdbc/util/threadpool/VtThreadFactoryBuilder.java
@@ -46,4 +46,22 @@ public static ThreadFactory build() {
     public static ThreadFactory build(final String nameFormat) {
         return new ThreadFactoryBuilder().setDaemon(true).setNameFormat(NAME_FORMAT_PREFIX + nameFormat + "%d").build();
     }
+
+    public static final class DefaultThreadFactory implements ThreadFactory {
+        private final String threadName;
+
+        private final boolean daemon;
+
+        public DefaultThreadFactory(String threadName, boolean daemon) {
+            this.threadName = NAME_FORMAT_PREFIX + threadName;
+            this.daemon = daemon;
+        }
+
+        @Override
+        public Thread newThread(Runnable r) {
+            Thread thread = new Thread(r, this.threadName);
+            thread.setDaemon(this.daemon);
+            return thread;
+        }
+    }
 }