pom.xml

@@ -6,7 +6,7 @@
 
     <groupId>net.tccn</groupId>
     <artifactId>zhub-client-spring</artifactId>
-    <version>17.0.0409.dev</version>
+    <version>0.1.3.dev</version>
 
     <parent>
         <groupId>org.springframework.boot</groupId>

src/main/java/tccn/AbstractConsumer.java

@@ -8,7 +8,6 @@ import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Map;
 import java.util.Set;
-import java.util.concurrent.ConcurrentHashMap;
 import java.util.function.Consumer;
 
 /**
@@ -19,7 +18,7 @@ public abstract class AbstractConsumer implements IConsumer {
 
     public Gson gson = Rpc.gson;
 
-    protected Map<String, EventType<?>> eventMap = new ConcurrentHashMap<>();
+    private Map<String, EventType> eventMap = new HashMap<>();
 
     protected abstract String getGroupid();
 
@@ -37,7 +36,6 @@ public abstract class AbstractConsumer implements IConsumer {
         return set;
     }
 
-    // topic 消息消费前处理
     protected void accept(String topic, String value) {
         EventType eventType = eventMap.get(topic);
 
@@ -51,19 +49,6 @@ public abstract class AbstractConsumer implements IConsumer {
         eventType.accept(data);
     }
 
-    // rpc 被调用端
-    protected <T> void rpcAccept(String topic, T value) {
-        EventType eventType = eventMap.get(topic);
-
-        /*// eventType 与 T 的类型比较、
-        if (!eventType.typeToken.getType().getTypeName().equals(value.getClass().getTypeName())) {
-            eventType.accept(toStr(value));
-        } else {
-            eventType.accept(value);
-        }*/
-        eventType.accept(value);
-    }
-
     protected final void removeEventType(String topic) {
         eventMap.remove(topic);
     }
@@ -91,13 +76,4 @@ public abstract class AbstractConsumer implements IConsumer {
         }
     }
 
-    protected String toStr(Object v) {
-        if (v instanceof String) {
-            return (String) v;
-        } else if (v == null) {
-            return null;
-        }
-        return gson.toJson(v);
-    }
-
 }

src/main/java/tccn/Event.java

@@ -15,8 +15,9 @@ 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/IType.java

