uncaught exception handler

dromara · Feb 17, 2024 · cecaf24 · cecaf24
1 parent 8044cde
commit cecaf24
Show file tree

Hide file tree

Showing 15 changed files with 66 additions and 41 deletions.
diff --git a/disjob-common/src/main/java/cn/ponfee/disjob/common/base/Symbol.java b/disjob-common/src/main/java/cn/ponfee/disjob/common/base/Symbol.java
@@ -62,7 +62,7 @@ public interface Str {
         /**
          * Tab symbol
          */
-        String TAB = "	";
+        String TAB = "\t";
 
         /**
          * Backslash symbol
@@ -159,7 +159,7 @@ public interface Char {
         /**
          * Tab symbol(the same as '\t')
          */
-        char TAB = '	';
+        char TAB = '\t';
 
         /**
          * Backslash symbol

diff --git a/disjob-common/src/main/java/cn/ponfee/disjob/common/concurrent/AbstractHeartbeatThread.java b/disjob-common/src/main/java/cn/ponfee/disjob/common/concurrent/AbstractHeartbeatThread.java
@@ -54,7 +54,7 @@ protected AbstractHeartbeatThread(long heartbeatPeriodMs) {
         super.setDaemon(true);
         super.setName(CaseFormat.UPPER_CAMEL.to(CaseFormat.LOWER_UNDERSCORE, getClass().getSimpleName()) + "_thread");
         super.setPriority(Thread.MAX_PRIORITY);
-        super.setUncaughtExceptionHandler(LoggedUncaughtExceptionHandler.INSTANCE);
+        super.setUncaughtExceptionHandler(new LoggedUncaughtExceptionHandler(log));
     }
 
     /**

diff --git a/disjob-common/src/main/java/cn/ponfee/disjob/common/concurrent/AsyncDelayedExecutor.java b/disjob-common/src/main/java/cn/ponfee/disjob/common/concurrent/AsyncDelayedExecutor.java
@@ -81,7 +81,7 @@ public AsyncDelayedExecutor(int maximumPoolSize,
 
         super.setName("async_delayed_executor-" + Integer.toHexString(hashCode()));
         super.setDaemon(false);
-        super.setUncaughtExceptionHandler(LoggedUncaughtExceptionHandler.INSTANCE);
+        super.setUncaughtExceptionHandler(new LoggedUncaughtExceptionHandler(LOG));
         super.start();
     }
 

diff --git a/...mmon/src/main/java/cn/ponfee/disjob/common/concurrent/LoggedUncaughtExceptionHandler.java b/...mmon/src/main/java/cn/ponfee/disjob/common/concurrent/LoggedUncaughtExceptionHandler.java
@@ -17,29 +17,28 @@
 package cn.ponfee.disjob.common.concurrent;
 
 import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 /**
  * Logged uncaught exception handler
  *
  * @author Ponfee
  */
 public final class LoggedUncaughtExceptionHandler implements Thread.UncaughtExceptionHandler {
-    private static final Logger LOG = LoggerFactory.getLogger(LoggedUncaughtExceptionHandler.class);
 
-    public static final LoggedUncaughtExceptionHandler INSTANCE = new LoggedUncaughtExceptionHandler();
+    private final Logger log;
 
-    private LoggedUncaughtExceptionHandler() {
+    public LoggedUncaughtExceptionHandler(Logger log) {
+        this.log = log;
     }
 
     @Override
     public void uncaughtException(Thread t, Throwable e) {
         if (e instanceof java.lang.ThreadDeath) {
-            LOG.warn("Uncaught exception handle, thread death: {}, {}", t.getName(), e.getMessage());
+            log.warn("Uncaught exception handle, thread death: {}, {}", t.getName(), e.getMessage());
         } else if (e instanceof InterruptedException) {
-            LOG.warn("Uncaught exception handle, thread interrupted: {}, {}", t.getName(), e.getMessage());
+            log.warn("Uncaught exception handle, thread interrupted: {}, {}", t.getName(), e.getMessage());
         } else {
-            LOG.error("Uncaught exception handle, occur error: " + t.getName(), e);
+            log.error("Uncaught exception handle, occur error: " + t.getName(), e);
         }
     }
 

diff --git a/disjob-common/src/main/java/cn/ponfee/disjob/common/concurrent/LoopThread.java b/disjob-common/src/main/java/cn/ponfee/disjob/common/concurrent/LoopThread.java
@@ -49,7 +49,7 @@ public LoopThread(String name, boolean daemon, int priority,
         super.setName(name);
         super.setDaemon(daemon);
         super.setPriority(priority);
-        super.setUncaughtExceptionHandler(LoggedUncaughtExceptionHandler.INSTANCE);
+        super.setUncaughtExceptionHandler(new LoggedUncaughtExceptionHandler(LOG));
         this.periodMs = periodMs;
         this.delayMs = delayMs;
         this.action = action;

diff --git a/disjob-common/src/main/java/cn/ponfee/disjob/common/concurrent/NamedThreadFactory.java b/disjob-common/src/main/java/cn/ponfee/disjob/common/concurrent/NamedThreadFactory.java
@@ -78,7 +78,7 @@ public static class Builder {
         private String prefix;
         private Boolean daemon;
         private Integer priority;
-        private Thread.UncaughtExceptionHandler uncaughtExceptionHandler = LoggedUncaughtExceptionHandler.INSTANCE;
+        private Thread.UncaughtExceptionHandler uncaughtExceptionHandler;
 
         private Builder() { }
 

diff --git a/disjob-common/src/main/java/cn/ponfee/disjob/common/concurrent/Threads.java b/disjob-common/src/main/java/cn/ponfee/disjob/common/concurrent/Threads.java
@@ -41,7 +41,7 @@ public static Thread newThread(Runnable run) {
         Thread thread = new Thread(run);
         String callerClassName = Thread.currentThread().getStackTrace()[2].getClassName();
         thread.setName(callerClassName.substring(callerClassName.lastIndexOf(".") + 1));
-        thread.setUncaughtExceptionHandler(LoggedUncaughtExceptionHandler.INSTANCE);
+        thread.setUncaughtExceptionHandler(new LoggedUncaughtExceptionHandler(LOG));
         return thread;
     }
 
@@ -59,7 +59,7 @@ public static Thread newThread(String name, boolean daemon, int priority, Runnab
         thread.setName(name);
         thread.setDaemon(daemon);
         thread.setPriority(priority);
-        thread.setUncaughtExceptionHandler(LoggedUncaughtExceptionHandler.INSTANCE);
+        thread.setUncaughtExceptionHandler(new LoggedUncaughtExceptionHandler(LOG));
         return thread;
     }
 

diff --git a/disjob-common/src/main/java/cn/ponfee/disjob/common/spring/JdbcTemplateWrapper.java b/disjob-common/src/main/java/cn/ponfee/disjob/common/spring/JdbcTemplateWrapper.java
@@ -38,8 +38,8 @@
 import java.sql.ResultSet;
 import java.util.LinkedList;
 import java.util.List;
-import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
 
 /**
  * Wrapped spring jdbc template.
@@ -53,7 +53,7 @@ public final class JdbcTemplateWrapper {
     public static final RowMapper<String> STRING_ROW_MAPPER = new SingleColumnRowMapper<>(String.class);
     public static final RowMapper<Long> LONG_ROW_MAPPER = new SingleColumnRowMapper<>(Long.class);
     public static final RowMapper<Integer> INTEGER_ROW_MAPPER = new SingleColumnRowMapper<>(Integer.class);
-    private static final Set<String> EXISTS_TABLE = ConcurrentHashMap.newKeySet();
+    private static final ConcurrentMap<String, Boolean> EXISTS_TABLE = new ConcurrentHashMap<>();
 
     private final JdbcTemplate jdbcTemplate;
 
@@ -137,29 +137,22 @@ public <T> T executeInTransaction(ThrowingFunction<ThrowingFunction<String, Prep
         });
     }
 
-    public void createTableIfNotExists(String tableName0, String createTableDdl) {
-        String tableName = tableName0.toLowerCase();
-        if (EXISTS_TABLE.contains(tableName)) {
-            return;
-        }
-        synchronized (tableName.intern()) {
-            if (EXISTS_TABLE.contains(tableName)) {
-                return;
-            }
+    public void createTableIfNotExists(String tableName, String createTableDdl) {
+        EXISTS_TABLE.computeIfAbsent(tableName.trim().toLowerCase(), key -> {
             try {
-                RetryTemplate.execute(() -> {
-                    if (existsTable(tableName)) {
-                        return;
+                return RetryTemplate.execute(() -> {
+                    if (existsTable(key)) {
+                        return true;
                     }
                     jdbcTemplate.execute(createTableDdl);
-                    Assert.state(existsTable(tableName), () -> "Create table " + tableName + " failed.");
-                    EXISTS_TABLE.add(tableName);
-                    LOG.info("Created table {} success.", tableName);
+                    Assert.state(existsTable(key), () -> "Create table " + key + " failed.");
+                    LOG.info("Created table {} success.", key);
+                    return true;
                 }, 3, 1000L);
             } catch (Throwable e) {
-                ExceptionUtils.rethrow(e);
+                return ExceptionUtils.rethrow(e);
             }
-        }
+        });
     }
 
     public boolean existsTable(String tableName) {

diff --git a/disjob-common/src/main/java/cn/ponfee/disjob/common/util/ClassUtils.java b/disjob-common/src/main/java/cn/ponfee/disjob/common/util/ClassUtils.java
@@ -586,7 +586,7 @@ private static String toString(Class<?>[] parameterTypes) {
 
     private static String decodeURL(URL url) {
         try {
-            return URLDecoder.decode(url.getPath(), Files.UTF_8);
+            return URLDecoder.decode(Objects.requireNonNull(url).getPath(), Files.UTF_8);
         } catch (UnsupportedEncodingException e) {
             return ExceptionUtils.rethrow(e);
         }

diff --git a/disjob-common/src/main/java/cn/ponfee/disjob/common/util/Jsons.java b/disjob-common/src/main/java/cn/ponfee/disjob/common/util/Jsons.java
@@ -377,7 +377,7 @@ public static void registerSimpleModule(ObjectMapper objectMapper) {
         //simpleModule.addSerializer(Money.class, JacksonMoney.INSTANCE.serializer());
         //simpleModule.addDeserializer(Money.class, JacksonMoney.INSTANCE.deserializer());
 
-        // 返回给端上浏览器JavaScript Number数值过大时会有问题：Number.MAX_SAFE_INTEGER = 9007199254740991
+        // 返回给端上浏览器JavaScript Number数值过大时会有问题：Number.MAX_SAFE_INTEGER = 9007199254740991，即“0x1FFFFFFFFFFFFFL”
         // 当数值大于`9007199254740991`时就有可能会丢失精度：1234567891011121314 -> 1234567891011121400
         //simpleModule.addSerializer(Long.class, ToStringSerializer.instance);
         //simpleModule.addSerializer(long.class, ToStringSerializer.instance);

diff --git a/disjob-common/src/test/java/cn/ponfee/disjob/common/dag/DAGPrinter.java b/disjob-common/src/test/java/cn/ponfee/disjob/common/dag/DAGPrinter.java
@@ -80,7 +80,7 @@ private static void drawGraph(String expr, String fileName) throws IOException {
         for (EndpointPair<DAGNode> edge : new DAGExpressionParser(expr).parse().edges()) {
             DAGNode s = edge.source(), t = edge.target();
             Node source = Factory.node(s.toString()).with(s.isStart() ? Shape.M_DIAMOND : Shape.RECTANGLE, Label.of(s.getName()));
-            Node target = Factory.node(t.toString()).with(t.isEnd() ? Shape.M_SQUARE : Shape.RECTANGLE, Label.of(t.getName()));
+            Node target = Factory.node(t.toString()).with(t.isEnd()   ? Shape.M_SQUARE  : Shape.RECTANGLE, Label.of(t.getName()));
             graph.add(source.link(target));
         }
 

diff --git a/...atch-api/src/main/java/cn/ponfee/disjob/dispatch/route/ConsistentHashExecutionRouter.java b/...atch-api/src/main/java/cn/ponfee/disjob/dispatch/route/ConsistentHashExecutionRouter.java
@@ -83,6 +83,8 @@ private ConsistentHash<Worker> getConsistentHash(List<Worker> workers) {
                 List<Worker> newWorkers = workers;
                 oldWorkers.stream().filter(e -> !newWorkers.contains(e)).forEach(router::removeNode);
                 newWorkers.stream().filter(e -> !oldWorkers.contains(e)).forEach(e -> router.addNode(e, virtualCount));
+                pair = Pair.of(workers, router);
+                cache.put(group, pair);
             }
             return pair.getRight();
         }

diff --git a/...-registry-consul/src/main/java/cn/ponfee/disjob/registry/consul/ConsulServerRegistry.java b/...-registry-consul/src/main/java/cn/ponfee/disjob/registry/consul/ConsulServerRegistry.java
@@ -212,7 +212,7 @@ private ConsulSubscriberThread(long initConsulIndex) {
             super.setDaemon(true);
             super.setPriority(Thread.MAX_PRIORITY);
             super.setName("consul_subscriber_thread");
-            super.setUncaughtExceptionHandler(LoggedUncaughtExceptionHandler.INSTANCE);
+            super.setUncaughtExceptionHandler(new LoggedUncaughtExceptionHandler(log));
         }
 
         @Override

diff --git a/disjob-supervisor/src/test/java/cn/ponfee/disjob/supervisor/util/CommonTest.java b/disjob-supervisor/src/test/java/cn/ponfee/disjob/supervisor/util/CommonTest.java
@@ -18,6 +18,8 @@
 
 import cn.ponfee.disjob.common.collect.Collects;
 import cn.ponfee.disjob.common.tuple.Tuple2;
+import cn.ponfee.disjob.common.util.ClassUtils;
+import cn.ponfee.disjob.common.util.Files;
 import cn.ponfee.disjob.common.util.Numbers;
 import cn.ponfee.disjob.core.base.Worker;
 import cn.ponfee.disjob.core.enums.JobType;
@@ -27,6 +29,8 @@
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 
+import java.net.URL;
+import java.net.URLDecoder;
 import java.text.MessageFormat;
 import java.util.Arrays;
 import java.util.Collections;
@@ -50,7 +54,7 @@ public void testPath() {
         Assertions.assertEquals("classpath*:a/**/xml/*.xml", path(list, 0));
         Assertions.assertEquals("classpath*:/**/a/xml/*.xml", path(list, 1));
 
-        list = Arrays.asList("a","b","c");
+        list = Arrays.asList("a", "b", "c");
         Assertions.assertEquals("classpath*:a/b/c/xml/*.xml", path(list, -1));
         Assertions.assertEquals("classpath*:a/b/c/**/xml/*.xml", path(list, 0));
         Assertions.assertEquals("classpath*:a/b/**/c/xml/*.xml", path(list, 1));
@@ -76,6 +80,33 @@ private static String path(List<String> list, int wildcardLastIndex) {
         return MessageFormat.format("classpath*:{0}xml/*.xml", path);
     }
 
+    @Test
+    public void testURLString() throws Exception {
+        URL url1 = new URL("https://www.oschina.net/search?scope=bbs&q=C语言");
+        Assertions.assertEquals("https://www.oschina.net/search?scope=bbs&q=C语言", url1.toString());
+
+        Assertions.assertEquals("/search?scope=bbs&q=C语言", url1.getFile());
+        Assertions.assertEquals("https", url1.getProtocol());
+        Assertions.assertEquals("www.oschina.net", url1.getHost());
+        Assertions.assertEquals(-1, url1.getPort());
+        Assertions.assertEquals("/search", url1.getPath());
+        Assertions.assertEquals("scope=bbs&q=C语言", url1.getQuery());
+        Assertions.assertNull(url1.getRef());
+        Assertions.assertNull(url1.getUserInfo());
+        Assertions.assertEquals("www.oschina.net", url1.getAuthority());
+        Assertions.assertEquals(443, url1.getDefaultPort());
+
+        URL url2 = new URL("https://www.oschina.net/search?scope=bbs&q=C%E8%AF%AD%E8%A8%80");
+        Assertions.assertEquals("https://www.oschina.net/search?scope=bbs&q=C%E8%AF%AD%E8%A8%80", url2.toString());
+        Assertions.assertEquals("https://www.oschina.net/search?scope=bbs&q=C语言", URLDecoder.decode(url2.toString(), Files.UTF_8));
+
+        System.out.println(ClassUtils.getClassFilePath(ClassUtils.class));
+        System.out.println(ClassUtils.getClassFilePath(org.apache.commons.lang3.StringUtils.class));
+        System.out.println(ClassUtils.getClasspath(ClassUtils.class));
+        System.out.println(ClassUtils.getClasspath(org.apache.commons.lang3.StringUtils.class));
+        System.out.println(ClassUtils.getClasspath());
+    }
+
     @Test
     public void testTaskParam() {
         ExecuteTaskParam param1 = createExecuteTaskParam(

diff --git a/disjob-worker/src/main/java/cn/ponfee/disjob/worker/base/WorkerThreadPool.java b/disjob-worker/src/main/java/cn/ponfee/disjob/worker/base/WorkerThreadPool.java
@@ -131,7 +131,7 @@ public WorkerThreadPool(int maximumPoolSize,
         super.setDaemon(true);
         super.setName(getClass().getSimpleName());
         super.setPriority(Thread.MAX_PRIORITY);
-        super.setUncaughtExceptionHandler(LoggedUncaughtExceptionHandler.INSTANCE);
+        super.setUncaughtExceptionHandler(new LoggedUncaughtExceptionHandler(LOG));
 
         WorkerConfigurator.setWorkerThreadPool(this);
     }
@@ -673,7 +673,7 @@ private WorkerThread(WorkerThreadPool threadPool, long keepAliveTimeSeconds) {
 
             super.setDaemon(true);
             super.setName(getClass().getSimpleName() + "-" + NAMED_SEQ.getAndIncrement());
-            super.setUncaughtExceptionHandler(LoggedUncaughtExceptionHandler.INSTANCE);
+            super.setUncaughtExceptionHandler(new LoggedUncaughtExceptionHandler(LOG));
             super.start();
         }