修改：支持jdk17，@PostConstruct 高版本与低版本不兼容

pom.xml

@@ -6,44 +6,52 @@
 
     <groupId>net.tccn</groupId>
     <artifactId>zhub-client-spring</artifactId>
-    <version>0.1.2</version>
+    <version>0.1.3.dev</version>
+
+    <parent>
+        <groupId>org.springframework.boot</groupId>
+        <artifactId>spring-boot-starter-parent</artifactId>
+        <version>3.1.10</version>
+        <relativePath/> <!-- lookup parent from repository -->
+    </parent>
 
     <properties>
-        <maven.compiler.source>1.8</maven.compiler.source>
+        <maven.compiler.source>17</maven.compiler.source>
-        <maven.compiler.target>1.8</maven.compiler.target>
+        <maven.compiler.target>17</maven.compiler.target>
     </properties>
 
     <dependencies>
         <dependency>
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot-starter-web</artifactId>
-            <version>3.2.1</version>
+            <scope>provided</scope>
-            <scope>compile</scope>
         </dependency>
 
         <dependency>
             <groupId>com.google.code.gson</groupId>
             <artifactId>gson</artifactId>
-            <version>2.10.1</version>
-            <scope>compile</scope>
         </dependency>
         <dependency>
             <groupId>junit</groupId>
             <artifactId>junit</artifactId>
-            <version>4.13.1</version>
+            <!--<version>4.13.1</version>-->
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.projectlombok</groupId>
+            <artifactId>lombok</artifactId>
+            <!--<version>1.18.30</version>-->
+            <scope>provided</scope>
+        </dependency>
     </dependencies>
 
     <repositories>
         <repository>
-            <id>maven-nexus</id>
+            <id>maven-release</id>
             <name>maven-nexus</name>
             <url>https://nexus.1216.top/repository/maven-public/</url>
         </repository>
     </repositories>
-
-
     <distributionManagement>
         <repository>
             <id>mvn-release</id>

src/main/java/tccn/Event.java

@@ -15,8 +15,8 @@ public class Event<V> {
         this.value = value;
     }
 