@@ -7,19 +7,13 @@ import java.util.Map;
 
 public interface IType {
 
-    TypeToken<String> STRING = new TypeToken<>() {
+    TypeToken<String> STRING = new TypeToken<String>() {
     };
 
-    TypeToken<Short> SHORT = new TypeToken<>() {
+    TypeToken<Integer> INT = new TypeToken<Integer>() {
-    };
-    TypeToken<Integer> INT = new TypeToken<>() {
-    };
-    TypeToken<Long> LONG = new TypeToken<>() {
-    };
-    TypeToken<Double> DOUBLE = new TypeToken<>() {
     };
 
-    TypeToken<Map<String, String>> MAP = new TypeToken<>() {
+    TypeToken<Map<String, String>> MAP = new TypeToken<Map<String, String>>() {
     };
 
     TypeToken<List<Map<String, String>>> LMAP = new TypeToken<List<Map<String, String>>>() {

src/main/java/tccn/timer/TimerExecutor.java

@@ -24,7 +24,7 @@ public class TimerExecutor {
         for (Task t : task) {
             t.setTimerExecutor(this);
             queue.push(t);
-            // logger.finest("add new task : " + t.getName());
+            logger.finest("add new task : " + t.getName());
         }
     }
 

src/main/java/tccn/timer/TimerTask.java

@@ -93,10 +93,10 @@ public class TimerTask implements Task {
         if (!isComplete) {
             int count = execCount.incrementAndGet(); // 执行次数+1
 
-            // long start = System.currentTimeMillis();
+            long start = System.currentTimeMillis();
             job.execute(this);
-            // long end = System.currentTimeMillis();
+            long end = System.currentTimeMillis();
-            // logger.finest(String.format("task [%s] : not complete -> %s, time: %s ms, exec count: %s.", getName(), isComplete ? "had complete" : "not complete", end - start, count));
+            logger.finest(String.format("task [%s] : not complete -> %s, time: %s ms, exec count: %s.", getName(), isComplete ? "had complete" : "not complete", end - start, count));
 
             if (!isComplete) {
                 timerExecutor.add(this, true);

src/main/java/tccn/zhub/Rpc.java

@@ -2,14 +2,7 @@ package tccn.zhub;
 
 import com.google.gson.Gson;
 import com.google.gson.annotations.Expose;
-import com.google.gson.reflect.TypeToken;
-import lombok.Getter;
-import lombok.Setter;
 
-import java.util.UUID;
-
-@Getter
-@Setter
 public class Rpc<T> {
     /*public final static Gson gson = new GsonBuilder()
             .excludeFieldsWithoutExposeAnnotation()
@@ -23,18 +16,48 @@ public class Rpc<T> {
     @Expose(deserialize = false, serialize = false)
     private RpcResult rpcResult;
 
-    @Expose(deserialize = false, serialize = false)
+    public Rpc() {
-    private TypeToken typeToken;
+    }
 
-    /*public Rpc() {
+    public Rpc(String appname, String ruk, String topic, Object value) {
-    }*/
+        this.ruk = appname + "::" + ruk;
-
-    protected Rpc(String appname, String topic, T value) {
-        this.ruk = appname + "::" + UUID.randomUUID().toString().replaceAll("-", "");
         this.topic = topic;
+        this.value = (T) gson.toJson(value);
+    }
+
+    public String getRuk() {
+        return ruk;
+    }
+
+    public void setRuk(String ruk) {
+        this.ruk = ruk;
+    }
+
+    public String getTopic() {
+        return topic;
+    }
+
+    public void setTopic(String topic) {
+        this.topic = topic;
+    }
+
+    public T getValue() {
+        return value;
+    }
+
+    public void setValue(T value) {
         this.value = value;
     }
 
+
+    public RpcResult getRpcResult() {
+        return rpcResult;
+    }
+
+    public void setRpcResult(RpcResult rpcResult) {
+        this.rpcResult = rpcResult;
+    }
+
     public String getBackTopic() {
         return ruk.split("::")[0];
     }

src/main/java/tccn/zhub/RpcResult.java

@@ -34,7 +34,7 @@ public class RpcResult<R> {
         return result;
     }
 
-    public void setResult(Object result) {
+    public void setResult(R result) {
-        this.result = (R) result;
+        this.result = result;
     }
 }

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

@@ -5,7 +5,10 @@ import jakarta.annotation.PostConstruct;
 import lombok.Setter;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.stereotype.Component;
-import tccn.*;
+import tccn.AbstractConsumer;
+import tccn.Event;
+import tccn.IConsumer;
+import tccn.IProducer;
 import tccn.timer.Timers;
 
 import java.io.BufferedReader;
@@ -15,8 +18,14 @@ import java.io.OutputStream;
 import java.net.InetSocketAddress;
 import java.net.Socket;
 import java.net.SocketException;
-import java.util.*;
+import java.util.HashMap;
-import java.util.concurrent.*;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.UUID;
+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;
@@ -44,20 +53,26 @@ public class ZHubClient extends AbstractConsumer implements IConsumer, IProducer
         init(null);
     }
 
-    private Socket client;
     private OutputStream writer;
     private BufferedReader reader;
 
     private final LinkedBlockingQueue<Timer> timerQueue = new LinkedBlockingQueue<>();
     private final LinkedBlockingQueue<Event<String>> topicQueue = new LinkedBlockingQueue<>();
-    private final LinkedBlockingQueue<Event<Object>> rpcBackQueue = new LinkedBlockingQueue<>(); // RPC BACK MSG [=> Object]
+    private final LinkedBlockingQueue<Event<String>> rpcBackQueue = new LinkedBlockingQueue<>(); // RPC BACK MSG
-    private final LinkedBlockingQueue<Event<Object>> rpcCallQueue = new LinkedBlockingQueue<>(); // RPC CALL MSG [=> Object]
+    private final LinkedBlockingQueue<Event<String>> rpcCallQueue = new LinkedBlockingQueue<>(); // RPC CALL MSG
     private final LinkedBlockingQueue<String> sendMsgQueue = new LinkedBlockingQueue<>(); // SEND MSG
 
-    private static Map<String, ZHubClient> mainHub = new HashMap<>(); // 127.0.0.1:1216 - ZHubClient
+    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) {
@@ -81,226 +96,220 @@ public class ZHubClient extends AbstractConsumer implements IConsumer, IProducer
         }
 
         // 设置第一个启动的 实例为主实例
+        /*if (isFirst) {
+            isMain = true;
+            isFirst = false;
+        }*/
         if (!mainHub.containsKey(addr)) { // 确保同步执行此 init 逻辑
             mainHub.put(addr, this);
         }
 
-        CompletableFuture.runAsync(() -> {
+        if (!initSocket(0)) {
-            if (!initSocket(0)) {
+            return;
-                return;
+        }
+        // 消息 事件接收
+        new Thread(() -> {
+            while (true) {
+                try {
+                    String readLine = reader.readLine();
+                    if (readLine == null && initSocket(Integer.MAX_VALUE)) { // 连接中断 处理
+                        continue;
+                    }
+
+                    String type;
+
+                    // +ping
+                    if ("+ping".equals(readLine)) {
+                        send("+pong");
+                        continue;
+                    }
+
+                    // 主题订阅消息
+                    if ("*3".equals(readLine)) {
+                        reader.readLine(); // $7 len()
+
+                        type = reader.readLine(); // message
+                        if (!"message".equals(type)) {
+                            continue;
+                        }
+                        reader.readLine(); //$n len(key)
+                        String topic = reader.readLine(); // topic
+
+                        String lenStr = reader.readLine();//$n len(value)
+                        int clen = 0;
+                        if (lenStr.startsWith("$")) {
+                            clen = Integer.parseInt(lenStr.replace("$", ""));
+                        }
+
+                        String value = "";
+                        do {
+                            if (value.length() > 0) {
+                                value += "\r\n";
+                            }
+                            String s = reader.readLine();
+                            value += s; // value
+                        } while (clen > 0 && clen > strLength(value));
+
+
+                        // lock msg
+                        if ("lock".equals(topic)) {
+                            Lock lock = lockTag.get(value);
+                            if (lock != null) {
+                                synchronized (lock) {
+                                    lock.success = true;
+                                    lock.notifyAll();
+                                }
+                            }
+                            continue;
+                        }
+                        // trylock msg
+                        if ("trylock".equals(topic)) {
+                            Lock lock = lockTag.get(value);
+                            if (lock != null) {
+                                synchronized (lock) {
+                                    lock.success = false;
+                                    lock.notifyAll();
+                                }
+                            }
+                            continue;
+                        }
+
+                        // rpc back msg
+                        if (appid.equals(topic)) {
+                            rpcBackQueue.add(Event.of(topic, value));
+                            continue;
+                        }
+
+                        // rpc call msg
+                        if (rpcTopics.contains(topic)) {
+                            rpcCallQueue.add(Event.of(topic, value));
+                            continue;
+                        }
+
+                        // oth msg
+                        topicQueue.add(Event.of(topic, value));
+                        continue;
+                    }
+
+                    // timer 消息
+                    if ("*2".equals(readLine)) {
+                        reader.readLine(); // $7 len()
+
+                        type = reader.readLine(); // message
+                        if (!"timer".equals(type)) {
+                            continue;
+                        }
+                        reader.readLine(); //$n len(key)
+                        String topic = reader.readLine(); // name
+
+                        timerQueue.add(timerMap.get(topic));
+                        continue;
+                    }
+
+                    logger.finest(readLine);
+                } catch (IOException e) {
+                    if (e instanceof SocketException) {
+                        initSocket(Integer.MAX_VALUE);
+                    }
+                    e.printStackTrace();
+                }
             }
-            // 消息 事件接收
+        }).start();
-            new Thread(() -> {
-                while (true) {
-                    try {
-                        String readLine = reader.readLine();
-                        if (readLine == null && initSocket(Integer.MAX_VALUE)) { // 连接中断 处理
-                            continue;
-                        }
 
-                        String type = "";
+        // 定时调度事件
-
+        threadBuilder.accept(() -> {
-                        // +ping
+            while (true) {
-                        if ("+ping".equals(readLine)) {
+                Timer timer = null;
-                            send("+pong");
+                try {
-                            continue;
+                    if ((timer = timerQueue.take()) == null) {
-                        }
+                        return;
-
-                        // 主题订阅消息
-                        if ("*3".equals(readLine)) {
-                            readLine = reader.readLine(); // $7 len()
-                            type = reader.readLine(); // message
-                            if (!"message".equals(type)) {
-                                continue;
-                            }
-                            reader.readLine(); //$n len(key)
-                            String topic = reader.readLine(); // topic
-
-                            String lenStr = reader.readLine();//$n len(value)
-                            int clen = 0;
-                            if (lenStr.startsWith("$")) {
-                                clen = Integer.parseInt(lenStr.replace("$", ""));
-                            }
-
-                            String value = "";
-                            do {
-                                if (!value.isEmpty()) {
-                                    value += "\r\n";
-                                }
-                                String s = reader.readLine();
-                                value += s; // value
-                            } while (clen > 0 && clen > strLength(value));
-
-                            logger.finest("topic[" + topic + "]: " + value);
-
-                            // lock msg
-                            if ("lock".equals(topic)) {
-                                Lock lock = lockTag.get(value);
-                                if (lock != null) {
-                                    synchronized (lock) {
-                                        lock.success = true;
-                                        lock.notifyAll();
-                                    }
-                                }
-                                continue;
-                            }
-                            // trylock msg
-                            if ("trylock".equals(topic)) {
-                                Lock lock = lockTag.get(value);
-                                if (lock != null) {
-                                    synchronized (lock) {
-                                        lock.success = false;
-                                        lock.notifyAll();
-                                    }
-                                }
-                                continue;
-                            }
-
-                            // rpc back msg
-                            if (appid.equals(topic)) {
-                                rpcBackQueue.add(Event.of(topic, value));
-                                continue;
-                            }
-
-                            // rpc call msg
-                            if (rpcTopics.contains(topic)) {
-                                rpcCallQueue.add(Event.of(topic, value));
-                                continue;
-                            }
-
-                            // oth msg
-                            topicQueue.add(Event.of(topic, value));
-                            continue;
-                        }
-
-                        // timer 消息
-                        if ("*2".equals(readLine)) {
-                            readLine = reader.readLine(); // $7 len()
-                            type = reader.readLine(); // message
-                            if (!"timer".equals(type)) {
-                                continue;
-                            }
-                            reader.readLine(); //$n len(key)
-                            String topic = reader.readLine(); // name
-
-                            logger.finest("timer[" + topic + "]: ");
-                            timerQueue.add(timerMap.get(topic));
-                            continue;
-                        }
-
-                        logger.finest(readLine);
-                    } catch (IOException e) {
-                        if (e instanceof SocketException) {
-                            initSocket(Integer.MAX_VALUE);
-                        }
-                        e.printStackTrace();
                     }
+                    long start = System.currentTimeMillis();
+                    timer.runnable.run();
+                    long end = System.currentTimeMillis();
+                    logger.finest(String.format("timer [%s] : elapsed time %s ms", timer.name, end - start));
+                } catch (InterruptedException e) {
+                    e.printStackTrace();
+                } catch (Exception e) {
+                    logger.log(Level.WARNING, "timer [" + timer.name + "]", e);
                 }
-            }).start();
+            }
-        }).thenAcceptAsync(x -> {
+        }, 1);
-            // 定时调度事件，已加入耗时监控
-            new Thread(() -> {
-                ExecutorService pool = Executors.newFixedThreadPool(1);
-                while (true) {
-                    Timer timer = null;
-                    try {
-                        timer = timerQueue.take();
 
-                        long start = System.currentTimeMillis();
+        // topic msg
-                        pool.submit(timer.runnable).get(5, TimeUnit.SECONDS);
+        threadBuilder.accept(() -> {
-                        long end = System.currentTimeMillis();
+            while (true) {
-                        logger.finest(String.format("timer [%s] : elapsed time %s ms", timer.name, end - start));
+                Event<String> event = null;
-                    } catch (InterruptedException e) {
+                try {
-                        e.printStackTrace();
+                    if ((event = topicQueue.take()) == null) {
-                    } catch (TimeoutException e) {
+                        continue;
-                        logger.log(Level.SEVERE, "timer [" + timer.name + "] time out: " + 5 + " S", e);
-                        pool = Executors.newFixedThreadPool(1);
-                    } catch (Exception e) {
-                        logger.log(Level.WARNING, "timer [" + timer.name + "]", 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);
                 }
-            }).start();
+            }
+        }, 1);
 
-            // topic msg，已加入耗时监控
+        // rpc back
-            new Thread(() -> {
+        threadBuilder.accept(() -> {
-                ExecutorService pool = Executors.newFixedThreadPool(1);
+            while (true) {
-                while (true) {
+                Event<String> event = null;
-                    Event<String> event = null;
+                try {
-                    try {
+                    if ((event = rpcBackQueue.take()) == null) {
-                        event = topicQueue.take();
+                        continue;
-
-                        String topic = event.topic;
-                        String value = event.value;
-                        pool.submit(() -> accept(topic, value)).get(5, TimeUnit.SECONDS);
-                    } catch (InterruptedException e) {
-                        e.printStackTrace();
-                    } catch (TimeoutException e) {
-                        logger.log(Level.SEVERE, "topic[" + event.topic + "] event deal time out: " + 5 + " S, value: " + toStr(event.value), e);
-                        pool = Executors.newFixedThreadPool(1);
-                    } catch (Exception e) {
-                        logger.log(Level.WARNING, "topic[" + event.topic + "] event accept error :" + toStr(event.value), e);
                     }
+                    //if (event)
+                    logger.info(String.format("rpc-back:[%s]: %s", event.topic, event.value));
+                    rpcAccept(event.value);
+                } catch (InterruptedException e) {
+                    e.printStackTrace();
+                } catch (Exception e) {
+                    logger.log(Level.WARNING, "rpc-back[" + event.topic + "] event accept error :" + event.value, e);
                 }
-            }).start();
+            }
+        }, 1);
 
-            // rpc back ,仅做数据解析，暂无耗时监控
+        // rpc call
-            new Thread(() -> {
+        threadBuilder.accept(() -> {
-                while (true) {
+            while (true) {
-                    Event<Object> event = null;
+                Event<String> event = null;
-                    try {
+                try {
-                        event = rpcBackQueue.take();
+                    if ((event = rpcCallQueue.take()) == null) {
-                        //if (event)
+                        continue;
-                        logger.finest(String.format("rpc-back:[%s]: %s", event.topic, toStr(event.value)));
-                        rpcAccept(event.value);
-                    } catch (InterruptedException e) {
-                        e.printStackTrace();
-                    } catch (Exception e) {
-                        logger.log(Level.WARNING, "rpc-back[" + event.topic + "] event accept error :" + toStr(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);
                 }
-            }).start();
+            }
+        }, 1);
 
-            // rpc call，已加入耗时监控
+        // send msg
-            new Thread(() -> {
+        threadBuilder.accept(() -> {
-                ExecutorService pool = Executors.newFixedThreadPool(1);
+            while (true) {
-                while (true) {
+                String msg = null;
-                    Event<Object> event = null;
+                try {
-                    try {
+                    if ((msg = sendMsgQueue.take()) == null) {
-                        event = rpcCallQueue.take();
+                        continue;
-
-                        logger.finest(String.format("rpc-call:[%s] %s", event.topic, toStr(event.value)));
-                        String topic = event.topic;
-                        Object value = event.value;
-                        pool.submit(() -> rpcAccept(topic, value)).get(5, TimeUnit.SECONDS);
-                    } catch (InterruptedException e) {
-                        e.printStackTrace();
-                    } catch (TimeoutException e) {
-                        logger.log(Level.SEVERE, "topic[" + event.topic + "] event deal time out: " + 5 + " S, value: " + toStr(event.value), e);
-                        pool = Executors.newFixedThreadPool(1);
-                    } catch (Exception e) {
-                        logger.log(Level.WARNING, "rpc-call[" + event.topic + "] event accept error :" + toStr(event.value), e);
                     }
+                    // 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);
 
-            // send msg
-            new Thread(() -> {
-                while (true) {
-                    String msg = null;
-                    try {
-                        msg = sendMsgQueue.take();
-
-                        // 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();
-        });
     }
 
     // ---------------------
@@ -347,6 +356,16 @@ public class ZHubClient extends AbstractConsumer implements IConsumer, IProducer
         return str.length();
     }
 
+    private String toStr(Object v) {
+        if (v instanceof String) {
+            return (String) v;
+        } else if (v == null) {
+            return null;
+        }
+
+        return gson.toJson(v);
+    }
+
     protected boolean initSocket(int retry) {
         for (int i = 0; i <= retry; i++) {
             try {
@@ -354,7 +373,8 @@ public class ZHubClient extends AbstractConsumer implements IConsumer, IProducer
                 String host = hostPort[0];
                 int port = Integer.parseInt(hostPort[1]);
 
-                client = new Socket();
+                //private ReentrantLock lock = new ReentrantLock();
+                Socket client = new Socket();
                 client.connect(new InetSocketAddress(host, port));
                 client.setKeepAlive(true);
 
@@ -368,7 +388,9 @@ public class ZHubClient extends AbstractConsumer implements IConsumer, IProducer
                 send("auth", auth);
                 send("groupid " + groupid);
 
-                StringBuilder buf = new StringBuilder("subscribe lock trylock");
+                StringBuffer buf = new StringBuffer("subscribe lock trylock");
+                /*if (isMain) {
+                }*/
                 if (mainHub.containsValue(this)) {
                     buf.append(" ").append(appid);
                 }
@@ -380,13 +402,13 @@ public class ZHubClient extends AbstractConsumer implements IConsumer, IProducer
                 // 重连 timer 订阅
                 timerMap.forEach((name, timer) -> send("timer", name));
                 if (retry > 0) {
-                    logger.warning(String.format("ZHubClient[%s][%s] %s Succeed！", getGroupid(), i + 1, "reconnection"));
+                    logger.warning(String.format("ZHubClient[%s][%s] %s Succeed！", getGroupid(), i + 1, retry > 0 ? "reconnection" : "init"));
                 } else {
-                    logger.fine(String.format("ZHubClient[%s] %s Succeed！", getGroupid(), "init"));
+                    logger.fine(String.format("ZHubClient[%s] %s Succeed！", getGroupid(), retry > 0 ? "reconnection" : "init"));
                 }
                 return true;
             } catch (Exception e) {
-                if (i == 0) {
+                if (retry == 0 || i == 0) {
                     logger.log(Level.WARNING, String.format("ZHubClient[%s] %s Failed 初始化失败！", getGroupid(), retry == 0 ? "init" : "reconnection"), e);
                     try {
                         Thread.sleep(1000);
@@ -413,15 +435,11 @@ public class ZHubClient extends AbstractConsumer implements IConsumer, IProducer
     }
 
     public boolean publish(String topic, Object v) {
-        /*if (eventMap.containsKey(topic)) { // 本地调用
-            topicQueue.add(Event.of(topic, v));
-            return true;
-        }*/
         return send("publish", topic, toStr(v));
     }
 
     public void broadcast(String topic, Object v) {
-        send("broadcast", topic, toStr(v)); // 广播必须走远端模式
+        send("broadcast", topic, toStr(v));
     }
 
     // 发送 publish 主题消息，若多次发送的 topic + "-" + value 相同，将会做延时重置
@@ -447,16 +465,22 @@ public class ZHubClient extends AbstractConsumer implements IConsumer, IProducer
             }
         }
 
-        if ("M".equals(endchar)) {
+        switch (endchar) {
-            delay *= (1000 * 60 * 60 * 24 * 30);
+            case "M":
-        } else if ("d".equals(endchar)) {
+                delay *= (1000 * 60 * 60 * 24 * 30);
-            delay *= (1000 * 60 * 60 * 24);
+                break;
-        } else if ("H".equals(endchar)) {
+            case "d":
-            delay *= (1000 * 60 * 60);
+                delay *= (1000 * 60 * 60 * 24);
-        } else if ("m".equals(endchar)) {
+                break;
-            delay *= (1000 * 60);
+            case "H":
-        } else if ("s".equals(endchar)) {
+                delay *= (1000 * 60 * 60);
-            delay *= 1000;
+                break;
+            case "m":
+                delay *= (1000 * 60);
+                break;
+            case "s":
+                delay *= 1000;
+                break;
         }
 
         delay(topic, v, delay);
@@ -473,7 +497,7 @@ public class ZHubClient extends AbstractConsumer implements IConsumer, IProducer
     }
 
     // ================================================== lock ==================================================
-    private Map<String, Lock> lockTag = new ConcurrentHashMap<>();
+    private final Map<String, Lock> lockTag = new ConcurrentHashMap<>();
 
     /**
      * 尝试加锁，立即返回，
@@ -514,7 +538,7 @@ public class ZHubClient extends AbstractConsumer implements IConsumer, IProducer
     }
 
     // ================================================== timer ==================================================
-    private ConcurrentHashMap<String, Timer> timerMap = new ConcurrentHashMap();
+    private final ConcurrentHashMap<String, Timer> timerMap = new ConcurrentHashMap();
 
     class Timer {
         String name;
@@ -542,7 +566,6 @@ public class ZHubClient extends AbstractConsumer implements IConsumer, IProducer
         send("timer", name);
     }
 
-    @Deprecated
     public void reloadTimer() {
         send("cmd", "reload-timer");
     }
@@ -550,7 +573,7 @@ public class ZHubClient extends AbstractConsumer implements IConsumer, IProducer
     // ================================================== rpc ==================================================
     // -- 调用端 --
     private static final Map<String, Rpc> rpcMap = new ConcurrentHashMap<>();
-    // private static final Map<String, TypeToken> rpcRetType = new ConcurrentHashMap<>();
+    private static final Map<String, TypeToken> rpcRetType = new ConcurrentHashMap<>();
 
     // rpc call
     public RpcResult<Void> rpc(String topic, Object v) {
@@ -564,18 +587,14 @@ public class ZHubClient extends AbstractConsumer implements IConsumer, IProducer
 
     // rpc call
     public <T, R> RpcResult<R> rpc(String topic, T v, TypeToken<R> typeToken, long timeout) {
-        Rpc rpc = new Rpc<>(appid, topic, v);
+        Rpc rpc = new Rpc<>(appid, UUID.randomUUID().toString().replaceAll("-", ""), topic, v);
-        rpc.setTypeToken(typeToken);
-
         String ruk = rpc.getRuk();
         rpcMap.put(ruk, rpc);
+        if (typeToken != null) {
+            rpcRetType.put(ruk, typeToken);
+        }
         try {
-            if (eventMap.containsKey(topic)) { // 本地调用
+            publish(topic, rpc); // send("rpc", topic, toStr(rpc));
-                rpcCallQueue.add(Event.of(topic, rpc));
-            } else {
-                rpc.setValue(toStr(rpc.getValue()));
-                publish(topic, rpc); // send("rpc", topic, toStr(rpc));
-            }
             synchronized (rpc) {
                 if (timeout <= 0) {
                     timeout = 1000 * 15;
@@ -625,30 +644,8 @@ public class ZHubClient extends AbstractConsumer implements IConsumer, IProducer
 
     // RpcResult: {ruk:xxx-xxxx, retcode:0}
     // rpc call back consumer
-    private <T> void rpcAccept(T value) {
+    private void rpcAccept(String value) {
-        // 接收到 本地调用返回的 RpcResult
+        RpcResult resp = gson.fromJson(value, new TypeToken<RpcResult<String>>() {
-        if (value instanceof RpcResult) {
-            String ruk = ((RpcResult) value).getRuk();
-            Rpc rpc = rpcMap.remove(ruk);
-            if (rpc == null) {
-                return;
-            }
-
-            // TODO, 本地模式下返回的数据对象类型需要和处理端一致，不然会出现类型转换异常 - 解决办法，当出现不一致的情况取数据做转换
-            TypeToken typeToken = rpc.getTypeToken();
-            if (typeToken.getType() != ((RpcResult<?>) value).getResult().getClass()) {
-                Object result = gson.fromJson(toStr(((RpcResult<?>) value).getResult()), typeToken.getType());
-                ((RpcResult<?>) value).setResult(result);
-            }
-
-            rpc.setRpcResult((RpcResult) value);
-            synchronized (rpc) {
-                rpc.notify();
-            }
-            return;
-        }
-
-        RpcResult resp = gson.fromJson((String) value, new TypeToken<RpcResult<String>>() {
         }.getType());
 
         String ruk = resp.getRuk();
@@ -656,10 +653,10 @@ public class ZHubClient extends AbstractConsumer implements IConsumer, IProducer
         if (rpc == null) {
             return;
         }
-        TypeToken typeToken = rpc.getTypeToken();
+        TypeToken typeToken = rpcRetType.get(ruk);
 
         Object result = resp.getResult();
-        if (result != null && typeToken != null && !"java.lang.String".equals(typeToken.getType().getTypeName()) && !"java.lang.Void".equals(typeToken.getType().getTypeName())) {
+        if (result != null && typeToken != null && !"java.lang.String".equals(typeToken.getType().getTypeName())) {
             result = gson.fromJson((String) resp.getResult(), typeToken.getType());
         }
 
@@ -671,38 +668,22 @@ public class ZHubClient extends AbstractConsumer implements IConsumer, IProducer
     }
 
     // -- 订阅端 --
-    private Set<String> rpcTopics = new HashSet();
+    private final HashSet rpcTopics = new HashSet();
-
-    // rpc call consumer
-    public <R> void rpcSubscribe(String topic, Function<Rpc<String>, RpcResult<R>> fun) {
-        rpcSubscribe(topic, IType.STRING, fun);
-    }
 
     // rpc call consumer
     public <T, R> void rpcSubscribe(String topic, TypeToken<T> typeToken, Function<Rpc<T>, RpcResult<R>> fun) {
-        Consumer<T> consumer = v -> {
+        Consumer<String> consumer = v -> {
             Rpc<T> rpc = null;
             try {
-                if (v instanceof String) {
+                rpc = gson.fromJson(v, new TypeToken<Rpc<String>>() {
-                    rpc = gson.fromJson((String) v, new TypeToken<Rpc<String>>() {
+                }.getType());
-                    }.getType());
-                } else {
-                    rpc = (Rpc<T>) v;
-                }
 
                 // 参数转换
-                if (rpc.getValue() instanceof String && !"java.lang.String".equals(typeToken.getType().getTypeName())) {
+                T paras = gson.fromJson((String) rpc.getValue(), typeToken.getType());
-                    T paras = gson.fromJson((String) rpc.getValue(), typeToken.getType());
+                rpc.setValue(paras);
-                    rpc.setValue(paras);
-                }
-
                 RpcResult result = fun.apply(rpc);
-                if (appid.equals(rpc.getBackTopic())) {
+                result.setResult(toStr(result.getResult()));
-                    rpcBackQueue.add(Event.of(topic, result));
+                publish(rpc.getBackTopic(), result);
-                } else {
-                    result.setResult(toStr(result.getResult())); // 远程模式 结果转换
-                    publish(rpc.getBackTopic(), result);
-                }
             } catch (Exception e) {
                 logger.log(Level.WARNING, "rpc call consumer error: " + v, e);
                 publish(rpc.getBackTopic(), rpc.retError("服务调用失败！"));
@@ -711,12 +692,6 @@ public class ZHubClient extends AbstractConsumer implements IConsumer, IProducer
         };
 
         rpcTopics.add(topic);
-        subscribe(topic, typeToken, consumer);
+        subscribe(topic, consumer);
-    }
-
-    public static void main(String[] args) {
-        System.out.println(IType.INT.getType());
-        Integer a = 1;
-        System.out.println(a.getClass());
     }
 }