-    public static <V> Event<V> of(String topic, V value) {
+    public static <V> Event of(String topic, V value) {
-        return new Event<>(topic, value);
+        return new Event<V>(topic, value);
     }
 
 

src/main/java/tccn/zhub/ZHubClient.java

@@ -1,6 +1,8 @@
 package tccn.zhub;
 
 import com.google.gson.reflect.TypeToken;
+import jakarta.annotation.PostConstruct;
+import lombok.Setter;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.stereotype.Component;
 import tccn.AbstractConsumer;
@@ -9,7 +11,6 @@ import tccn.IConsumer;
 import tccn.IProducer;
 import tccn.timer.Timers;
 
-import javax.annotation.PostConstruct;
 import java.io.BufferedReader;
 import java.io.IOException;
 import java.io.InputStreamReader;
@@ -21,27 +22,29 @@ import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Map;
 import java.util.UUID;
-import java.util.concurrent.*;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.function.BiConsumer;
 import java.util.function.Consumer;
 import java.util.function.Function;
 import java.util.logging.Level;
 import java.util.logging.Logger;
 
-import static java.nio.charset.StandardCharsets.UTF_8;
-
 @Component
 public class ZHubClient extends AbstractConsumer implements IConsumer, IProducer {
 
     public Logger logger = Logger.getLogger(ZHubClient.class.getSimpleName());
+    @Setter
     @Value("${zhub.addr}")
     private String addr = "127.0.0.1:1216";
-
+    @Setter
     @Value("${zhub.groupid}")
     private String groupid = "";
-
+    @Setter
     @Value("${zhub.auth}")
     private String auth = "";
-
+    @Setter
     @Value("${zhub.appid}")
     protected String appid = "";
 
@@ -59,10 +62,19 @@ public class ZHubClient extends AbstractConsumer implements IConsumer, IProducer
     private final LinkedBlockingQueue<Event<String>> rpcCallQueue = new LinkedBlockingQueue<>(); // RPC CALL MSG
     private final LinkedBlockingQueue<String> sendMsgQueue = new LinkedBlockingQueue<>(); // SEND MSG
 
+    private final BiConsumer<Runnable, Integer> threadBuilder = (r, n) -> {
+        for (int i = 0; i < n; i++) {
+            new Thread(() -> r.run()).start();
+        }
+    };
+
     /*private static boolean isFirst = true;
     private boolean isMain = false;*/
     private static final Map<String, ZHubClient> mainHub = new HashMap<>(); // 127.0.0.1:1216 - ZHubClient
 
+    public ZHubClient() {
+    }
+
     public ZHubClient(String addr, String groupid, String appid, String auth) {
         this.addr = addr;
         this.groupid = groupid;
@@ -205,119 +217,98 @@ public class ZHubClient extends AbstractConsumer implements IConsumer, IProducer
         }).start();
 
         // 定时调度事件
-        new Thread(() -> {
+        threadBuilder.accept(() -> {
-            ExecutorService executor = Executors.newSingleThreadExecutor();
             while (true) {
                 Timer timer = null;
                 try {
-                    timer = timerQueue.take();
+                    if ((timer = timerQueue.take()) == null) {
+                        return;
+                    }
                     long start = System.currentTimeMillis();
-                    executor.submit(timer.runnable).get(5, TimeUnit.SECONDS);
+                    timer.runnable.run();
                     long end = System.currentTimeMillis();
                     logger.finest(String.format("timer [%s] : elapsed time %s ms", timer.name, end - start));
-                } catch (InterruptedException | ExecutionException | TimeoutException e) {
+                } catch (InterruptedException e) {
-                    if (e instanceof TimeoutException) {
+                    e.printStackTrace();
-                        executor = Executors.newSingleThreadExecutor();
+                } catch (Exception e) {
-                        logger.log(Level.WARNING, "TimeoutException [" + timer.name + "]", e);
+                    logger.log(Level.WARNING, "timer [" + timer.name + "]", e);
-                    } else {
-                        e.printStackTrace();
-                    }
                 }
             }
-        }).start();
+        }, 1);
 
         // topic msg
-        new Thread(() -> {
+        threadBuilder.accept(() -> {
-            ExecutorService executor = Executors.newSingleThreadExecutor();
             while (true) {
                 Event<String> event = null;
                 try {
-                    event = topicQueue.take();
+                    if ((event = topicQueue.take()) == null) {
-                    logger.log(Level.FINE, "topic[" + event.topic + "] :" + event.value);
+                        continue;
-
-                    String topic = event.topic;
-                    String value = event.value;
-                    executor.submit(() -> accept(topic, value)).get(5, TimeUnit.SECONDS);
-                } catch (InterruptedException | ExecutionException | TimeoutException e) {
-                    if (e instanceof TimeoutException) {
-                        executor = Executors.newSingleThreadExecutor();
-                        logger.log(Level.WARNING, "TimeoutException, topic[" + event.topic + "], value[" + event.value + "]", e);
-                    } else if (event != null) {
-                        logger.log(Level.WARNING, "topic[" + event.topic + "] event accept error :" + event.value, e);
                     }
+                    logger.log(Level.FINE, "topic[" + event.topic + "] :" + event.value);
+                    accept(event.topic, event.value);
+                } catch (InterruptedException e) {
+                    e.printStackTrace();
+                } catch (Exception e) {
+                    logger.log(Level.WARNING, "topic[" + event.topic + "] event accept error :" + event.value, e);
                 }
             }
-        }, "ZHub-topic-accept").start();
+        }, 1);
+
         // rpc back
-        new Thread(() -> {
+        threadBuilder.accept(() -> {
-            ExecutorService executor = Executors.newSingleThreadExecutor();
             while (true) {
                 Event<String> event = null;
                 try {
-                    event = rpcBackQueue.take();
+                    if ((event = rpcBackQueue.take()) == null) {
+                        continue;
+                    }
+                    //if (event)
                     logger.info(String.format("rpc-back:[%s]: %s", event.topic, event.value));
                     rpcAccept(event.value);
-
+                } catch (InterruptedException e) {
-                    String value = event.value;
+                    e.printStackTrace();
-                    executor.submit(() -> rpcAccept(value)).get(5, TimeUnit.SECONDS);
+                } catch (Exception e) {
-
+                    logger.log(Level.WARNING, "rpc-back[" + event.topic + "] event accept error :" + event.value, e);
-                } catch (InterruptedException | ExecutionException | TimeoutException e) {
-                    if (e instanceof TimeoutException) {
-                        executor = Executors.newSingleThreadExecutor();
-                        logger.log(Level.WARNING, "rpc-back TimeoutException, topic[" + event.topic + "], value[" + event.value + "]", e);
-                    } else if (event != null) {
-                        logger.log(Level.WARNING, "rpc-back[" + event.topic + "] event accept error :" + event.value, e);
-                    }
-                    throw new RuntimeException(e);
                 }
             }
-        }).start();
+        }, 1);
 
         // rpc call
-        new Thread(() -> {
+        threadBuilder.accept(() -> {
-            ExecutorService executor = Executors.newSingleThreadExecutor();
             while (true) {
                 Event<String> event = null;
                 try {
-                    event = rpcCallQueue.take();
+                    if ((event = rpcCallQueue.take()) == null) {
-                    logger.info(String.format("rpc-call:[%s] %s", event.topic, event.value));
+                        continue;
-
-                    String topic = event.topic;
-                    String value = event.value;
-                    executor.submit(() -> accept(topic, value)).get(5, TimeUnit.SECONDS);
-                } catch (InterruptedException | ExecutionException | TimeoutException e) {
-                    if (e instanceof TimeoutException) {
-                        executor = Executors.newSingleThreadExecutor();
-                        logger.log(Level.WARNING, "rpc-call TimeoutException, topic[" + event.topic + "], value[" + event.value + "]", e);
-                    } else if (event != null) {
-                        logger.log(Level.WARNING, "rpc-call[" + event.topic + "] event accept error :" + event.value, e);
                     }
+                    logger.info(String.format("rpc-call:[%s] %s", event.topic, event.value));
+                    accept(event.topic, event.value);
+                } catch (InterruptedException e) {
+                    e.printStackTrace();
+                } catch (Exception e) {
+                    logger.log(Level.WARNING, "rpc-call[" + event.topic + "] event accept error :" + event.value, e);
                 }
             }
-        }, "ZHub-rpc-call").start();
+        }, 1);
 
         // send msg
-        new Thread(() -> {
+        threadBuilder.accept(() -> {
             while (true) {
                 String msg = null;
                 try {
-                    msg = sendMsgQueue.take();
+                    if ((msg = sendMsgQueue.take()) == null) {
-                    writer.write(msg.getBytes(UTF_8));
+                        continue;
-                    writer.flush();
-                } catch (InterruptedException | IOException e) {
-                    logger.log(Level.WARNING, "send-msg[" + msg + "] event accept error :", e);
-
-                    try {
-                        Thread.sleep(3000);
-                        assert msg != null;
-                        writer.write(msg.getBytes(UTF_8));
-                        writer.flush();
-                    } catch (IOException | InterruptedException | NullPointerException ex) {
-                        e.printStackTrace();
                     }
+                    // logger.log(Level.FINEST, "send-msg: [" + msg + "]");
+                    writer.write(msg.getBytes());
+                    writer.flush();
+                } catch (InterruptedException e) {
+                    e.printStackTrace();
+                } catch (Exception e) {
+                    logger.log(Level.WARNING, "send-msg[" + msg + "] event accept error :", e);
                 }
             }
-        }).start();
+        }, 1);
 
     }