修改：pom 依赖为编译级别

2.新增Timers.tryDelay、Timers.delay 方法

conf/application.properties

@@ -1,13 +0,0 @@
-redkale.name=zhub-dev
-redkale.port=6560
-redkale.server[0].protocol=HTTP
-redkale.server[0].host=127.0.0.1
-redkale.server[0].port=80
-# redkale.server[0].root = root
-redkale.server[0].rest.autoload=true
-redkale.server[0].rest.path=
-redkale.server[0].services[0].autoload=true
-# zhub
-redkale.cluster.zhub[hub].addr=47.111.150.118:6066
-redkale.cluster.zhub[hub].auth=zchd@123456
-redkale.cluster.zhub[hub].groupid=venue-zhub

conf/config.properties

@@ -1,14 +0,0 @@
-# redis
-redis.host=47.111.150.118
-redis.password=*Zhong9307!
-redis.port=6064
-
-# pulsar
-pulsar.serviceurl=pulsar://47.113.228.247:6650
-
-# zdb
-# zdb.host = 127.0.0.1
-# zdb.port = 1216
-
-zhub.host = 127.0.0.1
-zhub.port = 1216

conf/logging.properties

@@ -1,18 +0,0 @@
-handlers=java.util.logging.ConsoleHandler
-# handlers = java.util.logging.FileHandler
-############################################################
-.level=FINEST
-java.level=INFO
-javax.level=INFO
-com.sun.level=INFO
-sun.level=INFO
-jdk.level=INFO
-java.util.logging.FileHandler.level=FINER
-#10M
-java.util.logging.FileHandler.limit=10M
-java.util.logging.FileHandler.count=20
-java.util.logging.FileHandler.encoding=UTF-8
-java.util.logging.FileHandler.pattern=${APP_HOME}/logs-%tY%tm/log-%tY%tm%td.log
-java.util.logging.FileHandler.unusual=${APP_HOME}/logs-%tY%tm/log-warnerr-%tY%tm%td.log
-java.util.logging.FileHandler.append=true
-java.util.logging.ConsoleHandler.level=FINEST

conf/source.properties

@@ -1,5 +0,0 @@
-############ ClusterSource   @Resource(name="hub") ############
-# redkale.cluster.zhub[hub].addr = 47.111.150.118:6066
-# redkale.cluster.zhub[hub].auth = zchd@123456
-# redkale.cluster.zhub[hub].groupid = venue-zhub
-

pom.xml

@@ -4,23 +4,46 @@
          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
     <modelVersion>4.0.0</modelVersion>
 
-    <groupId>dev.zhub</groupId>
+    <groupId>net.tccn</groupId>
-    <artifactId>zhub-client-redkale</artifactId>
+    <artifactId>zhub-client-spring</artifactId>
-    <version>x.1.1.dev</version>
+    <version>0.1.2</version>
 
     <properties>
-        <maven.compiler.source>17</maven.compiler.source>
+        <maven.compiler.source>1.8</maven.compiler.source>
-        <maven.compiler.target>17</maven.compiler.target>
+        <maven.compiler.target>1.8</maven.compiler.target>
-        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
     </properties>
 
+    <dependencies>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-web</artifactId>
+            <version>3.2.1</version>
+            <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>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+
     <repositories>
         <repository>
-            <id>maven-release</id>
+            <id>maven-nexus</id>
             <name>maven-nexus</name>
             <url>https://nexus.1216.top/repository/maven-public/</url>
         </repository>
     </repositories>
+
+
     <distributionManagement>
         <repository>
             <id>mvn-release</id>
@@ -29,18 +52,4 @@
         </repository>
     </distributionManagement>
 
-    <dependencies>
-        <dependency>
-            <groupId>org.redkale</groupId>
-            <artifactId>redkale</artifactId>
-            <version>2.8.0.dev</version>
-            <scope>compile</scope>
-        </dependency>
-        <dependency>
-            <groupId>junit</groupId>
-            <artifactId>junit</artifactId>
-            <version>4.13.1</version>
-            <scope>compile</scope>
-        </dependency>
-    </dependencies>
 </project>

src/main/java/dev/zhub/RedisTest.java

@@ -1,392 +0,0 @@
-package dev.zhub;
-
-import org.junit.Test;
-import org.redkale.net.AsyncIOGroup;
-import org.redkale.util.AnyValue;
-import org.redkale.util.ResourceFactory;
-import org.redkalex.cache.redis.MyRedisCacheSource;
-
-import static org.redkale.boot.Application.RESNAME_APP_CLIENT_ASYNCGROUP;
-import static org.redkale.source.AbstractCacheSource.*;
-
-public class RedisTest {
-
-    static MyRedisCacheSource source = new MyRedisCacheSource();
-
-    /**
-     * 3
-     */
-    @Test
-    public void keyTest() {
-        source.set("a", 3);
-        System.out.println(source.get("a"));
-        source.del("a");
-    }
-
-    /**
-     * ax:false
-     * ax:true
-     * ax:false
-     */
-    @Test
-    public void bitTest() {
-        boolean ax = source.getBit("ax", 6);
-        System.out.println("ax:"+ ax); // false
-        source.setBit("ax", 6, true);
-
-        ax = source.getBit("ax", 6);
-        System.out.println("ax:"+ ax); // true
-
-        source.setBit("ax", 6, false);
-        ax = source.getBit("ax", 6);
-        System.out.println("ax:"+ ax); // false
-        source.del("ax");
-    }
-
-    @Test
-    public void setTest() {
-        source.del("setx");
-        source.sadd("setx", int.class, 1, 2, 3, 5, 6);
-        int setx = source.spop("setx", int.class);
-        System.out.println(setx);
-
-        setx = source.spop("setx", int.class);
-        System.out.println(setx);
-        source.del("setx");
-
-
-        source.srem("setx", int.class,213, 2312);
-
-        /*//source.sadd("setx", list.toArray(Integer[]::new));
-        List<Integer> list = List.of(2, 3, 5);
-        // source.sadd("setx", list.toArray(Integer[]::new));
-        source.sadd("setx", list.toArray(Integer[]::new));
-        source.sadd("setx", 12, 2312, 213);
-        source.sadd("setx", List.of(1011, 10222));*/
-
-    }
-
-
-
-    static { // redis://:*Zhong9307!@47.111.150.118:6064?db=2
-        AnyValue.DefaultAnyValue conf = new AnyValue.DefaultAnyValue().addValue(CACHE_SOURCE_MAXCONNS, "1");
-        conf.addValue(CACHE_SOURCE_NODES, "redis://:123456@127.0.0.1:6379?db=0");
-
-        final ResourceFactory factory = ResourceFactory.create();
-        final AsyncIOGroup asyncGroup = new AsyncIOGroup(8192, 16);
-        asyncGroup.start();
-        factory.register(RESNAME_APP_CLIENT_ASYNCGROUP, asyncGroup);
-        factory.inject(source);
-        //source.defaultConvert = JsonFactory.root().getConvert();
-        source.init(conf);
-
-
-        /*
-        source.lock("lockx", 5000);
-        */
-
-
-        source.keysStartsWith("more-hot").forEach(x -> {
-            System.out.println(x);
-            source.del(x);
-
-            int i = (short) 3;
-        });
-
-        //--------------------- set ------------------------------
-        /*
-         */
-
-        /*
-
-
-
-
-
-
-        Collection<String> setx1 = source.getCollection("setx", String.class);
-
-        System.out.println(setx1);
-
-
-        //source.getexLong()
-
-        source.setHms("hmx", Map.of("a", "5", "b", "51", "c", "ads"));
-
-        List<Serializable> hmget = source.hmget("hmx", int.class, "a");
-
-        System.out.println(hmget);
-
-        Integer hm = source.getHm("hmx", int.class, "ads");
-        System.out.println(hm);
-
-        Map<String, String> hms = source.getHms("hmx", "a", "b");
-        System.out.println("hmx:" + hms);
-
-        *//*System.out.println("======================================================");
-        System.out.println(source.incrHm("hmx", "a", -6.0));
-        hms = source.getHms("hmx", "a", "b");
-        System.out.println("hmx：a+1后的结果 " + hms);*//*
-        System.out.println("======================================================");
-        source.setHm("hmx", "c", 12);
-        hms = source.getHms("hmx", "a", "b", "c", "d", "a");
-        System.out.println("hmx：设置 c=12 后的结果 " + hms);
-        System.out.println("======================================================");
-        Double c = source.getHm("hmx", double.class, "c");
-        System.out.println("hmx 中 c 值：" + c);*/
-
-        /*Map<String, Object> hmx = source.getHmall("hmx");
-        System.out.println("Hmall：" + hmx);*/
-
-
-
-
-        /*AnyValue.DefaultAnyValue conf = new AnyValue.DefaultAnyValue();
-        conf.addValue("node", new AnyValue.DefaultAnyValue().addValue("addr", "47.111.150.118").addValue("port", "6064").addValue("password", "*Zhong9307!").addValue("db", 2));
-
-        source.defaultConvert = JsonFactory.root().getConvert();
-        source.initValueType(String.class); //value用String类型
-        source.init(conf);*/
-    }
-
-    public static void main(String[] args) {
-
-
-        //source.setLong("a", 125);
-
-        /*long a = source.getLong("a", 0);
-        System.out.println(a);
-
-        List<String> keys = source.keys("farm*");
-        keys.forEach(x -> System.out.println(x));
-
-        try {
-            Thread.sleep(1000);
-        } catch (InterruptedException e) {
-            throw new RuntimeException(e);
-        }*/
-
-
-        // ===========================================
-        //System.out.println(source.remove("a", "b"));
-
-        // bit
-        /*source.initValueType(Integer.class);
-        source.remove("a");
-        boolean a = source.getBit("a", 1);
-        System.out.println(a);
-
-        source.setBit("a", 1, true);
-        a = source.getBit("a", 1);
-        System.out.println("bit-a-1: " + a);
-
-        source.setBit("a", 1, false);
-        a = source.getBit("a", 1);
-        System.out.println("bit-a-1: " + a);*/
-
-        /*source.remove("a");
-
-        // setnx
-        System.out.println(source.setnx("a", 1));
-        source.remove("a");
-        System.out.println(source.setnx("a", 1));
-
-        // set
-        source.remove("abx1");
-        source.appendSetItems("abx1", "a", "b", "c");
-        List<String> list = source.srandomItems("abx1", 2);
-        String str = source.srandomItem("abx1"); //r
-        System.out.println(list);//[r1, r2]  */
-
-        /*int[] arr = {0};
-        ExecutorService executor = Executors.newFixedThreadPool(10);
-        CountDownLatch latch = new CountDownLatch(10_0000);
-        for (int i = 0; i < 10_0000; i++) {
-            executor.submit(() -> {
-                try {
-                    source.lock("a", 50000);
-                    arr[0]++;
-                    System.out.println("Thread: " + Thread.currentThread().getName());
-                    Thread.sleep(2000);
-
-                } catch (Exception e) {
-                    e.printStackTrace();
-                } finally {
-                    source.unlock("a");
-                    latch.countDown();
-                }
-            });
-        }
-        try {
-            latch.await();
-            System.out.println("n=" + arr[0]);
-            executor.shutdown();
-        } catch (InterruptedException e) {
-            e.printStackTrace();
-        }*/
-
-
-        /*List<String> list = (List) source.getCollection("gamerank-comment-stat");
-        System.out.println(list);*/
-
-        /*for (int i = 0; i < 10; i++) {
-            String brpop = source.brpop("z", 2);
-            System.out.println(brpop);
-        }*/
-
-        // key 测试
-        /*source.set("a", "123321");
-        System.out.println(source.get("a")); // 123321
-        System.out.println(source.getTtl("a")); // -1
-        System.out.println(source.getPttl("a")); // -1
-        System.out.println(source.getPttl("x")); // -2*/
-
-        // hashmap 测试
-        /*source.remove("sk");
-        source.setHm("sk", "a", "1");
-        source.setHm("sk", "b", "2");
-        System.out.println(source.getHm("sk", "a")); // 1
-        source.remove("sk");
-
-        source.setHms("sk", Map.of("b", "5", "c", "3", "a", "1"));
-        source.hdel("sk", "a");
-
-        Map map = source.getHms("sk", "a", "x", "b", "c", "f"); //  {b=5, c=3}
-        System.out.println(map);
-        System.out.println(source.getHmall("sk")); //{b=5, c=3}
-        System.out.println(source.incrHm("sk", "b", 1.1d)); // b = 6.1
-        System.out.println(source.incrHm("sk", "c", 1)); // c = 4
-        System.out.println(source.getHmall("sk")); //{b=6.1, c=4}
-
-        System.out.println("--------------");
-        System.out.println(source.hexists("sk", "b")); // true
-        System.out.println(source.getCollectionSize("sk")); // 2*/
-
-
-        /*Map<String, String> hms = source.getHms("supportusers", "5-kfeu0f", "xxxx", "3-0kbt7u8t", "95q- ");
-        hms.forEach((k, v) -> {
-            System.out.println(k + " : " + v);
-        });*/
-
-
-        /*MyRedisCacheSource<String> source2 = new MyRedisCacheSource();
-        source2.defaultConvert = JsonFactory.root().getConvert();
-        source2.initValueType(String.class); //value用String类型
-        source2.init(conf);*/
-
-        /*Map<String, String> gcMap = source.getHmall("hot-gamecomment");
-        gcMap.forEach((k,v) -> {
-            System.out.println(k + " : " + v);
-        });*/
-
-
-        //Map<String, String> gameinfo = source.getHms("gameinfo", "22759", "22838", "10097", "22751", "22632", "22711", "22195", "15821", "10099", "16313", "11345", "10534", "22768", "22647", "22924", "18461", "15871", "17099", "22640", "22644", "10744", "10264", "18032", "22815", "13584", "10031", "22818", "22452", "22810", "10513", "10557", "15848", "11923", "15920", "22808", "20073", "22809", "15840", "12332", "15803", "10597", "22624", "17113", "19578", "22664", "22621", "20722", "16226", "10523", "12304", "10597","11923","10031");
-        //Map<String, String> gameinfo = source.getHms("gameinfo", "22759","22838","10097","22751","22632","22711","22195","15821","10099","16313","11345","10534","22768","22647","22924","18461","15871","17099","22363","22640","22644","10744","10264","18032","22815","13584","22818","22452","22810","10513","10557","15848","15920","22808","20073","22809","15840","12332","15803","10597","22624","17113","19578","22627","22664","22621","20722","16226","10523","12304");
-
-        /*gameinfo.forEach((k,v ) -> {
-            System.out.println(v);
-        });*/
-
-
-        /*source.queryKeysStartsWith("articlebean:").forEach(x -> {
-            System.out.println(x);
-            //source.remove(x);
-            //System.out.println(source.getHmall(x));
-        });*/
-
-        // list 测试
-        /*MyRedisCacheSource<Integer> sourceInt = new MyRedisCacheSource();
-        sourceInt.defaultConvert = JsonFactory.root().getConvert();
-        sourceInt.initValueType(Integer.class); //value用String类型
-        sourceInt.init(conf);
-        sourceInt.remove("list");
-        Collection<Integer> list = sourceInt.getCollection("list");
-        System.out.println(list);
-        for (int i = 1; i <= 10; i++) {
-            sourceInt.appendListItem("list", i);
-        }
-        System.out.println(sourceInt.getCollection("list")); // [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
-
-        sourceInt.appendListItems("list", 11, 12, 13);
-        System.out.println(sourceInt.getCollection("list")); // [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13]
-        System.out.println(sourceInt.getCollection("list", 0, 5)); // [1, 2, 3, 4, 5]
-        System.out.println(sourceInt.getCollectionSize("list")); // 13
-
-        List<Integer> ids = new ArrayList<>(100);
-        for (int i = 0; i < 5000; i++) {
-            ids.add(i);
-        }
-        sourceInt.remove("abx");
-        sourceInt.appendListItems("abx", ids.toArray(Integer[]::new));
-        System.out.println(sourceInt.getCollection("abx"));*/
-
-        /*System.out.println(sourceInt.getCollectionSize("recommend-user-quality"));
-        Collection<Integer> uid = sourceInt.getCollection("recommend-user-quality");
-        System.out.println(uid);*/
-
-        // zset 测试
-        /*source.initValueType(String.class); //value用Integer类型
-        source.remove("zx");
-        source.zadd("zx", Map.of("a", 1, "b", 2));
-
-        source.zadd("zx", Map.of("a", 1, "c", 5L));
-        source.zadd("zx", Map.of("x", 20, "j", 3.5));
-        source.zadd("zx", Map.of("f", System.currentTimeMillis(), "c", 5L));
-        source.zadd("zx", Map.of("a", 1, "c", 5L));
-
-        System.out.println(source.zincr("zx", "a", 1.34)); // 2.34
-        System.out.println(source.getZsetDoubleScore("zx")); // {f=1592924555704, x=20, c=5, j=3, b=2.34, a=1}
-        source.zrem("zx", "b", "c", "e", "x");
-        System.out.println(source.getZsetLongScore("zx")); // {f=1592924555704, j=3, a=2}
-
-        System.out.println("--------------");
-        System.out.println(source.getZsetLongScore("zx", "f"));
-
-        System.out.println(source.getZrevrank("zx", "f")); // 0
-        System.out.println(source.getZrank("zx", "f")); // 2
-        System.out.println(source.getZrank("zx", "Y")); // -1
-        System.out.println(source.getCollectionSize("zx")); // 3
-
-        System.out.println(source.getZset("zx"));
-        System.out.println(source.zexists("zx", "f", "x", "a"));*/
-
-        /*LocalDate date = LocalDate.of(2019, 12, 31);
-        for (int i = 0; i < 60; i++) {
-            LocalDate localDate = date.plusDays(-i);
-            String day = localDate.format(DateTimeFormatter.ISO_LOCAL_DATE);
-            System.out.println(String.format("mkdir %s; mv *%s*.zip %s", day, day, day));
-        }*/
-
-
-        /*MyRedisCacheSource<UserDetail> source = new MyRedisCacheSource();
-        source.defaultConvert = JsonFactory.root().getConvert();
-        source.initValueType(UserDetail.class);
-        source.init(conf);
-
-
-        Map<String, UserDetail> map = source.getHmall("user-detail");
-
-        Integer[] array = map.values().stream().map(x -> x.getUserid()).toArray(Integer[]::new);
-
-        System.out.println(JsonConvert.root().convertTo(array));
-
-        Map<Integer, UserDetail> hms = source.getHms("user-detail", 11746, 11988, 11504, 11987, 11745, 11503, 11748, 11506, 11747, 11989, 11505, 11508, 11507, 11509, 11980, 11740, 11982, 11981, 11984, 11742, 11500, 11983, 11741, 11502, 11744, 11986, 11985, 11501, 11743, 11999, 11757, 11515, 1, 11514, 11998, 11756, 2, 11517, 11516, 11758, 3, 11519, 4, 5, 11518, 6, 7, 11991, 8, 11990, 9, 11993, 11751, 11750, 11992, 11753, 11511, 11995, 11994, 11510, 11752, 11755, 11513, 11997, 11512, 11996, 11754, 11724, 11966, 11965, 11723, 11968, 11726, 11967, 11725, 11728, 11969, 11727, 11729, 11960, 11720, 11962, 11961, 11722, 11964, 11721);
-
-        System.out.println(hms.size());*/
-
-        /*source.getCollection("article-comment-list", 19, 1).forEach(x -> System.out.println(x));
-
-
-        while (true) {
-            System.out.println("---" + Utility.now() + "---");
-            source.getHmall("ck").forEach((k, v) -> {
-                System.out.println(k + ":" + v);
-            });
-            try {
-                Thread.sleep(60 * 1000);
-            } catch (InterruptedException e) {
-                e.printStackTrace();
-            }
-        }*/
-    }
-}

src/main/java/dev/zhub/ZhubAgentProvider.java

@@ -1,63 +0,0 @@
-package dev.zhub;
-
-import org.redkale.boot.Application;
-import org.redkale.boot.NodeServer;
-import org.redkale.cluster.CacheClusterAgent;
-import org.redkale.cluster.ClusterAgent;
-import org.redkale.service.Service;
-import org.redkale.util.ResourceEvent;
-
-import java.net.InetSocketAddress;
-import java.util.Set;
-import java.util.concurrent.CompletableFuture;
-
-public abstract class ZhubAgentProvider extends ClusterAgent {
-
-    @Override
-    public void onResourceChange(ResourceEvent[] events) {
-
-    }
-
-    @Override
-    public void register(Application application) {
-
-    }
-
-    @Override
-    public void deregister(Application application) {
-
-    }
-
-    @Override
-    public CompletableFuture<Set<InetSocketAddress>> queryHttpAddress(String protocol, String module, String resname) {
-        return null;
-    }
-
-    @Override
-    public CompletableFuture<Set<InetSocketAddress>> querySncpAddress(String protocol, String restype, String resname) {
-        return null;
-    }
-
-    @Override
-    protected CompletableFuture<Set<InetSocketAddress>> queryAddress(ClusterEntry entry) {
-        return null;
-    }
-
-    @Override
-    protected ClusterEntry register(NodeServer ns, String protocol, Service service) {
-        deregister(ns, protocol, service);
-        ClusterEntry clusterEntry = new ClusterEntry(ns, protocol, service);
-        CacheClusterAgent.AddressEntry entry = new CacheClusterAgent.AddressEntry();
-        entry.addr = clusterEntry.address;
-        entry.resname = clusterEntry.resourceName;
-        entry.nodeid = this.nodeid;
-        entry.time = System.currentTimeMillis();
-        //source.hset(clusterEntry.serviceName, clusterEntry.serviceid, CacheClusterAgent.AddressEntry.class, entry);
-        return clusterEntry;
-    }
-
-    @Override
-    protected void deregister(NodeServer ns, String protocol, Service service) {
-
-    }
-}

src/main/java/dev/zhub/ZhubProvider.java

@@ -1,21 +0,0 @@
-package dev.zhub;
-
-import dev.zhub.client.ZHubClient;
-import org.redkale.annotation.Priority;
-import org.redkale.cluster.ClusterAgent;
-import org.redkale.cluster.ClusterAgentProvider;
-import org.redkale.util.AnyValue;
-
-@Priority(1)
-public class ZhubProvider implements ClusterAgentProvider {
-
-    @Override
-    public boolean acceptsConf(AnyValue config) {
-        return new ZHubClient().acceptsConf(config);
-    }
-
-    @Override
-    public ClusterAgent createInstance() {
-        return new ZHubClient();
-    }
-}

src/main/java/dev/zhub/client/ZHubClient.java

@@ -1,757 +0,0 @@
-package dev.zhub.client;
-
-import dev.zhub.*;
-import dev.zhub.timer.Timers;
-import org.redkale.annotation.AutoLoad;
-import org.redkale.annotation.ResourceType;
-import org.redkale.service.Local;
-import org.redkale.service.Service;
-import org.redkale.util.AnyValue;
-import org.redkale.util.Comment;
-import org.redkale.util.TypeToken;
-
-import java.io.BufferedReader;
-import java.io.IOException;
-import java.io.InputStreamReader;
-import java.io.OutputStream;
-import java.net.InetSocketAddress;
-import java.net.Socket;
-import java.net.SocketException;
-import java.util.*;
-import java.util.concurrent.*;
-import java.util.function.Consumer;
-import java.util.function.Function;
-import java.util.logging.Level;
-import java.util.logging.Logger;
-
-@Local
-@AutoLoad(false)
-@ResourceType(ZHubClient.class)
-public class ZHubClient extends AbstractConsumer implements IConsumer, IProducer, Service {
-
-    public Logger logger = Logger.getLogger(ZHubClient.class.getSimpleName());
-    private String addr = "127.0.0.1:1216";
-    private String auth = "";
-    private String groupid = "";
-
-    //private ReentrantLock lock = new ReentrantLock();
-    private Socket client;
-    private OutputStream writer;
-    private BufferedReader reader;
-
-    private final LinkedBlockingQueue<Timer> timerQueue = new LinkedBlockingQueue<>();
-    private final LinkedBlockingQueue<Event<String>> topicQueue = new LinkedBlockingQueue<>(); // [=> Object]
-    private final LinkedBlockingQueue<Event<Object>> rpcBackQueue = new LinkedBlockingQueue<>(); // RPC BACK MSG [=> Object]
-    private final LinkedBlockingQueue<Event<Object>> rpcCallQueue = new LinkedBlockingQueue<>(); // RPC CALL MSG [=> Object]
-    private final LinkedBlockingQueue<String> sendMsgQueue = new LinkedBlockingQueue<>(); // SEND MSG
-
-    private static Map<String, ZHubClient> mainHub = new HashMap<>(); // 127.0.0.1:1216 - ZHubClient
-
-    public ZHubClient() {
-
-    }
-
-    public ZHubClient(String name, Map<String, String> attr) {
-        this.APP_NAME = name;
-        this.addr = attr.get("addr");
-        this.groupid = attr.get("groupid");
-        this.auth = attr.get("auth");
-
-        this.initClient(null);
-    }
-
-    @Override
-    public void init(AnyValue config) {
-        APP_NAME = application.getName();
-        /*if (!preInit()) {
-            return;
-        }*/
-
-        if (config == null) {
-            initClient(null);
-        } else {
-            Map<String, AnyValue> nodes = getNodes(config);
-            for (String rsName : nodes.keySet()) {
-                ZHubClient client = new ZHubClient().initClient(nodes.get(rsName));
-                application.getResourceFactory().register(rsName, client);
-            }
-        }
-    }
-
-    private ZHubClient initClient(AnyValue config) {
-        // 自动注入
-        if (config != null) {
-            addr = config.getValue("addr", addr);
-
-            // 合并 addr = host:port, 做历史兼容
-            int port = config.getIntValue("port", 0);
-            if (port != 0 && !addr.contains(":")) {
-                addr = addr + ":" + port;
-            }
-
-            auth = config.getOrDefault("auth", auth);
-            groupid = config.getValue("groupid", groupid);
-        }
-
-        // 设置第一个启动的 实例为主实例
-        if (!mainHub.containsKey(addr)) { // 确保同步执行此 init 逻辑
-            mainHub.put(addr, this);
-        }
-
-        CompletableFuture.runAsync(() -> {
-            if (!initSocket(0)) {
-                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)) {
-                            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 (APP_NAME.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();
-                    }
-                }
-            }).start();
-        }).thenAcceptAsync(x -> {
-            // 定时调度事件，已加入耗时监控
-            new Thread(() -> {
-                ExecutorService pool = Executors.newFixedThreadPool(1);
-                while (true) {
-                    Timer timer = null;
-                    try {
-                        timer = timerQueue.take();
-
-                        long start = System.currentTimeMillis();
-                        pool.submit(timer.runnable).get(5, TimeUnit.SECONDS);
-                        long end = System.currentTimeMillis();
-                        logger.finest(String.format("timer [%s] : elapsed time %s ms", timer.name, end - start));
-                    } catch (InterruptedException e) {
-                        e.printStackTrace();
-                    } catch (TimeoutException e) {
-                        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);
-                    }
-                }
-            }).start();
-
-            // topic msg，已加入耗时监控
-            new Thread(() -> {
-                ExecutorService pool = Executors.newFixedThreadPool(1);
-                while (true) {
-                    Event<String> event = null;
-                    try {
-                        event = topicQueue.take();
-
-                        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);
-                    }
-                }
-            }).start();
-
-            // rpc back ,仅做数据解析，暂无耗时监控
-            new Thread(() -> {
-                while (true) {
-                    Event<Object> event = null;
-                    try {
-                        event = rpcBackQueue.take();
-                        //if (event)
-                        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);
-                    }
-                }
-            }).start();
-
-            // rpc call，已加入耗时监控
-            new Thread(() -> {
-                ExecutorService pool = Executors.newFixedThreadPool(1);
-                while (true) {
-                    Event<Object> event = null;
-                    try {
-                        event = rpcCallQueue.take();
-
-                        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);
-                    }
-                }
-            }).start();
-
-            // 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();
-        });
-
-        return this;
-    }
-
-    public boolean acceptsConf(AnyValue config) {
-        if (config == null) {
-            return false;
-        }
-
-        if (!getNodes(config).isEmpty()) {
-            return true;
-        }
-        return false;
-    }
-
-    private HashMap<String, AnyValue> getNodes(AnyValue config) {
-        AnyValue[] zhubs = config.getAnyValues("zhub");
-        HashMap<String, AnyValue> confMap = new HashMap<>();
-
-        for (AnyValue zhub : zhubs) {
-            String[] names = zhub.getNames();
-            for (String name : names) {
-                confMap.put(name, zhub.getAnyValue(name));
-            }
-        }
-
-        return confMap;
-    }
-
-    // ---------------------
-    // -- 消息发送 --
-    protected boolean send(String... data) {
-        if (data.length == 1) {
-            sendMsgQueue.add(data[0] + "\r\n");
-        } else if (data.length > 1) {
-            StringBuffer buf = new StringBuffer();
-            buf.append("*" + data.length + "\r\n");
-            for (String d : data) {
-                buf.append("$" + strLength(d) + "\r\n");
-                buf.append(d + "\r\n");
-            }
-            sendMsgQueue.add(buf.toString());
-        }
-        return true;
-    }
-
-    /*protected boolean send(String... data) {
-        try {
-            lock.lock();
-            if (data.length == 1) {
-                writer.write((data[0] + "\r\n").getBytes());
-            } else if (data.length > 1) {
-                writer.write(("*" + data.length + "\r\n").getBytes());
-                for (String d : data) {
-                    writer.write(("$" + d.length() + "\r\n").getBytes());
-                    writer.write((d + "\r\n").getBytes());
-                }
-            }
-            writer.flush();
-            return true;
-        } catch (IOException e) {
-            logger.log(Level.WARNING, "", e);
-        } finally {
-            lock.unlock();
-        }
-        return false;
-    }*/
-
-    private int strLength(String str) {
-        str = str.replaceAll("[^\\x00-\\xff]", "*");
-        return str.length();
-    }
-
-    protected boolean initSocket(int retry) {
-        for (int i = 0; i <= retry; i++) {
-            try {
-                String[] hostPort = addr.split(":");
-                String host = hostPort[0];
-                int port = Integer.parseInt(hostPort[1]);
-
-                client = new Socket();
-                client.connect(new InetSocketAddress(host, port));
-                client.setKeepAlive(true);
-
-                writer = client.getOutputStream();
-                reader = new BufferedReader(new InputStreamReader(client.getInputStream()));
-
-                String groupid = getGroupid();
-                if (groupid == null || groupid.isEmpty()) {
-                    throw new RuntimeException("ZHubClient groupid can not is empty");
-                }
-                send("auth", auth);
-                send("groupid " + groupid);
-
-                StringBuilder buf = new StringBuilder("subscribe lock trylock");
-                if (mainHub.containsValue(this)) {
-                    buf.append(" " + APP_NAME);
-                }
-                for (String topic : getTopics()) {
-                    buf.append(" ").append(topic);
-                }
-                send(buf.toString());
-
-                // 重连 timer 订阅
-                timerMap.forEach((name, timer) -> {
-                    send("timer", name);
-                });
-                if (retry > 0) {
-                    logger.warning(String.format("ZHubClient[%s][%s] %s Succeed！", getGroupid(), i + 1, "reconnection"));
-                } else {
-                    logger.fine(String.format("ZHubClient[%s] %s Succeed！", getGroupid(), "init"));
-                }
-                return true;
-            } catch (Exception e) {
-                if (i == 0) {
-                    logger.log(Level.WARNING, String.format("ZHubClient[%s] %s Failed 初始化失败！", getGroupid(), retry == 0 ? "init" : "reconnection"), e);
-                    try {
-                        Thread.sleep(1000);
-                    } catch (InterruptedException interruptedException) {
-                        interruptedException.printStackTrace();
-                    }
-                } else {
-                    logger.log(Level.WARNING, String.format("ZHubClient[%s][%s] reconnection Failed！", getGroupid(), i + 1));
-                    try {
-                        Thread.sleep(1000 * 5);
-                    } catch (InterruptedException interruptedException) {
-                        interruptedException.printStackTrace();
-                    }
-                }
-            }
-        }
-        return false;
-    }
-
-    @Override
-    public void unsubscribe(String topic) {
-        send("unsubscribe " + topic);
-        super.removeEventType(topic);
-    }
-
-    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)); // 广播必须走远端模式
-    }
-
-    // 发送 publish 主题消息，若多次发送的 topic + "-" + value 相同，将会做延时重置
-    public void delay(String topic, Object v, long millis) {
-        send("delay", topic, toStr(v), String.valueOf(millis));
-    }
-
-    // 表达式支持：d+[d,H,m,s]
-    public void delay(String topic, Object v, String delayExpr) {
-        String endchar = "";
-        int delay;
-        if (delayExpr.matches("^\\d+[d,H,m,s]$")) {
-            endchar = delayExpr.substring(delayExpr.length() - 1);
-            delay = Integer.parseInt(delayExpr.substring(0, delayExpr.length() - 1));
-        } else {
-            if (!delayExpr.matches("^\\d+$")) {
-                throw new IllegalArgumentException(String.format("ScheduledCycle period config error: [%s]", delayExpr));
-            }
-
-            delay = Integer.parseInt(delayExpr);
-            if (delay <= 0L) {
-                throw new IllegalArgumentException(String.format("ScheduledCycle period config error: [%s]", delayExpr));
-            }
-        }
-
-        if ("M".equals(endchar)) {
-            delay *= (1000 * 60 * 60 * 24 * 30);
-        } else if ("d".equals(endchar)) {
-            delay *= (1000 * 60 * 60 * 24);
-        } else if ("H".equals(endchar)) {
-            delay *= (1000 * 60 * 60);
-        } else if ("m".equals(endchar)) {
-            delay *= (1000 * 60);
-        } else if ("s".equals(endchar)) {
-            delay *= 1000;
-        }
-
-        delay(topic, v, delay);
-    }
-
-    @Override
-    protected String getGroupid() {
-        return groupid;
-    }
-
-    @Override
-    protected void subscribe(String topic) {
-        send("subscribe " + topic); //新增订阅
-    }
-
-    // ================================================== lock ==================================================
-    private Map<String, Lock> lockTag = new ConcurrentHashMap<>();
-
-    /**
-     * 尝试加锁，立即返回，
-     *
-     * @param key
-     * @param duration
-     * @return Lock: lock.success 锁定是否成功标识
-     */
-    public Lock tryLock(String key, int duration) {
-        return lock("trylock", key, duration);
-    }
-
-    public Lock lock(String key, int duration) {
-        return lock("lock", key, duration);
-    }
-
-    /**
-     * @param cmd      lock|trylock
-     * @param key      加锁 key
-     * @param duration 锁定时长
-     * @return
-     */
-    private Lock lock(String cmd, String key, int duration) {
-        String uuid = UUID.randomUUID().toString().replaceAll("-", "");
-        Lock lock = new Lock(key, uuid, duration, this);
-        lockTag.put(uuid, lock);
-
-        try {
-            // c.send("lock", key, uuid, strconv.Itoa(duration))
-            send(cmd, key, uuid, String.valueOf(duration));
-            synchronized (lock) {
-                lock.wait();
-            }
-        } catch (InterruptedException e) {
-            e.printStackTrace();
-        }
-        return lock;
-    }
-
-    // ================================================== timer ==================================================
-    private ConcurrentHashMap<String, Timer> timerMap = new ConcurrentHashMap();
-
-    class Timer {
-        String name;
-        //String expr;
-        Runnable runnable;
-        //boolean single;
-
-        public String getName() {
-            return name;
-        }
-
-
-        public Runnable getRunnable() {
-            return runnable;
-        }
-
-        public Timer(String name, Runnable runnable) {
-            this.name = name;
-            this.runnable = runnable;
-        }
-    }
-
-    public void timer(String name, Runnable run) {
-        timerMap.put(name, new Timer(name, run));
-        send("timer", name);
-    }
-
-    @Deprecated
-    public void reloadTimer() {
-        send("cmd", "reload-timer");
-    }
-
-    // ================================================== rpc ==================================================
-    // -- 调用端 --
-    private static Map<String, Rpc> rpcMap = new ConcurrentHashMap<>();
-
-    @Comment("rpc call")
-    public RpcResult<Void> rpc(String topic, Object v) {
-        return rpc(topic, v, null);
-    }
-
-    @Comment("rpc call")
-    public <T, R> RpcResult<R> rpc(String topic, T v, TypeToken<R> typeToken) {
-        return rpc(topic, v, typeToken, 0);
-    }
-
-    @Comment("rpc call")
-    public <T, R> RpcResult<R> rpc(String topic, T v, TypeToken<R> typeToken, long timeout) {
-        Rpc rpc = new Rpc<>(APP_NAME, topic, v);
-        rpc.setTypeToken(typeToken);
-
-        String ruk = rpc.getRuk();
-        rpcMap.put(ruk, rpc);
-        try {
-            if (eventMap.containsKey(topic)) { // 本地调用
-                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;
-                }
-                // call timeout default: 15s
-                Timers.delay(() -> {
-                    synchronized (rpc) {
-                        Rpc rpc1 = rpcMap.get(ruk);
-                        if (rpc1 == null) {
-                            return;
-                        }
-
-                        RpcResult rpcResult = rpc.render(505, "请求超时");
-                        rpc.setRpcResult(rpcResult);
-                        logger.warning("rpc timeout: " + convert.convertTo(rpc));
-                        rpc.notify();
-                    }
-                }, timeout);
-
-                rpc.wait();
-                rpcMap.remove(ruk);
-            }
-        } catch (InterruptedException e) {
-            e.printStackTrace();
-            // call error
-            RpcResult rpcResult = rpc.render(501, "请求失败");
-            rpc.setRpcResult(rpcResult);
-        }
-        return rpc.getRpcResult();
-    }
-
-    public <T, R> CompletableFuture<RpcResult<R>> rpcAsync(String topic, T v) {
-        return CompletableFuture.supplyAsync(() -> rpc(topic, v, null));
-    }
-
-    public <T, R> CompletableFuture<RpcResult<R>> rpcAsync(String topic, T v, TypeToken<R> typeToken) {
-        return CompletableFuture.supplyAsync(() -> rpc(topic, v, typeToken));
-    }
-
-    public <T, R> CompletableFuture<RpcResult<R>> rpcAsync(String topic, T v, long timeout) {
-        return CompletableFuture.supplyAsync(() -> rpc(topic, v, null, timeout));
-    }
-
-    public <T, R> CompletableFuture<RpcResult<R>> rpcAsync(String topic, T v, TypeToken<R> typeToken, long timeout) {
-        return CompletableFuture.supplyAsync(() -> rpc(topic, v, typeToken, timeout));
-    }
-
-    // RpcResult: {ruk:xxx-xxxx, retcode:0}
-    @Comment("rpc call back consumer")
-    private <T> void rpcAccept(T value) {
-        // 接收到 本地调用返回的 RpcResult
-        if (value instanceof RpcResult) {
-            String ruk = ((RpcResult) value).getRuk();
-            Rpc rpc = rpcMap.remove(ruk);
-            if (rpc == null) {
-                return;
-            }
-
-            // 本地模式下返回的数据对象类型需要和处理端一致，不然会出现类型转换异常 - 解决办法，当出现不一致的情况取数据做转换
-            TypeToken typeToken = rpc.getTypeToken();
-            if (typeToken.getType() != ((RpcResult<?>) value).getResult().getClass()) {
-                Object result = convert.convertFrom(typeToken.getType(), toStr(((RpcResult<?>) value).getResult()));
-                ((RpcResult<Object>) value).setResult(result);
-            }
-
-            rpc.setRpcResult((RpcResult) value);
-            synchronized (rpc) {
-                rpc.notify();
-            }
-            return;
-        }
-
-        RpcResult resp = convert.convertFrom(new TypeToken<RpcResult<String>>() {
-        }.getType(), (String) value);
-
-        String ruk = resp.getRuk();
-        Rpc rpc = rpcMap.remove(ruk);
-        if (rpc == null) {
-            return;
-        }
-        TypeToken typeToken = rpc.getTypeToken();
-
-        Object result = resp.getResult();
-        if (result != null && typeToken != null && !"java.lang.String".equals(typeToken.getType().getTypeName()) && !"java.lang.Void".equals(typeToken.getType().getTypeName())) {
-            result = convert.convertFrom(typeToken.getType(), (String) resp.getResult());
-        }
-
-        resp.setResult(result);
-        rpc.setRpcResult(resp);
-        synchronized (rpc) {
-            rpc.notify();
-        }
-    }
-
-    // -- 订阅端 --
-    private Set<String> rpcTopics = new HashSet();
-
-    @Comment("rpc call consumer")
-    public <R> void rpcSubscribe(String topic, Function<Rpc<String>, RpcResult<R>> fun) {
-        rpcSubscribe(topic, IType.STRING, fun);
-    }
-
-    @Comment("rpc call consumer")
-    public <T, R> void rpcSubscribe(String topic, TypeToken<T> typeToken, Function<Rpc<T>, RpcResult<R>> fun) {
-        Consumer<T> consumer = v -> {
-            Rpc<T> rpc = null;
-            try {
-                if (v instanceof String) {
-                    rpc = convert.convertFrom(new TypeToken<Rpc<String>>() {
-                    }.getType(), (String) v);
-                } else {
-                    rpc = (Rpc<T>) v;
-                }
-
-                // 参数转换
-                if (rpc.getValue() instanceof String && !"java.lang.String".equals(typeToken.getType().getTypeName())) {
-                    T paras = convert.convertFrom(typeToken.getType(), (String) rpc.getValue());
-                    rpc.setValue(paras);
-                }
-
-                RpcResult result = fun.apply(rpc);
-                if (APP_NAME.equals(rpc.getBackTopic())) {
-                    rpcBackQueue.add(Event.of(topic, 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("服务调用失败！"));
-            }
-            // back
-        };
-
-        rpcTopics.add(topic);
-        subscribe(topic, typeToken, consumer);
-    }
-}

src/main/java/org/redkalex/cache/redis/AbstractRedisSource.java

@@ -1,234 +0,0 @@
-/*
- * Click nbfs://nbhost/SystemFileSystem/Templates/Licenses/license-default.txt to change this license
- * Click nbfs://nbhost/SystemFileSystem/Templates/Classes/Class.java to edit this template
- */
-package org.redkalex.cache.redis;
-
-import org.redkale.annotation.Resource;
-import org.redkale.convert.Convert;
-import org.redkale.convert.json.JsonConvert;
-import org.redkale.source.AbstractCacheSource;
-import org.redkale.util.*;
-
-import java.lang.reflect.Type;
-import java.nio.charset.StandardCharsets;
-import java.util.Arrays;
-import java.util.List;
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.atomic.AtomicInteger;
-import java.util.concurrent.locks.ReentrantLock;
-import java.util.function.Function;
-
-import static org.redkale.boot.Application.RESNAME_APP_EXECUTOR;
-import static org.redkale.boot.Application.RESNAME_APP_NAME;
-
-/**
- *
- * @author zhangjx
- *
- * @since 2.8.0
- */
-public abstract class AbstractRedisSource extends AbstractCacheSource {
-
-    public static final String CACHE_SOURCE_CRYPTOR = "cryptor";
-
-    protected String name;
-
-    @Resource(name = RESNAME_APP_NAME, required = false)
-    protected String appName = "";
-
-    @Resource(required = false)
-    protected ResourceFactory resourceFactory;
-
-    @Resource(required = false)
-    protected JsonConvert defaultConvert;
-
-    @Resource(name = Resource.PARENT_NAME + "_convert", required = false)
-    protected JsonConvert convert;
-
-    protected int db;
-
-    protected RedisCryptor cryptor;
-
-    protected AnyValue conf;
-
-    private ExecutorService subExecutor;
-
-    private final ReentrantLock subExecutorLock = new ReentrantLock();
-
-    @Resource(name = RESNAME_APP_EXECUTOR, required = false)
-    protected ExecutorService workExecutor;
-
-    @Override
-    public void init(AnyValue conf) {
-        this.conf = conf;
-        super.init(conf);
-        this.name = conf.getValue("name", "");
-        if (this.convert == null) {
-            this.convert = this.defaultConvert;
-        }
-        if (conf != null) {
-            String cryptStr = conf.getValue(CACHE_SOURCE_CRYPTOR, "").trim();
-            if (!cryptStr.isEmpty()) {
-                try {
-                    Class<RedisCryptor> cryptClass = (Class) getClass().getClassLoader().loadClass(cryptStr);
-                    RedkaleClassLoader.putReflectionPublicConstructors(cryptClass, cryptClass.getName());
-                    this.cryptor = cryptClass.getConstructor().newInstance();
-                } catch (ReflectiveOperationException e) {
-                    throw new RedkaleException(e);
-                }
-            }
-        }
-        if (cryptor != null) {
-            if (resourceFactory != null) {
-                resourceFactory.inject(cryptor);
-            }
-            cryptor.init(conf);
-        }
-    }
-
-    @Override
-    public void destroy(AnyValue conf) {
-        super.destroy(conf);
-        if (cryptor != null) {
-            cryptor.destroy(conf);
-        }
-    }
-
-    public boolean acceptsConf(AnyValue config) {
-        if (config == null) {
-            return false;
-        }
-        return "redis".equalsIgnoreCase(config.getValue(CACHE_SOURCE_TYPE))
-            || getClass().getName().equalsIgnoreCase(config.getValue(CACHE_SOURCE_TYPE))
-            || config.getValue(CACHE_SOURCE_NODES, config.getValue("url", "")).startsWith("redis://")
-            || config.getValue(CACHE_SOURCE_NODES, config.getValue("url", "")).startsWith("rediss://");
-    }
-
-    protected ExecutorService subExecutor() {
-        ExecutorService executor = subExecutor;
-        if (executor != null) {
-            return executor;
-        }
-        subExecutorLock.lock();
-        try {
-            if (subExecutor == null) {
-                String threadNameFormat = "CacheSource-" + resourceName() + "-SubThread-%s";
-                Function<String, ExecutorService> func = Utility.virtualExecutorFunction();
-                final AtomicInteger counter = new AtomicInteger();
-                subExecutor = func == null ? Executors.newFixedThreadPool(Utility.cpus(), r -> {
-                    Thread t = new Thread(r);
-                    t.setDaemon(true);
-                    int c = counter.incrementAndGet();
-                    t.setName(String.format(threadNameFormat, "Virtual-" + (c < 10 ? ("00" + c) : (c < 100 ? ("0" + c) : c))));
-                    return t;
-                }) : func.apply(threadNameFormat);
-            }
-            executor = subExecutor;
-        } finally {
-            subExecutorLock.unlock();
-        }
-        return executor;
-    }
-
-    protected String getNodes(AnyValue config) {
-        return config.getValue(CACHE_SOURCE_NODES, config.getValue("url", ""));
-    }
-
-    @Override
-    public void close() throws Exception {  //在 Application 关闭时调用
-        destroy(null);
-    }
-
-    @Override
-    public String resourceName() {
-        return name;
-    }
-
-    protected String decryptValue(String key, RedisCryptor cryptor, String value) {
-        return cryptor != null ? cryptor.decrypt(key, value) : value;
-    }
-
-    protected <T> T decryptValue(String key, RedisCryptor cryptor, Type type, byte[] bs) {
-        return decryptValue(key, cryptor, convert, type, bs);
-    }
-
-    protected <T> T decryptValue(String key, RedisCryptor cryptor, Convert c, Type type, byte[] bs) {
-        if (bs == null) {
-            return null;
-        }
-        if (type == byte[].class) {
-            return (T) bs;
-        }
-        if (cryptor == null && type == String.class) {
-            return (T) new String(bs, StandardCharsets.UTF_8);
-        }
-        if (cryptor == null || (type instanceof Class && (((Class) type).isPrimitive() || Number.class.isAssignableFrom((Class) type)))) {
-            return (T) (c == null ? this.convert : c).convertFrom(type, bs);
-        }
-        String deval = cryptor.decrypt(key, new String(bs, StandardCharsets.UTF_8));
-        if (type == String.class) {
-            return (T) deval;
-        }
-        return deval == null ? null : (T) (c == null ? this.convert : c).convertFrom(type, deval.getBytes(StandardCharsets.UTF_8));
-    }
-
-    protected String encryptValue(String key, RedisCryptor cryptor, String value) {
-        return cryptor != null ? cryptor.encrypt(key, value) : value;
-    }
-
-    protected <T> byte[] encryptValue(String key, RedisCryptor cryptor, Convert c, T value) {
-        return encryptValue(key, cryptor, null, c, value);
-    }
-
-    protected <T> byte[] encryptValue(String key, RedisCryptor cryptor, Type type, Convert c, T value) {
-        if (value == null) {
-            return null;
-        }
-        Type t = type == null ? value.getClass() : type;
-        if (cryptor == null && t == String.class) {
-            return value.toString().getBytes(StandardCharsets.UTF_8);
-        }
-        byte[] bs = (c == null ? this.convert : c).convertToBytes(t, value);
-        if (bs.length > 1 && t instanceof Class && !CharSequence.class.isAssignableFrom((Class) t)) {
-            if (bs[0] == '"' && bs[bs.length - 1] == '"') {
-                bs = Arrays.copyOfRange(bs, 1, bs.length - 1);
-            }
-        }
-        return encryptValue(key, cryptor, t, bs);
-    }
-
-    protected byte[] encryptValue(String key, RedisCryptor cryptor, Type type, byte[] bs) {
-        if (bs == null) {
-            return null;
-        }
-        if (cryptor == null || (type instanceof Class && (((Class) type).isPrimitive() || Number.class.isAssignableFrom((Class) type)))) {
-            return bs;
-        }
-        String enval = cryptor.encrypt(key, new String(bs, StandardCharsets.UTF_8));
-        return enval == null ? null : enval.getBytes(StandardCharsets.UTF_8);
-    }
-
-    protected <T extends Number> T decryptScore(Class<T> scoreType, Double score) {
-        if (score == null) {
-            return null;
-        }
-        if (scoreType == int.class || scoreType == Integer.class) {
-            return (T) (Number) score.intValue();
-        } else if (scoreType == long.class || scoreType == Long.class) {
-            return (T) (Number) score.longValue();
-        } else if (scoreType == float.class || scoreType == Float.class) {
-            return (T) (Number) score.floatValue();
-        } else if (scoreType == double.class || scoreType == Double.class) {
-            return (T) (Number) score;
-        } else {
-            return JsonConvert.root().convertFrom(scoreType, score.toString());
-        }
-    }
-
-    protected CompletableFuture<Integer> returnFutureSize(List<CompletableFuture<Void>> futures) {
-        return futures == null || futures.isEmpty() ? CompletableFuture.completedFuture(0) : Utility.allOfFutures(futures).thenApply(v -> futures.size());
-    }
-}

src/main/java/org/redkalex/cache/redis/MyRedisCacheSource.java

@@ -1,350 +0,0 @@
-package org.redkalex.cache.redis;
-
-import org.redkale.annotation.AutoLoad;
-import org.redkale.annotation.ResourceType;
-import org.redkale.service.Local;
-import org.redkale.source.CacheSource;
-import org.redkale.util.AnyValue;
-
-import java.io.Serializable;
-import java.nio.charset.StandardCharsets;
-import java.util.*;
-import java.util.concurrent.CompletableFuture;
-
-@Local
-@AutoLoad(false)
-@ResourceType(CacheSource.class)
-public class MyRedisCacheSource extends RedisCacheSource {
-
-    @Override
-    public void init(AnyValue conf) {
-        super.init(conf);
-    }
-/*
-//--------------------- zset ------------------------------
-
-    public int getZrank(String key, V v) {
-        byte[][] bytes = Stream.of(key, v).map(x -> String.valueOf(x).getBytes(StandardCharsets.UTF_8)).toArray(byte[][]::new);
-        Long t = (Long) send("ZRANK", CacheEntryType.OBJECT, (Type) null, key, bytes).join();
-
-        return t == null ? -1 : (int) (long) t;
-    }
-
-    public int getZrevrank(String key, V v) {
-        byte[][] bytes = Stream.of(key, v).map(x -> String.valueOf(x).getBytes(StandardCharsets.UTF_8)).toArray(byte[][]::new);
-        Long t = (Long) send("ZREVRANK", CacheEntryType.OBJECT, (Type) null, key, bytes).join();
-
-        return t == null ? -1 : (int) (long) t;
-    }
-
-    //ZRANGE/ZREVRANGE key start stop
-    public List<V> getZset(String key) {
-        byte[][] bytes = Stream.of(key, 0, -1).map(x -> String.valueOf(x).getBytes(StandardCharsets.UTF_8)).toArray(byte[][]::new);
-        List<V> vs = (List<V>) send("ZREVRANGE", CacheEntryType.OBJECT, (Type) null, key, bytes).join();
-        List<V> vs2 = new ArrayList(vs.size());
-
-        for (int i = 0; i < vs.size(); ++i) {
-            if (i % 2 == 1) {
-                vs2.add(this.convert.convertFrom(this.objValueType, String.valueOf(vs.get(i))));
-            } else {
-                vs2.add(vs.get(i));
-            }
-        }
-
-        return vs2;
-    }
-
-    public List<V> getZset(String key, int offset, int limit) {
-        byte[][] bytes = Stream.of(key, offset, offset + limit - 1).map(x -> String.valueOf(x).getBytes(StandardCharsets.UTF_8)).toArray(byte[][]::new);
-        List<V> vs = (List<V>) send("ZREVRANGE", CacheEntryType.OBJECT, (Type) null, key, bytes).join();
-        List<V> vs2 = new ArrayList(vs.size());
-
-        for (int i = 0; i < vs.size(); ++i) {
-            if (i % 2 == 1) {
-                vs2.add(this.convert.convertFrom(this.objValueType, String.valueOf(vs.get(i))));
-            } else {
-                vs2.add(vs.get(i));
-            }
-        }
-
-        return vs2;
-    }
-
-    public LinkedHashMap<V, Long> getZsetLongScore(String key) {
-        LinkedHashMap<V, Double> map = getZsetDoubleScore(key);
-        if (map.isEmpty()) {
-            return new LinkedHashMap<>();
-        }
-
-        LinkedHashMap<V, Long> map2 = new LinkedHashMap<>(map.size());
-        map.forEach((k, v) -> map2.put(k, (long) (double) v));
-        return map2;
-    }
-
-    public LinkedHashMap<V, Long> getZsetItemsLongScore(String key) {
-        byte[][] bytes = Stream.of(key, 0, -1, "WITHSCORES").map(x -> String.valueOf(x).getBytes(StandardCharsets.UTF_8)).toArray(byte[][]::new);
-        List vs = (List) send("ZRANGE", CacheEntryType.OBJECT, (Type) null, key, bytes).join();
-
-        LinkedHashMap<V, Long> map = new LinkedHashMap<>();
-        for (int i = 0; i < vs.size(); i += 2) {
-            map.put((V) vs.get(i), (long) Double.parseDouble((String) vs.get(i + 1)));
-        }
-        return map;
-    }
-
-    public Long getZsetLongScore(String key, V v) {
-        Double score = getZsetDoubleScore(key, v);
-        if (score == null) {
-            return null;
-        }
-        return (long) (double) score;
-    }
-
-    public LinkedHashMap<V, Double> getZsetDoubleScore(String key) {
-        byte[][] bytes = Stream.of(key, 0, -1, "WITHSCORES").map(x -> String.valueOf(x).getBytes(StandardCharsets.UTF_8)).toArray(byte[][]::new);
-        List vs = (List) send("ZREVRANGE", CacheEntryType.OBJECT, (Type) null, key, bytes).join();
-
-        LinkedHashMap<V, Double> map = new LinkedHashMap<>();
-        for (int i = 0; i < vs.size(); i += 2) {
-            map.put((V) vs.get(i), Double.parseDouble((String) vs.get(i + 1)));
-        }
-        return map;
-    }
-
-    public Double getZsetDoubleScore(String key, V v) {
-        byte[][] bytes = Stream.of(key, v).map(x -> String.valueOf(x).getBytes(StandardCharsets.UTF_8)).toArray(byte[][]::new);
-        Serializable zscore = send("ZSCORE", CacheEntryType.OBJECT, (Type) null, key, bytes).join();
-        if (zscore == null) {
-            return null;
-        }
-
-        return Double.parseDouble(String.valueOf(zscore));
-    }
-
-    public LinkedHashMap<V, Long> getZsetLongScore(String key, int offset, int limit) {
-        byte[][] bytes = Stream.of(key, offset, offset + limit - 1, "WITHSCORES").map(x -> String.valueOf(x).getBytes(StandardCharsets.UTF_8)).toArray(byte[][]::new);
-        List vs = (List) send("ZREVRANGE", CacheEntryType.OBJECT, (Type) null, key, bytes).join();
-
-        LinkedHashMap<V, Long> map = new LinkedHashMap<>();
-        for (int i = 0; i < vs.size(); i += 2) {
-            map.put((V) vs.get(i), (long) Double.parseDouble((String) vs.get(i + 1)));
-        }
-        return map;
-    }
-
-    public LinkedHashMap<V, Double> getZsetDoubleScore(String key, int offset, int limit) {
-        byte[][] bytes = Stream.of(key, offset, offset + limit - 1, "WITHSCORES").map(x -> String.valueOf(x).getBytes(StandardCharsets.UTF_8)).toArray(byte[][]::new);
-        List vs = (List) send("ZREVRANGE", CacheEntryType.OBJECT, (Type) null, key, bytes).join();
-
-        LinkedHashMap<V, Double> map = new LinkedHashMap<>();
-        for (int i = 0; i < vs.size(); i += 2) {
-            map.put((V) vs.get(i), Double.parseDouble(vs.get(i + 1) + ""));
-        }
-        return map;
-    }
-* */
-
-    // --------------------
-    /*
-    supper had support
-    public <N extends Number> void zadd(String key, Map<Serializable, N> kv) {
-        if (kv == null || kv.isEmpty()) {
-            return;
-        }
-        List<Serializable> args = new ArrayList();
-        kv.forEach((k, v) -> {
-            args.add(k);
-            args.add(v);
-        });
-
-        sendAsync(RedisCommand.ZADD, key, args.toArray(Serializable[]::new)).join();
-    }
-
-    public <N extends Number> double zincr(String key, Serializable number, N n) {
-        return sendAsync(RedisCommand.ZINCRBY, key, number, n).thenApply(x -> x.getDoubleValue(0d)).join();
-    }
-
-    @Override
-    public long zrem(String key, String... vs) {
-        return sendAsync(RedisCommand.ZREM, key, keysArgs(key, vs)).thenApply(x -> x.getLongValue(0L)).join();
-    }*/
-
-    /*public <T> List<T> zexists(String key, T... fields) {
-        if (fields == null || fields.length == 0) {
-            return new ArrayList<>();
-        }
-        List<String> para = new ArrayList<>();
-        para.add("" +
-                "   local key = KEYS[1];" +
-                "    local args = ARGV;" +
-                "    local result = {};" +
-                "    for i,v in ipairs(args) do" +
-                "        local inx = redis.call('ZREVRANK', key, v);" +
-                "        if(inx) then" +
-                "             table.insert(result,1,v);" +
-                "        end" +
-                "    end" +
-                "    return result;");
-        para.add("1");
-        para.add(key);
-        for (Object field : fields) {
-            para.add(String.valueOf(field));
-        }
-
-        // todo:
-        //sendAsync("EVAL", null, para.toArray(Serializable[]::new)).thenApply(x -> x.).join();
-
-        return null;
-    }*/
-
-    //--------------------- bit ------------------------------
-    public boolean getBit(String key, int offset) {
-        return sendAsync(RedisCommand.GETBIT, key, key.getBytes(StandardCharsets.UTF_8), String.valueOf(offset).getBytes(StandardCharsets.UTF_8)).thenApply(v -> v.getIntValue(0) > 0).join();
-    }
-
-    public void setBit(String key, int offset, boolean bool) {
-        sendAsync(RedisCommand.SETBIT, key, keysArgs(key, offset + "", bool ? "1" : "0")).join();
-    }
-    //--------------------- bit ------------------------------
-
-    //--------------------- lock ------------------------------
-    // 尝试加锁，成功返回0，否则返回上一锁剩余毫秒值
-    public long tryLock(String key, int millis) {
-        String[] obj = {"" +
-                "if (redis.call('EXISTS',KEYS[1]) == 0) then " +
-                "redis.call('PSETEX',KEYS[1],ARGV[1],1); " +
-                "return 0; " +
-                "else " +
-                "return redis.call('PTTL',KEYS[1]); " +
-                "end;", 1 + "", key, millis + ""
-        };
-
-        return sendAsync(RedisCommand.EVAL, null, keysArgs(null, obj)).thenApply(v -> v.getIntValue(1)).join();
-    }
-
-    // 加锁
-    public void lock(String key, int millis) {
-        long i;
-        do {
-            i = tryLock(key, millis);
-            try {
-                Thread.sleep(10);
-            } catch (InterruptedException e) {
-                e.printStackTrace();
-            }
-        } while (i > 0);
-    }
-
-    // 解锁
-    public void unlock(String key) {
-        remove(key);
-    }
-
-    //--------------------- key ------------------------------
-
-    public String get(String key) {
-        return get(key, String.class);
-    }
-
-    public void set(String key, Serializable value) {
-        sendAsync(RedisCommand.SET, key, keysArgs(key, value + "")).join();
-    }
-
-    //--------------------- set ------------------------------
-    /*public <T> void sadd(String key, Collection<T> args) {
-        saddAsync(key, args.toArray(T[]::new)).join();
-    }*/
-
-    /*public void sadd(String key, Serializable... args) {
-        String[] arr = new String[args.length];
-        for (int i = 0; i < args.length; i++) {
-            arr[i] = args[i] + "";
-        }
-        saddAsync(key, arr).join();
-    }
-
-    public void srem(String key, String... args) {
-        sremAsync(key, args).join();
-    }
-
-    public CompletableFuture<RedisCacheResult> saddAsync(String key, Serializable... args) {
-        return sendAsync(RedisCommand.SADD, key, keysArgs(key, args));
-    }
-
-    public CompletableFuture<RedisCacheResult> sremAsync(String key, String... args) {
-        return sendAsync(RedisCommand.SREM, key, keysArgs(key, args));
-    }*/
-
-    //--------------------- hm ------------------------------
-
-    /*public Long incrHm(String key, String field, int value) {
-        return sendAsync("HINCRBY", key, field, value).thenApply(x -> x.getLongValue(0l)).join();
-    }
-
-    public Double incrHm(String key, String field, double value) {
-        return sendAsync("HINCRBYFLOAT", key, field, value).thenApply(x -> x.getDoubleValue(0d)).join();
-    }*/
-
-    public void setHm(String key, String field, Serializable value) {
-        setHmsAsync(key, Map.of(field, value)).join();
-    }
-
-    public void setHms(String key, Map kv) {
-        setHmsAsync(key, kv).join();
-    }
-
-    public CompletableFuture<RedisCacheResult> setHmsAsync(String key, Map<String, Serializable> kv) {
-        List<String> args = new ArrayList();
-        kv.forEach((k, v) -> {
-            args.add(k);
-            args.add(v + "");
-        });
-
-        return sendAsync(RedisCommand.HMSET, key, keysArgs(key, args.toArray(String[]::new)));
-    }
-
-    public String getHm(String key, String field) {
-        return getHm(key, String.class, field);
-    }
-
-    public <T extends Serializable> T getHm(String key, Class<T> type, String field) {
-        List<Serializable> list = super.hmget(key, type, field);
-        if (list == null && list.isEmpty()) {
-            return null;
-        }
-        return (T) list.get(0);
-    }
-
-    public Map<String, String> getHms(String key, String... field) {
-        return getHms(key, String.class, field);
-    }
-
-    public <T extends Serializable> Map<String, T> getHms(String key, Class<T> type, String... field) {
-        List<Serializable> list = super.hmget(key, type, field);
-        if (list == null && list.isEmpty()) {
-            return null;
-        }
-        Map<String, T> map = new HashMap<>(field.length);
-
-        for (int i = 0; i < field.length; i++) {
-            if (list.get(i) == null) {
-                continue;
-            }
-            map.put(field[i], (T) list.get(i));
-        }
-        return map;
-    }
-
-    /*public Map<String, Object> getHmall(String key) {
-        List<String> list = null;  // TODO:
-        Map<String, Object> map = new HashMap<>();
-        if (list.isEmpty()) {
-            return map;
-        }
-
-        for (int i = 0; i + 1 < list.size(); i += 2) {
-            map.put((String) list.get(i), list.get(i + 1));
-        }
-        return map;
-    }*/
-}

src/main/java/org/redkalex/cache/redis/RedisCRC16.java

@@ -1,79 +0,0 @@
-/*
- *
- */
-package org.redkalex.cache.redis;
-
-import org.redkale.util.Utility;
-
-import java.util.Arrays;
-
-/**
- *
- * @author zhangjx
- */
-public class RedisCRC16 {
-
-    private static final int MAX_SLOT = 16384;
-
-    private static final int[] LOOKUP_TABLE = {0x0000, 0x1021, 0x2042, 0x3063, 0x4084, 0x50A5, 0x60C6,
-        0x70E7, 0x8108, 0x9129, 0xA14A, 0xB16B, 0xC18C, 0xD1AD, 0xE1CE, 0xF1EF, 0x1231, 0x0210, 0x3273,
-        0x2252, 0x52B5, 0x4294, 0x72F7, 0x62D6, 0x9339, 0x8318, 0xB37B, 0xA35A, 0xD3BD, 0xC39C, 0xF3FF,
-        0xE3DE, 0x2462, 0x3443, 0x0420, 0x1401, 0x64E6, 0x74C7, 0x44A4, 0x5485, 0xA56A, 0xB54B, 0x8528,
-        0x9509, 0xE5EE, 0xF5CF, 0xC5AC, 0xD58D, 0x3653, 0x2672, 0x1611, 0x0630, 0x76D7, 0x66F6, 0x5695,
-        0x46B4, 0xB75B, 0xA77A, 0x9719, 0x8738, 0xF7DF, 0xE7FE, 0xD79D, 0xC7BC, 0x48C4, 0x58E5, 0x6886,
-        0x78A7, 0x0840, 0x1861, 0x2802, 0x3823, 0xC9CC, 0xD9ED, 0xE98E, 0xF9AF, 0x8948, 0x9969, 0xA90A,
-        0xB92B, 0x5AF5, 0x4AD4, 0x7AB7, 0x6A96, 0x1A71, 0x0A50, 0x3A33, 0x2A12, 0xDBFD, 0xCBDC, 0xFBBF,
-        0xEB9E, 0x9B79, 0x8B58, 0xBB3B, 0xAB1A, 0x6CA6, 0x7C87, 0x4CE4, 0x5CC5, 0x2C22, 0x3C03, 0x0C60,
-        0x1C41, 0xEDAE, 0xFD8F, 0xCDEC, 0xDDCD, 0xAD2A, 0xBD0B, 0x8D68, 0x9D49, 0x7E97, 0x6EB6, 0x5ED5,
-        0x4EF4, 0x3E13, 0x2E32, 0x1E51, 0x0E70, 0xFF9F, 0xEFBE, 0xDFDD, 0xCFFC, 0xBF1B, 0xAF3A, 0x9F59,
-        0x8F78, 0x9188, 0x81A9, 0xB1CA, 0xA1EB, 0xD10C, 0xC12D, 0xF14E, 0xE16F, 0x1080, 0x00A1, 0x30C2,
-        0x20E3, 0x5004, 0x4025, 0x7046, 0x6067, 0x83B9, 0x9398, 0xA3FB, 0xB3DA, 0xC33D, 0xD31C, 0xE37F,
-        0xF35E, 0x02B1, 0x1290, 0x22F3, 0x32D2, 0x4235, 0x5214, 0x6277, 0x7256, 0xB5EA, 0xA5CB, 0x95A8,
-        0x8589, 0xF56E, 0xE54F, 0xD52C, 0xC50D, 0x34E2, 0x24C3, 0x14A0, 0x0481, 0x7466, 0x6447, 0x5424,
-        0x4405, 0xA7DB, 0xB7FA, 0x8799, 0x97B8, 0xE75F, 0xF77E, 0xC71D, 0xD73C, 0x26D3, 0x36F2, 0x0691,
-        0x16B0, 0x6657, 0x7676, 0x4615, 0x5634, 0xD94C, 0xC96D, 0xF90E, 0xE92F, 0x99C8, 0x89E9, 0xB98A,
-        0xA9AB, 0x5844, 0x4865, 0x7806, 0x6827, 0x18C0, 0x08E1, 0x3882, 0x28A3, 0xCB7D, 0xDB5C, 0xEB3F,
-        0xFB1E, 0x8BF9, 0x9BD8, 0xABBB, 0xBB9A, 0x4A75, 0x5A54, 0x6A37, 0x7A16, 0x0AF1, 0x1AD0, 0x2AB3,
-        0x3A92, 0xFD2E, 0xED0F, 0xDD6C, 0xCD4D, 0xBDAA, 0xAD8B, 0x9DE8, 0x8DC9, 0x7C26, 0x6C07, 0x5C64,
-        0x4C45, 0x3CA2, 0x2C83, 0x1CE0, 0x0CC1, 0xEF1F, 0xFF3E, 0xCF5D, 0xDF7C, 0xAF9B, 0xBFBA, 0x8FD9,
-        0x9FF8, 0x6E17, 0x7E36, 0x4E55, 0x5E74, 0x2E93, 0x3EB2, 0x0ED1, 0x1EF0};
-
-    private RedisCRC16() {
-    }
-
-    public static int crc16(byte[] bytes) {
-        int crc = 0x0000;
-        for (byte b : bytes) {
-            crc = (crc << 8) ^ LOOKUP_TABLE[((crc >>> 8) ^ (b & 0xFF)) & 0xFF];
-        }
-        return crc & 0xFFFF;
-    }
-
-    public static int calcSlot(byte[] key) {
-        if (key == null) {
-            return 0;
-        }
-        int start = Utility.indexOf(key, (byte) '{');
-        if (start != -1) {
-            int end = Utility.indexOf(key, start + 1, (byte) '}');
-            if (end != -1) {
-                key = Arrays.copyOfRange(key, start + 1, end);
-            }
-        }
-        return crc16(key) % MAX_SLOT;
-    }
-
-    public static int calcSlot(String key) {
-        if (key == null) {
-            return 0;
-        }
-        int start = key.indexOf('{');
-        if (start != -1) {
-            int end = key.indexOf('}');
-            if (end != -1 && start + 1 < end) {
-                key = key.substring(start + 1, end);
-            }
-        }
-        return crc16(key.getBytes()) % MAX_SLOT;
-    }
-}

src/main/java/org/redkalex/cache/redis/RedisCacheClient.java

@@ -1,58 +0,0 @@
-/*
- * To change this license header, choose License Headers in Project Properties.
- * To change this template file, choose Tools | Templates
- * and open the template in the editor.
- */
-package org.redkalex.cache.redis;
-
-import org.redkale.net.AsyncConnection;
-import org.redkale.net.AsyncGroup;
-import org.redkale.net.client.Client;
-import org.redkale.net.client.ClientAddress;
-import org.redkale.util.Traces;
-
-/**
- *
- * @author zhangjx
- */
-public class RedisCacheClient extends Client<RedisCacheConnection, RedisCacheRequest, RedisCacheResult> {
-
-    public RedisCacheClient(String appName, String name, AsyncGroup group, String key,
-        ClientAddress address, int maxConns, int maxPipelines, RedisCacheReqAuth authReq, RedisCacheReqDB dbReq) {
-        super(name, group, true, address, maxConns, maxPipelines, () -> new RedisCacheReqPing(), () -> new RedisCacheReqClose(), null); //maxConns
-        RedisCacheReqClientName clientNameReq = new RedisCacheReqClientName(appName, name);
-        if (authReq != null && dbReq != null) {
-            this.authenticate = traceid -> {
-                Traces.currentTraceid(traceid);
-                return conn -> writeChannelBatch(conn, authReq.createTime(), dbReq.createTime(), clientNameReq.createTime())
-                    .thenApply(v -> conn);
-            };
-        } else if (authReq != null) {
-            this.authenticate = traceid -> {
-                Traces.currentTraceid(traceid);
-                return conn -> writeChannelBatch(conn, authReq.createTime(), clientNameReq.createTime())
-                    .thenApply(v -> conn);
-            };
-        } else if (dbReq != null) {
-            this.authenticate = traceid -> {
-                Traces.currentTraceid(traceid);
-                return conn -> writeChannelBatch(conn, dbReq.createTime(), clientNameReq.createTime())
-                    .thenApply(v -> conn);
-            };
-        } else {
-            this.authenticate = traceid -> {
-                Traces.currentTraceid(traceid);
-                return conn -> writeChannel(conn, clientNameReq.createTime())
-                    .thenApply(v -> conn);
-            };
-        }
-        this.readTimeoutSeconds = 3;
-        this.writeTimeoutSeconds = 3;
-    }
-
-    @Override
-    protected RedisCacheConnection createClientConnection(final int index, AsyncConnection channel) {
-        return new RedisCacheConnection(this, index, channel);
-    }
-
-}

src/main/java/org/redkalex/cache/redis/RedisCacheCodec.java

@@ -1,255 +0,0 @@
-/*
- * To change this license header, choose License Headers in Project Properties.
- * To change this template file, choose Tools | Templates
- * and open the template in the editor.
- */
-package org.redkalex.cache.redis;
-
-import org.redkale.net.client.ClientCodec;
-import org.redkale.net.client.ClientConnection;
-import org.redkale.util.ByteArray;
-import org.redkale.util.RedkaleException;
-
-import java.nio.ByteBuffer;
-import java.nio.charset.StandardCharsets;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.logging.Logger;
-
-/**
- *
- * @author zhangjx
- */
-public class RedisCacheCodec extends ClientCodec<RedisCacheRequest, RedisCacheResult> {
-
-    protected static final byte TYPE_BULK = '$';  //字符串块类型, 例如：$6\r\n\abcdef\r\n，NULL字符串：$-1\r\n
-
-    protected static final byte TYPE_MULTI = '*'; //数组，紧接的数字为数组长度
-
-    protected static final byte TYPE_STRING = '+';  //字符串值类型，字符串以\r\n结尾, 例如：+OK\r\n 
-
-    protected static final byte TYPE_ERROR = '-'; //错误字符串类型，字符串以\r\n结尾, 例如：-ERR unknown command 'red'\r\n
-
-    protected static final byte TYPE_NUMBER = ':'; //整型， 例如：:2\r\n
-
-    private static final Logger logger = Logger.getLogger(RedisCacheCodec.class.getSimpleName());
-
-    private ByteArray halfFrameBytes;
-
-    private int halfFrameBulkLength = Integer.MIN_VALUE;
-
-    private int halfFrameMultiSize = Integer.MIN_VALUE;
-
-    private int halfFrameMultiItemIndex; //从0开始
-
-    private byte halfFrameMultiItemType;
-
-    private int halfFrameMultiItemLength = Integer.MIN_VALUE;
-
-    private byte frameType;
-
-    private byte[] frameCursor;
-
-    private byte[] frameValue;
-
-    private List<byte[]> frameList;
-
-    private ByteArray recyclableArray;
-
-    public RedisCacheCodec(ClientConnection connection) {
-        super(connection);
-    }
-
-    private ByteArray pollArray(ByteArray array) {
-        if (recyclableArray == null) {
-            recyclableArray = new ByteArray();
-        }
-        recyclableArray.clear();
-        if (array != null) {
-            recyclableArray.put(array);
-        }
-        return recyclableArray;
-    }
-
-    private boolean readFrames(RedisCacheConnection conn, ByteBuffer buffer, ByteArray array) {
-//        byte[] dbs = new byte[buffer.remaining()];
-//        for (int i = 0; i < dbs.length; i++) {
-//            dbs[i] = buffer.get(buffer.position() + i);
-//        }
-//        (System. out).println("[" + Utility.nowMillis() + "] [" + Thread.currentThread().getName() + "]: " + conn + ", 原始数据: " + new String(dbs).replace("\r\n", "  "));
-
-        array.clear();
-        if (this.frameType == 0) {
-            this.frameType = buffer.get();
-        } else if (halfFrameBytes != null) {
-            array.put(halfFrameBytes);
-            halfFrameBytes = null;
-        }
-        if (frameType == TYPE_STRING || frameType == TYPE_ERROR || frameType == TYPE_NUMBER) {
-            if (!readComplete(buffer, array)) {
-                halfFrameBytes = pollArray(array);
-                return false;
-            }
-            frameValue = array.getBytes();
-        } else if (frameType == TYPE_BULK) {
-            if (halfFrameBulkLength == Integer.MIN_VALUE) {
-                if (!readComplete(buffer, array)) { //没有读到bulkLength
-                    halfFrameBytes = pollArray(array);
-                    return false;
-                }
-                halfFrameBulkLength = readInt(array);
-                array.clear();
-            }
-            if (halfFrameBulkLength == -1) {
-                frameValue = null;
-            } else {
-                int expect = halfFrameBulkLength + 2 - array.length();
-                if (buffer.remaining() < expect) {
-                    array.put(buffer);
-                    halfFrameBytes = pollArray(array);
-                    return false;
-                }
-                array.put(buffer, expect);
-                array.removeLastByte(); //移除\n
-                array.removeLastByte(); //移除\r
-                frameValue = array.getBytes();
-            }
-        } else if (frameType == TYPE_MULTI) {
-            int size = halfFrameMultiSize;
-            if (size == Integer.MIN_VALUE) {
-                if (!readComplete(buffer, array)) { //没有读到bulkLength
-                    halfFrameBytes = pollArray(array);
-                    return false;
-                }
-                size = readInt(array);
-                halfFrameMultiSize = size;
-                array.clear();
-                frameValue = null;
-            }
-            if (frameList == null) {
-                frameList = new ArrayList<>();
-            }
-            if (size > 0) {
-                int index = halfFrameMultiItemIndex;
-                for (int i = index; i < size; i++) {
-                    if (!buffer.hasRemaining()) {
-                        return false;
-                    }
-                    if (halfFrameMultiItemType == 0) {
-                        halfFrameMultiItemType = buffer.get();
-                    }
-                    halfFrameMultiItemIndex = i;
-                    final byte itemType = halfFrameMultiItemType;
-                    if (itemType == TYPE_STRING || itemType == TYPE_ERROR || itemType == TYPE_NUMBER) {
-                        if (!readComplete(buffer, array)) {
-                            halfFrameBytes = pollArray(array);
-                            return false;
-                        }
-                        frameList.add(array.getBytes());
-                    } else if (itemType == TYPE_BULK) {
-                        if (halfFrameMultiItemLength == Integer.MIN_VALUE) {
-                            if (!readComplete(buffer, array)) { //没有读到bulkLength
-                                halfFrameBytes = pollArray(array);
-                                return false;
-                            }
-                            halfFrameMultiItemLength = readInt(array);
-                            array.clear();
-                        }
-                        if (halfFrameMultiItemLength == -1) {
-                            frameList.add(null);
-                        } else {
-                            int expect = halfFrameMultiItemLength + 2 - array.length();
-                            if (buffer.remaining() < expect) {
-                                array.put(buffer);
-                                halfFrameBytes = pollArray(array);
-                                return false;
-                            }
-                            array.put(buffer, expect);
-                            array.removeLastByte(); //移除\n
-                            array.removeLastByte(); //移除\r
-                            frameList.add(array.getBytes());
-                        }
-                    } else if (itemType == TYPE_MULTI) { //数组中嵌套数组，例如: SCAN、HSCAN   
-                        if (size == 2 && frameList != null && frameList.size() == 1) {
-                            //读游标 数据例如: *2  $1  0  *4  $4  key1  $2  10  $4  key2  $2  30 
-                            frameCursor = frameList.get(0);
-                            frameList.clear();
-                            clearHalfFrame();
-                            return readFrames(conn, buffer, array);
-                        } else {
-                            throw new RedkaleException("Not support multi type in array data");
-                        }
-                    }
-                    halfFrameMultiItemType = 0;
-                    halfFrameMultiItemLength = Integer.MIN_VALUE;
-                    array.clear();
-                }
-            }
-        }
-        return true;
-    }
-
-    private void clearHalfFrame() {
-        halfFrameBytes = null;
-        halfFrameBulkLength = Integer.MIN_VALUE;
-        halfFrameMultiSize = Integer.MIN_VALUE;
-        halfFrameMultiItemLength = Integer.MIN_VALUE;
-        halfFrameMultiItemIndex = 0; //从0开始
-        halfFrameMultiItemType = 0;
-    }
-
-    @Override
-    public void decodeMessages(ByteBuffer realbuf, ByteArray array) {
-        RedisCacheConnection conn = (RedisCacheConnection) connection;
-        if (!realbuf.hasRemaining()) {
-            return;
-        }
-        ByteBuffer buffer = realbuf;
-
-        while (buffer.hasRemaining()) {
-            if (!readFrames(conn, buffer, array)) {
-                break;
-            }
-            RedisCacheRequest request = nextRequest();
-            if (frameType == TYPE_ERROR) {
-                addMessage(request, new RedkaleException(new String(frameValue, StandardCharsets.UTF_8)));
-            } else {
-                addMessage(request, conn.pollResultSet(request).prepare(frameType, frameCursor, frameValue, frameList));
-            }
-            frameType = 0;
-            frameCursor = null;
-            frameValue = null;
-            frameList = null;
-            clearHalfFrame();
-            buffer = realbuf;
-        }
-    }
-
-    protected RedisCacheRequest nextRequest() {
-        return super.nextRequest();
-    }
-
-    private boolean readComplete(ByteBuffer buffer, ByteArray array) {
-        while (buffer.hasRemaining()) {
-            byte b = buffer.get();
-            if (b == '\n') {
-                array.removeLastByte(); //移除 \r
-                return true;
-            }
-            array.put(b);
-        }
-        return false;
-    }
-
-    private int readInt(ByteArray array) {
-        String val = array.toString(StandardCharsets.ISO_8859_1);
-        if (val.length() == 1 && val.charAt(0) == '0') {
-            return 0;
-        }
-        if (val.length() == 2 && val.charAt(0) == '-' && val.charAt(1) == '1') {
-            return -1;
-        }
-        return Integer.parseInt(val);
-    }
-
-}

src/main/java/org/redkalex/cache/redis/RedisCacheConnection.java

@@ -1,64 +0,0 @@
-/*
- * To change this license header, choose License Headers in Project Properties.
- * To change this template file, choose Tools | Templates
- * and open the template in the editor.
- */
-package org.redkalex.cache.redis;
-
-import org.redkale.net.AsyncConnection;
-import org.redkale.net.WorkThread;
-import org.redkale.net.client.Client;
-import org.redkale.net.client.ClientCodec;
-import org.redkale.net.client.ClientConnection;
-import org.redkale.net.client.ClientFuture;
-
-import java.io.Serializable;
-import java.util.List;
-import java.util.concurrent.CompletableFuture;
-import java.util.function.Function;
-
-/**
- *
- * @author zhangjx
- */
-public class RedisCacheConnection extends ClientConnection<RedisCacheRequest, RedisCacheResult> {
-
-    public RedisCacheConnection(Client client, int index, AsyncConnection channel) {
-        super(client, index, channel);
-    }
-
-    @Override
-    protected ClientCodec createCodec() {
-        return new RedisCacheCodec(this);
-    }
-
-    protected CompletableFuture<RedisCacheResult> writeRequest(RedisCacheRequest request) {
-        return super.writeChannel(request);
-    }
-
-    protected CompletableFuture<List<RedisCacheResult>> writeRequest(RedisCacheRequest[] requests) {
-        return super.writeChannel(requests);
-    }
-
-    protected <T> CompletableFuture<T> writeRequest(RedisCacheRequest request, Function<RedisCacheResult, T> respTransfer) {
-        return super.writeChannel(request, respTransfer);
-    }
-
-    protected <T> CompletableFuture<List<T>> writeRequest(RedisCacheRequest[] requests, Function<RedisCacheResult, T> respTransfer) {
-        return super.writeChannel(requests, respTransfer);
-    }
-
-    public RedisCacheResult pollResultSet(RedisCacheRequest request) {
-        RedisCacheResult rs = new RedisCacheResult();
-        return rs;
-    }
-
-    public RedisCacheRequest pollRequest(WorkThread workThread, String traceid) {
-        RedisCacheRequest rs = new RedisCacheRequest().workThread(workThread).traceid(traceid);
-        return rs;
-    }
-
-    protected ClientFuture<RedisCacheRequest, RedisCacheResult> pollRespFuture(Serializable requestid) {
-        return super.pollRespFuture(requestid);
-    }
-}

src/main/java/org/redkalex/cache/redis/RedisCacheReqAuth.java

@@ -1,44 +0,0 @@
-/*
- * To change this license header, choose License Headers in Project Properties.
- * To change this template file, choose Tools | Templates
- * and open the template in the editor.
- */
-package org.redkalex.cache.redis;
-
-import org.redkale.net.client.ClientConnection;
-import org.redkale.util.ByteArray;
-
-import java.nio.charset.StandardCharsets;
-
-/**
- *
- * @author zhangjx
- */
-public class RedisCacheReqAuth extends RedisCacheRequest {
-
-    private static final byte[] PS = "AUTH\r\n".getBytes(StandardCharsets.UTF_8);
-
-    protected String password;
-
-    public RedisCacheReqAuth(String password) {
-        this.password = password;
-    }
-
-    @Override
-    public void writeTo(ClientConnection conn, ByteArray writer) {
-        byte[] pwd = password.getBytes();
-        writer.put(mutliLengthBytes(2));
-        writer.put(bulkLengthBytes(4));
-        writer.put(PS);
-
-        writer.put(bulkLengthBytes(pwd.length));
-        writer.put(pwd);
-        writer.put(CRLF);
-
-    }
-
-    @Override
-    public String toString() {
-        return getClass().getSimpleName() + "{AUTH " + password + "}";
-    }
-}

src/main/java/org/redkalex/cache/redis/RedisCacheReqClientName.java

@@ -1,45 +0,0 @@
-/*
- *
- */
-package org.redkalex.cache.redis;
-
-import org.redkale.net.client.ClientConnection;
-import org.redkale.util.ByteArray;
-import org.redkale.util.Utility;
-
-import java.nio.charset.StandardCharsets;
-
-/**
- *
- * @author zhangjx
- */
-public class RedisCacheReqClientName extends RedisCacheRequest {
-
-    private final String clientName;
-
-    public RedisCacheReqClientName(String appName, String resourceName) {
-        this.clientName = "redkalex" + (Utility.isEmpty(appName) ? "" : ("-" + appName))
-            + (Utility.isEmpty(resourceName) ? "" : (":" + resourceName));
-    }
-
-    @Override
-    public void writeTo(ClientConnection conn, ByteArray writer) {
-        writer.put(mutliLengthBytes(3));
-
-        writer.put(bulkLengthBytes(6));
-        writer.put("CLIENT\r\n".getBytes(StandardCharsets.UTF_8));
-
-        writer.put(bulkLengthBytes(7));
-        writer.put("SETNAME\r\n".getBytes(StandardCharsets.UTF_8));
-
-        byte[] ns = clientName.getBytes(StandardCharsets.UTF_8);
-        writer.put(bulkLengthBytes(ns.length));
-        writer.put(ns);
-        writer.put(CRLF);
-    }
-
-    @Override
-    public String toString() {
-        return getClass().getSimpleName() + "{CLIENT SETNAME " + clientName + "}";
-    }
-}

src/main/java/org/redkalex/cache/redis/RedisCacheReqClose.java

@@ -1,43 +0,0 @@
-/*
- * To change this license header, choose License Headers in Project Properties.
- * To change this template file, choose Tools | Templates
- * and open the template in the editor.
- */
-package org.redkalex.cache.redis;
-
-import org.redkale.net.client.ClientConnection;
-import org.redkale.util.ByteArray;
-
-import java.nio.charset.StandardCharsets;
-
-/**
- *
- * @author zhangjx
- */
-public class RedisCacheReqClose extends RedisCacheRequest {
-
-    private static final byte[] BYTES = new ByteArray()
-        .put((byte) '*')
-        .put((byte) '1')
-        .put((byte) '\r', (byte) '\n')
-        .put((byte) '$')
-        .put((byte) '4')
-        .put((byte) '\r', (byte) '\n')
-        .put("QUIT".getBytes(StandardCharsets.UTF_8))
-        .put((byte) '\r', (byte) '\n').getBytes();
-
-    @Override
-    public final boolean isCloseType() {
-        return true;
-    }
-
-    @Override
-    public void writeTo(ClientConnection conn, ByteArray writer) {
-        writer.put(BYTES);
-    }
-
-    @Override
-    public String toString() {
-        return getClass().getSimpleName() + "{QUIT}";
-    }
-}

src/main/java/org/redkalex/cache/redis/RedisCacheReqDB.java

@@ -1,42 +0,0 @@
-/*
- * To change this license header, choose License Headers in Project Properties.
- * To change this template file, choose Tools | Templates
- * and open the template in the editor.
- */
-package org.redkalex.cache.redis;
-
-import org.redkale.net.client.ClientConnection;
-import org.redkale.util.ByteArray;
-
-import java.nio.charset.StandardCharsets;
-
-/**
- *
- * @author zhangjx
- */
-public class RedisCacheReqDB extends RedisCacheRequest {
-
-    protected int db;
-
-    public RedisCacheReqDB(int db) {
-        this.db = db;
-    }
-
-    @Override
-    public void writeTo(ClientConnection conn, ByteArray writer) {
-        writer.put(mutliLengthBytes(2));
-        writer.put(bulkLengthBytes(6));
-        writer.put("SELECT\r\n".getBytes(StandardCharsets.UTF_8));
-
-        byte[] dbs = String.valueOf(db).getBytes(StandardCharsets.UTF_8);
-        writer.put(bulkLengthBytes(dbs.length));
-        writer.put(dbs);
-        writer.put(CRLF);
-
-    }
-
-    @Override
-    public String toString() {
-        return getClass().getSimpleName() + "{SELECT " + db + "}";
-    }
-}

src/main/java/org/redkalex/cache/redis/RedisCacheReqPing.java

@@ -1,38 +0,0 @@
-/*
- * To change this license header, choose License Headers in Project Properties.
- * To change this template file, choose Tools | Templates
- * and open the template in the editor.
- */
-package org.redkalex.cache.redis;
-
-import org.redkale.net.client.ClientConnection;
-import org.redkale.util.ByteArray;
-
-import java.nio.charset.StandardCharsets;
-
-/**
- *
- * @author zhangjx
- */
-public class RedisCacheReqPing extends RedisCacheRequest {
-
-    private static final byte[] BYTES = new ByteArray()
-        .put((byte) '*')
-        .put((byte) '1')
-        .put((byte) '\r', (byte) '\n')
-        .put((byte) '$')
-        .put((byte) '4')
-        .put((byte) '\r', (byte) '\n')
-        .put("PING".getBytes(StandardCharsets.UTF_8))
-        .put((byte) '\r', (byte) '\n').getBytes();
-
-    @Override
-    public void writeTo(ClientConnection conn, ByteArray writer) {
-        writer.put(BYTES);
-    }
-
-    @Override
-    public String toString() {
-        return getClass().getSimpleName() + "{PING}";
-    }
-}

src/main/java/org/redkalex/cache/redis/RedisCacheRequest.java

@@ -1,116 +0,0 @@
-/*
- * To change this license header, choose License Headers in Project Properties.
- * To change this template file, choose Tools | Templates
- * and open the template in the editor.
- */
-package org.redkalex.cache.redis;
-
-import org.redkale.net.client.ClientConnection;
-import org.redkale.net.client.ClientRequest;
-import org.redkale.util.ByteArray;
-
-import java.nio.charset.StandardCharsets;
-
-/**
- *
- * @author zhangjx
- */
-public class RedisCacheRequest extends ClientRequest {
-
-    static final byte[] BYTES_TRUE = new byte[]{'t'};
-
-    static final byte[] BYTES_FALSE = new byte[]{'f'};
-
-    static final byte[] BYTES_MATCH = "MATCH".getBytes(StandardCharsets.UTF_8);
-
-    static final byte[] BYTES_COUNT = "COUNT".getBytes(StandardCharsets.UTF_8);
-
-    protected static final byte[] CRLF = new byte[]{'\r', '\n'};
-
-    private static final byte[][] starLengthBytes;
-
-    private static final byte[][] dollarLengthBytes;
-
-    static {
-        starLengthBytes = new byte[1024][];
-        dollarLengthBytes = new byte[1024][];
-        for (int i = 0; i < dollarLengthBytes.length; i++) {
-            starLengthBytes[i] = ("*" + i + "\r\n").getBytes(StandardCharsets.ISO_8859_1);
-            dollarLengthBytes[i] = ("$" + i + "\r\n").getBytes(StandardCharsets.ISO_8859_1);
-        }
-    }
-
-    protected RedisCommand command;
-
-    protected String key;
-
-    protected byte[][] args;
-
-    public static RedisCacheRequest create(RedisCommand command, String key, String... args) {
-        return new RedisCacheRequest().prepare(command, key, RedisCacheSource.keysArgs(key, args));
-    }
-
-    public static RedisCacheRequest create(RedisCommand command, String key, byte[]... args) {
-        return new RedisCacheRequest().prepare(command, key, args);
-    }
-
-    public RedisCacheRequest prepare(RedisCommand command, String key, byte[]... args) {
-        super.prepare();
-        this.command = command;
-        this.key = key;
-        this.args = args;
-        return this;
-    }
-
-    public RedisCacheRequest createTime() {
-        this.createTime = System.currentTimeMillis();
-        return this;
-    }
-
-    @Override
-    public void writeTo(ClientConnection conn, ByteArray writer) {
-        writer.put(mutliLengthBytes(args.length + 1));
-        writer.put(command.getBytes());
-
-        for (final byte[] arg : args) {
-            putArgBytes(writer, arg);
-        }
-    }
-
-    protected void putArgBytes(ByteArray writer, byte[] arg) {
-        writer.put(bulkLengthBytes(arg.length));
-        writer.put(arg);
-        writer.put(CRLF);
-    }
-
-    protected static byte[] mutliLengthBytes(int length) {
-        if (length >= 0 && length < starLengthBytes.length) {
-            return starLengthBytes[length];
-        } else {
-            return ("*" + length + "\r\n").getBytes(StandardCharsets.ISO_8859_1);
-        }
-    }
-
-    protected static byte[] bulkLengthBytes(int length) {
-        if (length >= 0 && length < dollarLengthBytes.length) {
-            return dollarLengthBytes[length];
-        } else {
-            return ("$" + length + "\r\n").getBytes(StandardCharsets.ISO_8859_1);
-        }
-    }
-
-    @Override
-    public String toString() {
-        if (args == null || args.length == 0) {
-            return getClass().getSimpleName() + "{" + command + " " + key + "}";
-        } else {
-            StringBuilder sb = new StringBuilder();
-            sb.append(command);
-            sb.append(" ").append(key);
-            for (final byte[] arg : args) {
-                sb.append(" ").append(arg == null ? null : new String(arg, StandardCharsets.UTF_8));
-            }
-            return sb.toString();
-        }
-    }
-}

src/main/java/org/redkalex/cache/redis/RedisCacheResult.java

@@ -1,262 +0,0 @@
-/*
- * To change this license header, choose License Headers in Project Properties.
- * To change this template file, choose Tools | Templates
- * and open the template in the editor.
- */
-package org.redkalex.cache.redis;
-
-import org.redkale.convert.json.JsonConvert;
-import org.redkale.net.client.ClientResult;
-import org.redkale.source.CacheScoredValue;
-
-import java.lang.reflect.Type;
-import java.math.BigDecimal;
-import java.math.BigInteger;
-import java.nio.charset.StandardCharsets;
-import java.util.*;
-
-/**
- *
- * @author zhangjx
- */
-public class RedisCacheResult implements ClientResult {
-
-    //$   块字符串类型
-    //*   数组
-    //+   简单字符串类型
-    //-   错误类型
-    //:   整型
-    protected byte frameType;
-
-    protected byte[] frameCursor;
-
-    protected byte[] frameValue;
-
-    protected List<byte[]> frameList;
-
-    public RedisCacheResult prepare(byte byteType, byte[] frameCursor, byte[] frameValue, List<byte[]> frameList) {
-        this.frameType = byteType;
-        this.frameCursor = frameCursor;
-        this.frameValue = frameValue;
-        this.frameList = frameList;
-        return this;
-    }
-
-    @Override
-    public boolean isKeepAlive() {
-        return true;
-    }
-
-    public Void getVoidValue() {
-        return null;
-    }
-
-    public byte[] getFrameValue() {
-        return frameValue;
-    }
-
-    public int getCursor() {
-        if (frameCursor == null || frameCursor.length < 1) {
-            return -1;
-        } else {
-            return Integer.parseInt(new String(frameCursor));
-        }
-    }
-
-    public Boolean getBoolValue() {
-        if (frameValue == null) {
-            return false;
-        }
-        String val = new String(frameValue, StandardCharsets.UTF_8);
-        if ("OK".equals(val)) {
-            return true;
-        }
-        if (val.isEmpty()) {
-            return false;
-        }
-        for (char ch : val.toCharArray()) {
-            if (!Character.isDigit(ch)) {
-                return false;
-            }
-        }
-        return Integer.parseInt(val) > 0;
-    }
-
-    public Double getDoubleValue(Double defValue) {
-        if (frameValue == null) {
-            return defValue;
-        }
-        String val = new String(frameValue, StandardCharsets.UTF_8);
-        if ("nan".equalsIgnoreCase(val) || "-nan".equalsIgnoreCase(val)) {
-            return Double.NaN;
-        } else if ("inf".equalsIgnoreCase(val)) {
-            return Double.POSITIVE_INFINITY;
-        } else if ("-inf".equalsIgnoreCase(val)) {
-            return Double.NEGATIVE_INFINITY;
-        } else if ("-1".equalsIgnoreCase(val)) {
-            return -1.0;
-        } else if ("0".equalsIgnoreCase(val)) {
-            return 0.0;
-        } else if ("1".equalsIgnoreCase(val)) {
-            return 1.0;
-        } else {
-            return Double.parseDouble(val);
-        }
-    }
-
-    public Long getLongValue(Long defValue) {
-        if (frameValue == null) {
-            return defValue;
-        }
-        String val = new String(frameValue, StandardCharsets.UTF_8);
-        if ("-1".equalsIgnoreCase(val)) {
-            return -1L;
-        } else if ("0".equalsIgnoreCase(val)) {
-            return 0L;
-        } else if ("1".equalsIgnoreCase(val)) {
-            return 1L;
-        } else {
-            return Long.parseLong(val);
-        }
-    }
-
-    public Integer getIntValue(Integer defValue) {
-        if (frameValue == null) {
-            return defValue;
-        }
-        String val = new String(frameValue, StandardCharsets.UTF_8);
-        if ("-1".equalsIgnoreCase(val)) {
-            return -1;
-        } else if ("0".equalsIgnoreCase(val)) {
-            return 0;
-        } else if ("1".equalsIgnoreCase(val)) {
-            return 1;
-        } else {
-            return Integer.parseInt(val);
-        }
-    }
-
-    public <T> T getObjectValue(String key, RedisCryptor cryptor, Type type) {
-        return decodeValue(key, cryptor, frameValue, type);
-    }
-
-    protected <T> Set<T> getSetValue(String key, RedisCryptor cryptor, Type type) {
-        if (frameList == null || frameList.isEmpty()) {
-            return new LinkedHashSet<>();
-        }
-        Set<T> set = new LinkedHashSet<>();
-        for (byte[] bs : frameList) {
-            set.add(decodeValue(key, cryptor, bs, type));
-        }
-        return set;
-    }
-
-    protected List<CacheScoredValue> getScoreListValue(String key, RedisCryptor cryptor, Type scoreType) {
-        if (frameList == null || frameList.isEmpty()) {
-            return new ArrayList<>();
-        }
-        List<CacheScoredValue> set = new ArrayList<>();
-        for (int i = 0; i < frameList.size(); i += 2) {
-            byte[] bs1 = frameList.get(i);
-            byte[] bs2 = frameList.get(i + 1);
-            Number val = decodeValue(key, cryptor, bs2, scoreType);
-            if (val != null) {
-                set.add(CacheScoredValue.create(val, new String(bs1, StandardCharsets.UTF_8)));
-            }
-        }
-        return set;
-    }
-
-    protected <T> List<T> getListValue(String key, RedisCryptor cryptor, Type type) {
-        if (frameList == null || frameList.isEmpty()) {
-            return new ArrayList<>();
-        }
-        List<T> list = new ArrayList<>();
-        for (byte[] bs : frameList) {
-            list.add(decodeValue(key, cryptor, bs, type));
-        }
-        return list;
-    }
-
-    protected <T> Map<String, T> getMapValue(String key, RedisCryptor cryptor, Type type) {
-        if (frameList == null || frameList.isEmpty()) {
-            return new LinkedHashMap<>();
-        }
-        Map<String, T> map = new LinkedHashMap<>();
-        for (int i = 0; i < frameList.size(); i += 2) {
-            byte[] bs1 = frameList.get(i);
-            byte[] bs2 = frameList.get(i + 1);
-            T val = decodeValue(key, cryptor, bs2, type);
-            if (val != null) {
-                map.put(decodeValue(key, cryptor, bs1, String.class).toString(), val);
-            }
-        }
-        return map;
-    }
-
-    protected static <T> T decodeValue(String key, RedisCryptor cryptor, byte[] frames, Type type) {
-        if (frames == null) {
-            return null;
-        }
-        if (type == byte[].class) {
-            return (T) frames;
-        }
-        if (type == String.class) {
-            String val = new String(frames, StandardCharsets.UTF_8);
-            if (cryptor != null) {
-                val = cryptor.decrypt(key, val);
-            }
-            return (T) val;
-        }
-        if (type == int.class || type == Integer.class) {
-            return (T) (Integer) Integer.parseInt(new String(frames, StandardCharsets.UTF_8));
-        }
-        if (type == long.class || type == Long.class) {
-            return (T) (Long) Long.parseLong(new String(frames, StandardCharsets.UTF_8));
-        }
-        if (type == float.class || type == Float.class) {
-            return (T) (Float) Float.parseFloat(new String(frames, StandardCharsets.UTF_8));
-        }
-        if (type == BigInteger.class) {
-            return (T) new BigInteger(new String(frames, StandardCharsets.UTF_8));
-        }
-        if (type == BigDecimal.class) {
-            return (T) new BigDecimal(new String(frames, StandardCharsets.UTF_8));
-        }
-        if (type == boolean.class || type == Boolean.class) {
-            String v = new String(frames, StandardCharsets.UTF_8);
-            return (T) (Boolean) ("t".equalsIgnoreCase(v) || "1".equals(v));
-        }
-        if (type == double.class || type == Double.class) {
-            return (T) (Double) Double.parseDouble(new String(frames, StandardCharsets.UTF_8));
-        }
-        if (cryptor != null) {
-            String val = cryptor.decrypt(key, new String(frames, StandardCharsets.UTF_8));
-            return (T) JsonConvert.root().convertFrom(type, val);
-        }
-        return (T) JsonConvert.root().convertFrom(type, frames);
-    }
-
-    @Override
-    public String toString() {
-        StringBuilder sb = new StringBuilder();
-        sb.append("{type: ").append(frameType);
-        if (frameValue != null) {
-            sb.append(", value: ").append(new String(frameValue, StandardCharsets.UTF_8));
-        }
-        if (frameList != null) {
-            sb.append(", list: [");
-            boolean first = true;
-            for (byte[] bs : frameList) {
-                if (!first) {
-                    sb.append(", ");
-                }
-                sb.append(bs == null ? null : new String(bs, StandardCharsets.UTF_8));
-                first = false;
-            }
-            sb.append("]");
-        }
-        return sb.append("}").toString();
-    }
-
-}

src/main/java/org/redkalex/cache/redis/RedisCacheSourceProvider.java

@@ -1,30 +0,0 @@
-/*
- * To change this license header, choose License Headers in Project Properties.
- * To change this template file, choose Tools | Templates
- * and open the template in the editor.
- */
-package org.redkalex.cache.redis;
-
-import org.redkale.annotation.Priority;
-import org.redkale.source.CacheSource;
-import org.redkale.source.CacheSourceProvider;
-import org.redkale.util.AnyValue;
-
-/**
- *
- * @author zhangjx
- */
-@Priority(-900)
-public class RedisCacheSourceProvider implements CacheSourceProvider {
-
-    @Override
-    public boolean acceptsConf(AnyValue config) {
-        return new RedisCacheSource().acceptsConf(config);
-    }
-
-    @Override
-    public CacheSource createInstance() {
-        return new RedisCacheSource();
-    }
-
-}

src/main/java/org/redkalex/cache/redis/RedisCommand.java

@@ -1,124 +0,0 @@
-/*
- *
- */
-package org.redkalex.cache.redis;
-
-import java.nio.charset.StandardCharsets;
-
-/**
- * @author zhangjx
- */
-public enum RedisCommand {
-
-    EXISTS("EXISTS", true),
-    GET("GET", true),
-    GETEX("GETEX", true),
-    MSET("MSET", false),
-    SET("SET", false),
-    SETNX("SETNX", false),
-    GETSET("GETSET", false),
-    GETDEL("GETDEL", false),
-    SETEX("SETEX", false),
-    EXPIRE("EXPIRE", false),
-    PERSIST("PERSIST", false),
-    RENAME("RENAME", false),
-    RENAMENX("RENAMENX", false),
-    DEL("DEL", false),
-    INCR("INCR", false),
-    INCRBY("INCRBY", false),
-    INCRBYFLOAT("INCRBYFLOAT", false),
-    DECR("DECR", false),
-    DECRBY("DECRBY", false),
-    HDEL("HDEL", false),
-    HINCRBY("HINCRBY", false),
-    HINCRBYFLOAT("HINCRBYFLOAT", false),
-    LINSERT("LINSERT", false),
-    LTRIM("LTRIM", false),
-    LPOP("LPOP", false),
-    LPUSH("LPUSH", false),
-    LPUSHX("LPUSHX", false),
-    RPOP("RPOP", false),
-    RPOPLPUSH("RPOPLPUSH", false),
-    SMOVE("SMOVE", false),
-    RPUSH("RPUSH", false),
-    RPUSHX("RPUSHX", false),
-    LREM("LREM", false),
-    SADD("SADD", false),
-    SPOP("SPOP", false),
-    SSCAN("SSCAN", true),
-    SREM("SREM", false),
-    ZADD("ZADD", false),
-    ZINCRBY("ZINCRBY", false),
-    ZREM("ZREM", false),
-    ZMSCORE("ZMSCORE", true),
-    ZSCORE("ZSCORE", true),
-    ZCARD("ZCARD", true),
-    ZRANK("ZRANK", true),
-    ZREVRANK("ZREVRANK", true),
-    ZRANGE("ZRANGE", true),
-    ZSCAN("ZSCAN", true),
-    FLUSHDB("FLUSHDB", false),
-    FLUSHALL("FLUSHALL", false),
-    HMGET("HMGET", true),
-    HLEN("HLEN", true),
-    HKEYS("HKEYS", true),
-    HEXISTS("HEXISTS", true),
-    HSET("HSET", true),
-    HSETNX("HSETNX", true),
-    HMSET("HMSET", true),
-    HSCAN("HSCAN", true),
-    HGETALL("HGETALL", true),
-    HVALS("HVALS", true),
-    HGET("HGET", true),
-    HSTRLEN("HSTRLEN", true),
-    SMEMBERS("SMEMBERS", true),
-    SISMEMBER("SISMEMBER", true),
-    LRANGE("LRANGE", true),
-    LINDEX("LINDEX", true),
-    LLEN("LLEN", true),
-    SCARD("SCARD", true),
-    SMISMEMBER("SMISMEMBER", true),
-    SRANDMEMBER("SRANDMEMBER", true),
-    SDIFF("SDIFF", true),
-    SDIFFSTORE("SDIFFSTORE", false),
-    SINTER("SINTER", true),
-    SINTERSTORE("SINTERSTORE", false),
-    SUNION("SUNION", true),
-    SUNIONSTORE("SUNIONSTORE", false),
-    MGET("MGET", true),
-    KEYS("KEYS", true),
-    SCAN("SCAN", true),
-    DBSIZE("DBSIZE", true),
-    TYPE("TYPE", true),
-    SUBSCRIBE("SUBSCRIBE", false),
-    UNSUBSCRIBE("UNSUBSCRIBE", false),
-    PUBLISH("PUBLISH", false),
-    PUBSUB("PUBSUB", true),
-    GETBIT("GETBIT", true),
-    SETBIT("SETBIT", false),
-    EVAL("EVAL", false);
-
-    private final String command;
-
-    private final byte[] bytes;
-
-    private final boolean readOnly;
-
-    private RedisCommand(String command, boolean readOnly) {
-        this.command = command;
-        this.readOnly = readOnly;
-        this.bytes = ("$" + command.length() + "\r\n" + command + "\r\n").getBytes(StandardCharsets.ISO_8859_1);
-    }
-
-    public String getCommand() {
-        return command;
-    }
-
-    public byte[] getBytes() {
-        return bytes;
-    }
-
-    public boolean isReadOnly() {
-        return readOnly;
-    }
-}

src/main/java/org/redkalex/cache/redis/RedisConfig.java

@@ -1,190 +0,0 @@
-/*
- *
- */
-package org.redkalex.cache.redis;
-
-import org.redkale.convert.json.JsonConvert;
-import org.redkale.util.AnyValue;
-import org.redkale.util.RedkaleException;
-import org.redkale.util.Utility;
-
-import java.net.URI;
-import java.util.ArrayList;
-import java.util.List;
-
-import static org.redkale.source.AbstractCacheSource.*;
-import static org.redkale.util.Utility.isEmpty;
-import static org.redkale.util.Utility.isNotEmpty;
-
-/**
- *
- * @author zhangjx
- */
-public class RedisConfig {
-
-    private boolean ssl;
-
-    private int db;
-
-    private String username;
-
-    private String password;
-
-    private int maxconns;
-
-    private int pipelines;
-
-    private List<String> addresses;
-
-    public static RedisConfig create(AnyValue conf) {
-        RedisConfig result = new RedisConfig();
-        int gdb = conf.getIntValue(CACHE_SOURCE_DB, 0);
-        String gusername = null;
-        String gpassword = conf.getValue(CACHE_SOURCE_PASSWORD);
-        int gmaxconns = conf.getIntValue(CACHE_SOURCE_MAXCONNS, Utility.cpus());
-        int gpipelines = conf.getIntValue(CACHE_SOURCE_PIPELINES, org.redkale.net.client.Client.DEFAULT_MAX_PIPELINES);
-        String nodes = conf.getValue(CACHE_SOURCE_NODES, conf.getValue("url", ""));
-        if (Utility.isEmpty(nodes)) {
-            throw new RedkaleException("Not found nodes config for redis");
-        }
-        List<String> addrs = new ArrayList<>();
-        for (String url : nodes.replace(',', ';').split(";")) {
-            String urluser = null;
-            String urlpwd = null;
-            String urldb = null;
-            String urlmaxconns = null;
-            String urlpipelines = null;
-            addrs.add(url);
-            if (url.startsWith("redis://")) { //兼容 redis://:1234@127.0.0.1:6379?db=2
-                URI uri = URI.create(url);
-                String userInfo = uri.getUserInfo();
-                if (isEmpty(userInfo)) {
-                    String authority = uri.getAuthority();
-                    if (authority != null && authority.indexOf('@') > 0) {
-                        userInfo = authority.substring(0, authority.indexOf('@'));
-                    }
-                }
-                if (isNotEmpty(userInfo)) {
-                    urlpwd = userInfo;
-                    if (urlpwd.startsWith(":")) {
-                        urlpwd = urlpwd.substring(1);
-                    } else {
-                        int index = urlpwd.indexOf(':');
-                        if (index > 0) {
-                            urluser = urlpwd.substring(0, index);
-                            urlpwd = urlpwd.substring(index + 1);
-                        }
-                    }
-                }
-                if (isNotEmpty(uri.getQuery())) {
-                    String[] qrys = uri.getQuery().split("&|=");
-                    for (int i = 0; i < qrys.length; i += 2) {
-                        if (CACHE_SOURCE_USER.equals(qrys[i])) {
-                            urluser = i == qrys.length - 1 ? "" : qrys[i + 1];
-                        } else if (CACHE_SOURCE_PASSWORD.equals(qrys[i])) {
-                            urlpwd = i == qrys.length - 1 ? "" : qrys[i + 1];
-                        } else if (CACHE_SOURCE_DB.equals(qrys[i])) {
-                            urldb = i == qrys.length - 1 ? "" : qrys[i + 1];
-                        } else if (CACHE_SOURCE_MAXCONNS.equals(qrys[i])) {
-                            urlmaxconns = i == qrys.length - 1 ? "" : qrys[i + 1];
-                        } else if (CACHE_SOURCE_PIPELINES.equals(qrys[i])) {
-                            urlpipelines = i == qrys.length - 1 ? "" : qrys[i + 1];
-                        }
-                    }
-                }
-                if (isNotEmpty(urluser)) {
-                    gusername = urluser;
-                }
-                if (isNotEmpty(urlpwd)) {
-                    gpassword = urlpwd;
-                }
-                if (isNotEmpty(urlmaxconns)) {
-                    gmaxconns = Integer.parseInt(urlmaxconns);
-                }
-                if (isNotEmpty(urlpipelines)) {
-                    gpipelines = Integer.parseInt(urlpipelines);
-                }
-                if (isNotEmpty(urldb)) {
-                    gdb = Integer.parseInt(urldb);
-                }
-            }
-        }
-        result.ssl = nodes.startsWith("rediss://");
-        result.db = gdb;
-        if (isNotEmpty(gusername)) {
-            result.username = gusername;
-        }
-        if (isNotEmpty(gpassword)) {
-            result.password = gpassword;
-        }
-        result.maxconns = gmaxconns;
-        result.pipelines = gpipelines;
-        result.addresses = addrs;
-        return result;
-    }
-
-    public boolean isSsl() {
-        return ssl;
-    }
-
-    public void setSsl(boolean ssl) {
-        this.ssl = ssl;
-    }
-
-    public String getUsername() {
-        return username;
-    }
-
-    public void setUsername(String username) {
-        this.username = username;
-    }
-
-    public String getPassword() {
-        return password;
-    }
-
-    public void setPassword(String password) {
-        this.password = password;
-    }
-
-    public int getDb() {
-        return db;
-    }
-
-    public void setDb(int db) {
-        this.db = db;
-    }
-
-    public int getMaxconns() {
-        return maxconns;
-    }
-
-    public int getMaxconns(int min) {
-        return Math.max(maxconns, min);
-    }
-
-    public void setMaxconns(int maxconns) {
-        this.maxconns = maxconns;
-    }
-
-    public int getPipelines() {
-        return pipelines;
-    }
-
-    public void setPipelines(int pipelines) {
-        this.pipelines = pipelines;
-    }
-
-    public List<String> getAddresses() {
-        return addresses;
-    }
-
-    public void setAddresses(List<String> addresses) {
-        this.addresses = addresses;
-    }
-
-    @Override
-    public String toString() {
-        return JsonConvert.root().convertTo(this);
-    }
-}

src/main/java/org/redkalex/cache/redis/RedisCryptor.java

@@ -1,49 +0,0 @@
-/*
- * Click nbfs://nbhost/SystemFileSystem/Templates/Licenses/license-default.txt to change this license
- * Click nbfs://nbhost/SystemFileSystem/Templates/Classes/Class.java to edit this template
- */
-package org.redkalex.cache.redis;
-
-import org.redkale.util.AnyValue;
-
-/**
- *
- * @author zhangjx
- */
-public interface RedisCryptor {
-
-    /**
-     * 初始化
-     *
-     * @param conf 配置
-     */
-    public void init(AnyValue conf);
-
-    /**
-     * 加密, 无需加密的key对应的值需要直接返回value
-     *
-     * @param key   key
-     * @param value 明文
-     *
-     * @return 密文
-     */
-    public String encrypt(String key, String value);
-
-    /**
-     * 解密, 无需解密的key对应的值需要直接返回value
-     *
-     * @param key   key
-     * @param value 密文
-     *
-     * @return 明文
-     */
-    public String decrypt(String key, String value);
-
-    /**
-     * 销毁
-     *
-     * @param conf 配置
-     */
-    public void destroy(AnyValue conf);
-
-}

src/main/java/tccn/AbstractConsumer.java

@@ -1,37 +1,41 @@
-package dev.zhub;
+package tccn;
 
-import org.redkale.convert.json.JsonConvert;
+import com.google.gson.Gson;
-import org.redkale.util.Resourcable;
+import com.google.gson.reflect.TypeToken;
-import org.redkale.util.TypeToken;
+import tccn.zhub.Rpc;
 
+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;
 
 /**
  * @author Liang
  * @data 2020-09-05 23:18
  */
-public abstract class AbstractConsumer extends ZhubAgentProvider implements IConsumer, Resourcable {
+public abstract class AbstractConsumer implements IConsumer {
 
-    protected JsonConvert convert = JsonConvert.root();
+    public Gson gson = Rpc.gson;
 
-    protected static String APP_NAME = "";
+    private Map<String, EventType> eventMap = new HashMap<>();
-
-    protected Map<String, EventType<?>> eventMap = new ConcurrentHashMap<>();
 
     protected abstract String getGroupid();
 
+    protected boolean preInit() {
+        return true;
+    }
+
     protected final Set<String> getTopics() {
         if (!eventMap.isEmpty()) {
             return eventMap.keySet();
         }
+        HashSet<String> set = new HashSet<>();
+        set.add("-");
 
-        return Set.of("-");
+        return set;
     }
 
-    // topic 消息消费前处理
     protected void accept(String topic, String value) {
         EventType eventType = eventMap.get(topic);
 
@@ -39,18 +43,12 @@ public abstract class AbstractConsumer extends ZhubAgentProvider implements ICon
         if ("java.lang.String".equals(eventType.typeToken.getType().getTypeName())) {
             data = value;
         } else {
-            data = convert.convertFrom(eventType.typeToken.getType(), value);
+            data = gson.fromJson(value, eventType.typeToken.getType());
         }
 
         eventType.accept(data);
     }
 
-    // rpc 被调用端
-    protected <T> void rpcAccept(String topic, T value) {
-        EventType eventType = eventMap.get(topic);
-        eventType.accept(value);
-    }
-
     protected final void removeEventType(String topic) {
         eventMap.remove(topic);
     }
@@ -63,7 +61,7 @@ public abstract class AbstractConsumer extends ZhubAgentProvider implements ICon
     protected abstract void subscribe(String topic);
 
     public void subscribe(String topic, Consumer<String> consumer) {
-        subscribe(topic, IType.STRING, consumer);
+        subscribe(topic, TYPE_TOKEN_STRING, consumer);
     }
 
     @Override
@@ -78,19 +76,4 @@ public abstract class AbstractConsumer extends ZhubAgentProvider implements ICon
         }
     }
 
-    // --------------
-
-    @Override
-    public String resourceName() {
-        return super.getName();
-    }
-
-    protected String toStr(Object v) {
-        if (v instanceof String) {
-            return (String) v;
-        } else if (v == null) {
-            return null;
-        }
-        return convert.convertTo(v);
-    }
 }

src/main/java/tccn/Event.java

@@ -1,4 +1,4 @@
-package dev.zhub;
+package tccn;
 
 /**
  * 发布订阅 事件

src/main/java/tccn/EventType.java

@@ -1,6 +1,7 @@
-package dev.zhub;
+package tccn;
 
-import org.redkale.util.TypeToken;
+
+import com.google.gson.reflect.TypeToken;
 
 import java.util.function.Consumer;
 
@@ -9,7 +10,7 @@ public class EventType<T> {
     public final TypeToken<T> typeToken;
     private final Consumer<T> consumer;
 
-    private final static TypeToken<String> stringToken = new TypeToken<>() {
+    private final static TypeToken<String> stringToken = new TypeToken<String>() {
     };
 
     private EventType(String topic, TypeToken<T> typeToken, Consumer<T> consumer) {

src/main/java/tccn/IConsumer.java

@@ -1,10 +1,14 @@
-package dev.zhub;
+package tccn;
 
-import org.redkale.util.TypeToken;
+import com.google.gson.reflect.TypeToken;
 
 import java.util.function.Consumer;
 
 public interface IConsumer {
+    TypeToken<String> TYPE_TOKEN_STRING = new TypeToken<String>() {
+    };
+    TypeToken<Integer> TYPE_TOKEN_INT = new TypeToken<Integer>() {
+    };
 
     /**
      * 取消订阅

src/main/java/tccn/IProducer.java

@@ -1,4 +1,4 @@
-package dev.zhub;
+package tccn;
 
 import java.util.logging.Logger;
 

src/main/java/tccn/IType.java

@@ -1,6 +1,6 @@
-package dev.zhub;
+package tccn;
 
-import org.redkale.util.TypeToken;
+import com.google.gson.reflect.TypeToken;
 
 import java.util.List;
 import java.util.Map;
@@ -13,21 +13,9 @@ public interface IType {
     TypeToken<Integer> INT = new TypeToken<Integer>() {
     };
 
-    TypeToken<Short> SHORT = new TypeToken<Short>() {
-    };
-
-    TypeToken<Long> LONG = new TypeToken<Long>() {
-    };
-
     TypeToken<Map<String, String>> MAP = new TypeToken<Map<String, String>>() {
     };
 
     TypeToken<List<Map<String, String>>> LMAP = new TypeToken<List<Map<String, String>>>() {
     };
-
-    TypeToken<List<String>> LSTRING = new TypeToken<List<String>>() {
-    };
-
-    TypeToken<List<Integer>> LINT = new TypeToken<List<Integer>>() {
-    };
 }

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

@@ -1,7 +1,7 @@
-package dev.zhub.timer;
+package tccn.timer;
 
-import dev.zhub.timer.queue.TimerQueue;
+import tccn.timer.queue.TimerQueue;
-import dev.zhub.timer.task.Task;
+import tccn.timer.task.Task;
 
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
@@ -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

@@ -1,8 +1,8 @@
-package dev.zhub.timer;
+package tccn.timer;
 
-import dev.zhub.timer.scheduled.Scheduled;
+import tccn.timer.scheduled.Scheduled;
-import dev.zhub.timer.task.Job;
+import tccn.timer.task.Job;
-import dev.zhub.timer.task.Task;
+import tccn.timer.task.Task;
 
 import java.time.LocalDateTime;
 import java.time.ZoneId;
@@ -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/timer/Timers.java

@@ -1,22 +1,19 @@
-package dev.zhub.timer;
+package tccn.timer;
 
-import dev.zhub.timer.scheduled.ScheduledCycle;
+import tccn.timer.scheduled.ScheduledCycle;
-import org.redkale.util.Utility;
 
+import java.util.UUID;
 import java.util.function.Supplier;
 
 public class Timers {
 
-    private static TimerExecutor timerExecutor = new TimerExecutor(1);
+    private static final TimerExecutor timerExecutor = new TimerExecutor(1);
 
     /**
      * 本地延时重试
-     * @param supplier
-     * @param millis
-     * @param maxCount
      */
     public static void tryDelay(Supplier<Boolean> supplier, long millis, int maxCount) {
-        timerExecutor.add(TimerTask.by("try-delay-task-" + Utility.uuid(), ScheduledCycle.of(0), task -> {
+        timerExecutor.add(TimerTask.by("try-delay-task-" + UUID.randomUUID().toString().replaceAll("-", ""), ScheduledCycle.of(0), task -> {
             if (supplier.get() || task.getExecCount() == maxCount) {
                 task.setComplete(true);
             }
@@ -31,11 +28,9 @@ public class Timers {
 
     /**
      * 本地延时：延时时间极短的场景下使用 （如：1分钟内）
-     * @param runnable
-     * @param millis
      */
     public static void delay(Runnable runnable, long millis) {
-        timerExecutor.add(TimerTask.by("delay-task-" + Utility.uuid(), ScheduledCycle.of(millis), task -> {
+        timerExecutor.add(TimerTask.by("delay-task-" + UUID.randomUUID().toString().replaceAll("-", ""), ScheduledCycle.of(millis), task -> {
             runnable.run();
             task.setComplete(true);
         }));

src/main/java/tccn/timer/queue/TimerQueue.java

@@ -1,6 +1,6 @@
-package dev.zhub.timer.queue;
+package tccn.timer.queue;
 
-import dev.zhub.timer.task.Task;
+import tccn.timer.task.Task;
 
 import java.util.LinkedList;
 import java.util.concurrent.TimeUnit;

src/main/java/tccn/timer/scheduled/Scheduled.java

@@ -1,4 +1,4 @@
-package dev.zhub.timer.scheduled;
+package tccn.timer.scheduled;
 
 import java.time.LocalDateTime;
 

src/main/java/tccn/timer/scheduled/ScheduledCycle.java

@@ -1,4 +1,4 @@
-package dev.zhub.timer.scheduled;
+package tccn.timer.scheduled;
 
 import java.time.LocalDateTime;
 import java.time.temporal.ChronoUnit;

src/main/java/tccn/timer/scheduled/ScheduledExpres.java

@@ -1,4 +1,4 @@
-package dev.zhub.timer.scheduled;
+package tccn.timer.scheduled;
 
 import java.time.LocalDate;
 import java.time.LocalDateTime;

src/main/java/tccn/timer/task/Job.java

@@ -1,4 +1,4 @@
-package dev.zhub.timer.task;
+package tccn.timer.task;
 
 /**
  * @author: liangxianyou at 2018/12/8 17:24.

src/main/java/tccn/timer/task/Task.java

@@ -1,7 +1,7 @@
-package dev.zhub.timer.task;
+package tccn.timer.task;
 
-import dev.zhub.timer.TimerExecutor;
+import tccn.timer.TimerExecutor;
-import dev.zhub.timer.scheduled.Scheduled;
+import tccn.timer.scheduled.Scheduled;
 
 /**
  * @author: liangxianyou at 2018/8/5 19:32.
@@ -63,6 +63,7 @@ public interface Task extends Runnable {
 
     /**
      * 得到总执行次数
+     *
      * @return
      */
     int getExecCount();

src/main/java/tccn/zhub/Lock.java

@@ -1,4 +1,4 @@
-package dev.zhub.client;
+package tccn.zhub;
 
 // ================================================== lock ==================================================
 public class Lock {

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

@@ -1,26 +1,28 @@
-package dev.zhub.client;
+package tccn.zhub;
 
-import org.redkale.convert.ConvertColumn;
+import com.google.gson.Gson;
-import org.redkale.service.RetResult;
+import com.google.gson.annotations.Expose;
-import org.redkale.util.TypeToken;
-import org.redkale.util.Utility;
 
 public class Rpc<T> {
+    /*public final static Gson gson = new GsonBuilder()
+            .excludeFieldsWithoutExposeAnnotation()
+            .create();*/
+    public final static Gson gson = new Gson();
+
     private String ruk;     // request unique key:
     private String topic;   // call topic
     private T value;        // call paras
 
+    @Expose(deserialize = false, serialize = false)
     private RpcResult rpcResult;
 
-    private TypeToken typeToken;
-
     public Rpc() {
     }
 
-    protected Rpc(String appname, String topic, T value) {
+    public Rpc(String appname, String ruk, String topic, Object value) {
-        this.ruk = appname + "::" + Utility.uuid();
+        this.ruk = appname + "::" + ruk;
         this.topic = topic;
-        this.value = value;
+        this.value = (T) gson.toJson(value);
     }
 
     public String getRuk() {
@@ -47,7 +49,7 @@ public class Rpc<T> {
         this.value = value;
     }
 
-    @ConvertColumn(ignore = true)
+
     public RpcResult getRpcResult() {
         return rpcResult;
     }
@@ -56,44 +58,16 @@ public class Rpc<T> {
         this.rpcResult = rpcResult;
     }
 
-    @ConvertColumn(ignore = true)
-    public TypeToken getTypeToken() {
-        return typeToken;
-    }
-
-    @ConvertColumn(ignore = true)
-    public void setTypeToken(TypeToken typeToken) {
-        this.typeToken = typeToken;
-    }
-
-    @ConvertColumn(ignore = true)
     public String getBackTopic() {
         return ruk.split("::")[0];
     }
 
-    public <R> RpcResult<R> render(int retcode, String retinfo) {
-        RpcResult<R> response = new RpcResult<>();
-        response.setRuk(ruk);
-        response.setRetcode(retcode);
-        response.setRetinfo(retinfo);
-        return response;
-    }
-
     public <R> RpcResult<R> render() {
         RpcResult<R> response = new RpcResult<>();
         response.setRuk(ruk);
         return response;
     }
 
-    public RpcResult render(RetResult result) {
-        RpcResult resp = new RpcResult<>();
-        resp.setRuk(ruk);
-        resp.setRetcode(result.getRetcode());
-        resp.setRetinfo(result.getRetinfo());
-        resp.setResult(result.getResult());
-        return resp;
-    }
-
     public <R> RpcResult<R> render(R result) {
         RpcResult<R> response = new RpcResult<>();
         response.setRuk(ruk);
@@ -116,5 +90,4 @@ public class Rpc<T> {
         response.setRetinfo(retinfo);
         return response;
     }
-
 }

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

@@ -1,4 +1,4 @@
-package dev.zhub.client;
+package tccn.zhub;
 
 public class RpcResult<R> {
     private String ruk;

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

@@ -0,0 +1,706 @@
+package tccn.zhub;
+
+import com.google.gson.reflect.TypeToken;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.stereotype.Component;
+import tccn.AbstractConsumer;
+import tccn.Event;
+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;
+import java.io.OutputStream;
+import java.net.InetSocketAddress;
+import java.net.Socket;
+import java.net.SocketException;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.*;
+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());
+    @Value("${zhub.addr}")
+    private String addr = "127.0.0.1:1216";
+
+    @Value("${zhub.groupid}")
+    private String groupid = "";
+
+    @Value("${zhub.auth}")
+    private String auth = "";
+
+    @Value("${zhub.appid}")
+    protected String appid = "";
+
+    @PostConstruct
+    public void init() {
+        init(null);
+    }
+
+    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<String>> rpcBackQueue = new LinkedBlockingQueue<>(); // RPC BACK MSG
+    private final LinkedBlockingQueue<Event<String>> rpcCallQueue = new LinkedBlockingQueue<>(); // RPC CALL MSG
+    private final LinkedBlockingQueue<String> sendMsgQueue = new LinkedBlockingQueue<>(); // SEND MSG
+
+    /*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(String addr, String groupid, String appid, String auth) {
+        this.addr = addr;
+        this.groupid = groupid;
+        this.appid = appid;
+        this.auth = auth;
+        init(null);
+    }
+
+    public void init(Map<String, String> config) {
+        if (!preInit()) {
+            return;
+        }
+
+        // 自动注入
+        if (config != null) {
+            addr = config.getOrDefault("addr", addr);
+            groupid = config.getOrDefault("groupid", groupid);
+            appid = config.getOrDefault("appname", appid);
+        }
+
+        // 设置第一个启动的 实例为主实例
+        /*if (isFirst) {
+            isMain = true;
+            isFirst = false;
+        }*/
+        if (!mainHub.containsKey(addr)) { // 确保同步执行此 init 逻辑
+            mainHub.put(addr, this);
+        }
+
+        if (!initSocket(0)) {
+            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(() -> {
+            ExecutorService executor = Executors.newSingleThreadExecutor();
+            while (true) {
+                Timer timer = null;
+                try {
+                    timer = timerQueue.take();
+                    long start = System.currentTimeMillis();
+                    executor.submit(timer.runnable).get(5, TimeUnit.SECONDS);
+                    long end = System.currentTimeMillis();
+                    logger.finest(String.format("timer [%s] : elapsed time %s ms", timer.name, end - start));
+                } catch (InterruptedException | ExecutionException | TimeoutException e) {
+                    if (e instanceof TimeoutException) {
+                        executor = Executors.newSingleThreadExecutor();
+                        logger.log(Level.WARNING, "TimeoutException [" + timer.name + "]", e);
+                    } else {
+                        e.printStackTrace();
+                    }
+                }
+            }
+        }).start();
+
+        // topic msg
+        new Thread(() -> {
+            ExecutorService executor = Executors.newSingleThreadExecutor();
+            while (true) {
+                Event<String> event = null;
+                try {
+                    event = topicQueue.take();
+                    logger.log(Level.FINE, "topic[" + event.topic + "] :" + event.value);
+
+                    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);
+                    }
+                }
+            }
+        }, "ZHub-topic-accept").start();
+        // rpc back
+        new Thread(() -> {
+            ExecutorService executor = Executors.newSingleThreadExecutor();
+            while (true) {
+                Event<String> event = null;
+                try {
+                    event = rpcBackQueue.take();
+                    logger.info(String.format("rpc-back:[%s]: %s", event.topic, event.value));
+                    rpcAccept(event.value);
+
+                    String value = event.value;
+                    executor.submit(() -> rpcAccept(value)).get(5, TimeUnit.SECONDS);
+
+                } 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();
+
+        // rpc call
+        new Thread(() -> {
+            ExecutorService executor = Executors.newSingleThreadExecutor();
+            while (true) {
+                Event<String> event = null;
+                try {
+                    event = rpcCallQueue.take();
+                    logger.info(String.format("rpc-call:[%s] %s", event.topic, event.value));
+
+                    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);
+                    }
+                }
+            }
+        }, "ZHub-rpc-call").start();
+
+        // send msg
+        new Thread(() -> {
+            while (true) {
+                String msg = null;
+                try {
+                    msg = sendMsgQueue.take();
+                    writer.write(msg.getBytes(UTF_8));
+                    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();
+                    }
+                }
+            }
+        }).start();
+
+    }
+
+    // ---------------------
+    // -- 消息发送 --
+    protected boolean send(String... data) {
+        if (data.length == 1) {
+            sendMsgQueue.add(data[0] + "\r\n");
+        } else if (data.length > 1) {
+            StringBuilder buf = new StringBuilder();
+            buf.append("*" + data.length + "\r\n");
+            for (String d : data) {
+                buf.append("$" + strLength(d) + "\r\n");
+                buf.append(d + "\r\n");
+            }
+            sendMsgQueue.add(buf.toString());
+        }
+        return true;
+    }
+
+    /*protected boolean send(String... data) {
+        try {
+            lock.lock();
+            if (data.length == 1) {
+                writer.write((data[0] + "\r\n").getBytes());
+            } else if (data.length > 1) {
+                writer.write(("*" + data.length + "\r\n").getBytes());
+                for (String d : data) {
+                    writer.write(("$" + d.length() + "\r\n").getBytes());
+                    writer.write((d + "\r\n").getBytes());
+                }
+            }
+            writer.flush();
+            return true;
+        } catch (IOException e) {
+            logger.log(Level.WARNING, "", e);
+        } finally {
+            lock.unlock();
+        }
+        return false;
+    }*/
+
+    private int strLength(String str) {
+        str = str.replaceAll("[^\\x00-\\xff]", "*");
+        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 {
+                String[] hostPort = addr.split(":");
+                String host = hostPort[0];
+                int port = Integer.parseInt(hostPort[1]);
+
+                //private ReentrantLock lock = new ReentrantLock();
+                Socket client = new Socket();
+                client.connect(new InetSocketAddress(host, port));
+                client.setKeepAlive(true);
+
+                writer = client.getOutputStream();
+                reader = new BufferedReader(new InputStreamReader(client.getInputStream()));
+
+                String groupid = getGroupid();
+                if (groupid == null || groupid.isEmpty()) {
+                    throw new RuntimeException("ZHubClient groupid can not is empty");
+                }
+                send("auth", auth);
+                send("groupid " + groupid);
+
+                StringBuffer buf = new StringBuffer("subscribe lock trylock");
+                /*if (isMain) {
+                }*/
+                if (mainHub.containsValue(this)) {
+                    buf.append(" ").append(appid);
+                }
+                for (String topic : getTopics()) {
+                    buf.append(" ").append(topic);
+                }
+                send(buf.toString());
+
+                // 重连 timer 订阅
+                timerMap.forEach((name, timer) -> send("timer", name));
+                if (retry > 0) {
+                    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(), retry > 0 ? "reconnection" : "init"));
+                }
+                return true;
+            } catch (Exception e) {
+                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);
+                    } catch (InterruptedException interruptedException) {
+                        interruptedException.printStackTrace();
+                    }
+                } else {
+                    logger.log(Level.WARNING, String.format("ZHubClient[%s][%s] reconnection Failed！", getGroupid(), i + 1));
+                    try {
+                        Thread.sleep(1000 * 5);
+                    } catch (InterruptedException interruptedException) {
+                        interruptedException.printStackTrace();
+                    }
+                }
+            }
+        }
+        return false;
+    }
+
+    @Override
+    public void unsubscribe(String topic) {
+        send("unsubscribe " + topic);
+        super.removeEventType(topic);
+    }
+
+    public boolean publish(String topic, Object v) {
+        return send("publish", topic, toStr(v));
+    }
+
+    public void broadcast(String topic, Object v) {
+        send("broadcast", topic, toStr(v));
+    }
+
+    // 发送 publish 主题消息，若多次发送的 topic + "-" + value 相同，将会做延时重置
+    public void delay(String topic, Object v, long millis) {
+        send("delay", topic, toStr(v), String.valueOf(millis));
+    }
+
+    // 表达式支持：d+[d,H,m,s]
+    public void delay(String topic, Object v, String delayExpr) {
+        String endchar = "";
+        int delay;
+        if (delayExpr.matches("^\\d+[d,H,m,s]$")) {
+            endchar = delayExpr.substring(delayExpr.length() - 1);
+            delay = Integer.parseInt(delayExpr.substring(0, delayExpr.length() - 1));
+        } else {
+            if (!delayExpr.matches("^\\d+$")) {
+                throw new IllegalArgumentException(String.format("ScheduledCycle period config error: [%s]", delayExpr));
+            }
+
+            delay = Integer.parseInt(delayExpr);
+            if (delay <= 0L) {
+                throw new IllegalArgumentException(String.format("ScheduledCycle period config error: [%s]", delayExpr));
+            }
+        }
+
+        switch (endchar) {
+            case "M":
+                delay *= (1000 * 60 * 60 * 24 * 30);
+                break;
+            case "d":
+                delay *= (1000 * 60 * 60 * 24);
+                break;
+            case "H":
+                delay *= (1000 * 60 * 60);
+                break;
+            case "m":
+                delay *= (1000 * 60);
+                break;
+            case "s":
+                delay *= 1000;
+                break;
+        }
+
+        delay(topic, v, delay);
+    }
+
+    @Override
+    protected String getGroupid() {
+        return groupid;
+    }
+
+    @Override
+    protected void subscribe(String topic) {
+        send("subscribe " + topic); //新增订阅
+    }
+
+    // ================================================== lock ==================================================
+    private final Map<String, Lock> lockTag = new ConcurrentHashMap<>();
+
+    /**
+     * 尝试加锁，立即返回，
+     *
+     * @param key
+     * @param duration
+     * @return Lock: lock.success 锁定是否成功标识
+     */
+    public Lock tryLock(String key, int duration) {
+        return lock("trylock", key, duration);
+    }
+
+    public Lock lock(String key, int duration) {
+        return lock("lock", key, duration);
+    }
+
+    /**
+     * @param cmd      lock|trylock
+     * @param key      加锁 key
+     * @param duration 锁定时长
+     * @return
+     */
+    private Lock lock(String cmd, String key, int duration) {
+        String uuid = UUID.randomUUID().toString().replaceAll("-", "");
+        Lock lock = new Lock(key, uuid, duration, this);
+        lockTag.put(uuid, lock);
+
+        try {
+            // c.send("lock", key, uuid, strconv.Itoa(duration))
+            send(cmd, key, uuid, String.valueOf(duration));
+            synchronized (lock) {
+                lock.wait();
+            }
+        } catch (InterruptedException e) {
+            e.printStackTrace();
+        }
+        return lock;
+    }
+
+    // ================================================== timer ==================================================
+    private final ConcurrentHashMap<String, Timer> timerMap = new ConcurrentHashMap();
+
+    class Timer {
+        String name;
+        //String expr;
+        Runnable runnable;
+        //boolean single;
+
+        public String getName() {
+            return name;
+        }
+
+
+        public Runnable getRunnable() {
+            return runnable;
+        }
+
+        public Timer(String name, Runnable runnable) {
+            this.name = name;
+            this.runnable = runnable;
+        }
+    }
+
+    public void timer(String name, Runnable run) {
+        timerMap.put(name, new Timer(name, run));
+        send("timer", name);
+    }
+
+    public void reloadTimer() {
+        send("cmd", "reload-timer");
+    }
+
+    // ================================================== rpc ==================================================
+    // -- 调用端 --
+    private static final Map<String, Rpc> rpcMap = new ConcurrentHashMap<>();
+    private static final Map<String, TypeToken> rpcRetType = new ConcurrentHashMap<>();
+
+    // rpc call
+    public RpcResult<Void> rpc(String topic, Object v) {
+        return rpc(topic, v, null);
+    }
+
+    // rpc call
+    public <T, R> RpcResult<R> rpc(String topic, T v, TypeToken<R> typeToken) {
+        return rpc(topic, v, typeToken, 0);
+    }
+
+    // rpc call
+    public <T, R> RpcResult<R> rpc(String topic, T v, TypeToken<R> typeToken, long timeout) {
+        Rpc rpc = new Rpc<>(appid, UUID.randomUUID().toString().replaceAll("-", ""), topic, v);
+        String ruk = rpc.getRuk();
+        rpcMap.put(ruk, rpc);
+        if (typeToken != null) {
+            rpcRetType.put(ruk, typeToken);
+        }
+        try {
+            publish(topic, rpc); // send("rpc", topic, toStr(rpc));
+            synchronized (rpc) {
+                if (timeout <= 0) {
+                    timeout = 1000 * 15;
+                }
+                // call timeout default: 15s
+                Timers.delay(() -> {
+                    synchronized (rpc) {
+                        Rpc rpc1 = rpcMap.get(ruk);
+                        if (rpc1 == null) {
+                            return;
+                        }
+
+                        RpcResult rpcResult = rpc.retError(505, "请求超时");
+                        rpc.setRpcResult(rpcResult);
+                        logger.warning("rpc timeout: " + gson.toJson(rpc));
+                        rpc.notify();
+                    }
+                }, timeout);
+
+                rpc.wait();
+                rpcMap.remove(ruk);
+            }
+        } catch (InterruptedException e) {
+            e.printStackTrace();
+            // call error
+            RpcResult rpcResult = rpc.retError(501, "请求失败");
+            rpc.setRpcResult(rpcResult);
+        }
+        return rpc.getRpcResult();
+    }
+
+    public <T, R> CompletableFuture<RpcResult<R>> rpcAsync(String topic, T v) {
+        return CompletableFuture.supplyAsync(() -> rpc(topic, v, null));
+    }
+
+    public <T, R> CompletableFuture<RpcResult<R>> rpcAsync(String topic, T v, TypeToken<R> typeToken) {
+        return CompletableFuture.supplyAsync(() -> rpc(topic, v, typeToken));
+    }
+
+    public <T, R> CompletableFuture<RpcResult<R>> rpcAsync(String topic, T v, long timeout) {
+        return CompletableFuture.supplyAsync(() -> rpc(topic, v, null, timeout));
+    }
+
+    public <T, R> CompletableFuture<RpcResult<R>> rpcAsync(String topic, T v, TypeToken<R> typeToken, long timeout) {
+        return CompletableFuture.supplyAsync(() -> rpc(topic, v, typeToken, timeout));
+    }
+
+    // RpcResult: {ruk:xxx-xxxx, retcode:0}
+    // rpc call back consumer
+    private void rpcAccept(String value) {
+        RpcResult resp = gson.fromJson(value, new TypeToken<RpcResult<String>>() {
+        }.getType());
+
+        String ruk = resp.getRuk();
+        Rpc rpc = rpcMap.remove(ruk);
+        if (rpc == null) {
+            return;
+        }
+        TypeToken typeToken = rpcRetType.get(ruk);
+
+        Object result = resp.getResult();
+        if (result != null && typeToken != null && !"java.lang.String".equals(typeToken.getType().getTypeName())) {
+            result = gson.fromJson((String) resp.getResult(), typeToken.getType());
+        }
+
+        resp.setResult(result);
+        rpc.setRpcResult(resp);
+        synchronized (rpc) {
+            rpc.notify();
+        }
+    }
+
+    // -- 订阅端 --
+    private final HashSet rpcTopics = new HashSet();
+
+    // rpc call consumer
+    public <T, R> void rpcSubscribe(String topic, TypeToken<T> typeToken, Function<Rpc<T>, RpcResult<R>> fun) {
+        Consumer<String> consumer = v -> {
+            Rpc<T> rpc = null;
+            try {
+                rpc = gson.fromJson(v, new TypeToken<Rpc<String>>() {
+                }.getType());
+
+                // 参数转换
+                T paras = gson.fromJson((String) rpc.getValue(), typeToken.getType());
+                rpc.setValue(paras);
+                RpcResult result = fun.apply(rpc);
+                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("服务调用失败！"));
+            }
+            // back
+        };
+
+        rpcTopics.add(topic);
+        subscribe(topic, consumer);
+    }
+}

src/main/resources/META-INF/services/org.redkale.cluster.ClusterAgentProvider

@@ -1 +0,0 @@
-dev.zhub.ZhubProvider

src/main/resources/META-INF/services/org.redkale.source.CacheSourceProvider

@@ -1 +0,0 @@
-org.redkalex.cache.redis.RedisCacheSourceProvider

src/main/resources/application.yml

@@ -0,0 +1,7 @@
+
+# zhub 配置
+zhub:
+  appid: local_api
+  addr: 127.0.0.1:1216
+  groupid: hub-api
+  auth: token-12345

test/HelloService.java

@@ -0,0 +1,108 @@
+import org.junit.Before;
+import org.junit.Test;
+import tccn.IType;
+import tccn.zhub.Lock;
+import tccn.zhub.ZHubClient;
+
+// @RestService(automapping = true)
+public class HelloService {
+
+    // @Resource(name = "zhub")
+    private ZHubClient zhub;
+
+    @Before
+    public void init() {
+
+
+
+        //zhub = new ZHubClient("127.0.0.1:1216", "g-dev", "DEV-LOCAL", "zchd@123456");
+        zhub = new ZHubClient("127.0.0.1:1216", "g-dev", "DEV-LOCAL", "token-12345");
+
+        zhub.subscribe("tv:test", x -> {
+            System.out.println(x);
+        });
+
+        Lock lock = zhub.tryLock("lock-a", 5);
+        System.out.println("lock-1: " + lock.success());
+
+        //zhub.init(Kv.of("host", "47.111.150.118", "port", "6066", "groupid", "g-dev", "appname", "DEV-LOCAL"));
+
+        // Function<Rpc<T>, RpcResult<R>> fun
+        /*zhub.rpcSubscribe("x", new TypeToken<String>() {
+        }, r -> {
+            return r.buildResp(Map.of("v", r.getValue().toUpperCase() + ": Ok"));
+        });
+
+        zhub.subscribe("sport:reqtime", x -> {
+            //System.out.println(x);
+        });
+        zhub.subscribe("abx", x -> {
+            System.out.println(x);
+        });
+
+        try {
+            Thread.sleep(010);
+        } catch (InterruptedException e) {
+            e.printStackTrace();
+        }
+
+        zhub.delay("sport:reqtime", "别✈人家的✦女娃子❤🤞🏻", 0);
+        zhub.delay("sport:reqtime", "别人家的女娃子➾🤞🏻", 0);
+        zhub.delay("sport:reqtime", "❤别人家✉<E5AEB6>的女娃子❤🤞🏻", 0);*/
+        /*zhub.delay("sport:reqtime", "中文特殊符号：『』 ＄ ￡ ♀ ‖ 「」\n" +
+                "英文：# + = & ﹉ .. ^ \"\" ·{ } % – ' €\n" +
+                "数学：＋× ＝ － ° ± ＜ ＞ ℃ ㎡ ∑ ≥ ∫ ㏄ ⊥ ≯ ∠ ∴ ∈ ∧ ∵ ≮ ∪ ㎝ ㏑ ≌ ㎞ № § ℉ ÷ ％ ‰ ㎎ ㎏ ㎜ ㏒ ⊙ ∮ ∝ ∞ º ¹ ² ³ ½ ¾ ¼ ≈ ≡ ≠ ≤ ≦ ≧ ∽ ∷ ／ ∨ ∏ ∩ ⌒ √Ψ ¤ ‖ ¶\n" +
+                "特殊：♤ ♧ ♡ ♢ ♪ ♬ ♭ ✔ ✘ ♞ ♟ ↪ ↣ ♚ ♛ ♝ ☞ ☜ ⇔ ☆ ★ □ ■ ○ ● △ ▲ ▽ ▼ ◇ ◆ ♀ ♂ ※ ↓ ↑ ↔ ↖ ↙ ↗ ↘ ← → ♣ ♠ ♥ ◎ ◣ ◢ ◤ ◥ 卍 ℡ ⊙ ㊣ ® © ™ ㈱ 囍\n" +
+                "序号：①②③④⑤⑥⑦⑧⑨⑩㈠㈡㈢㈣㈤㈥㈦㈧㈨㈩⑴ ⑵ ⑶ ⑷ ⑸ ⑹ ⑺ ⑻ ⑼ ⑽ ⒈ ⒉ ⒊ ⒋ ⒌ ⒍ ⒎ ⒏ ⒐ ⒑ Ⅰ Ⅱ Ⅲ Ⅳ Ⅴ Ⅵ Ⅶ Ⅷ ⅨⅩ\n" +
+                "日文：アイウエオァィゥェォカキクケコガギグゲゴサシスセソザジズゼゾタチツテトダヂヅデドッナニヌネノハヒフヘホバビブベボパピプペポマミムメモャヤュユョラリヨルレロワヰヱヲンヴヵヶヽヾ゛゜ー、。「「あいうえおぁぃぅぇぉかきくけこがぎぐげごさしすせそざじずぜぞたちつてでどっなにぬねのはひふへ」」ほばびぶべぼぱぴぷぺぽまみむめもやゆよゃゅょらりるれろわをんゎ゛゜ー、。「」\n" +
+                "部首：犭 凵 巛 冖 氵 廴 讠 亻 钅 宀 亠 忄 辶 弋 饣 刂 阝 冫 卩 疒 艹 疋 豸 冂 匸 扌 丬 屮衤 礻 勹 彳 彡", 0);*/
+
+    }
+
+    @Test
+    public void rpcTest() {
+        //RpcResult<String> rpc = zhub.rpc("wx:users", Map.of("appId", "wxa554ec3ab3bf1fc7"), IConsumer.TYPE_TOKEN_STRING);
+        //RpcResult<String> rpc = zhub.rpc("a", "fa", IConsumer.TYPE_TOKEN_STRING);
+        zhub.publish("tv:test", "hello ym!");
+
+        zhub.subscribe("tv:abx", x -> {
+            System.out.println(x);
+        });
+
+        zhub.rpcSubscribe("rpc-x", IType.STRING, x -> {
+            return x.render(x.getValue().toUpperCase());
+        });
+
+        Lock lock = zhub.tryLock("lock-a", 5);
+        System.out.println("lock-2: " + lock.success());
+
+        try {
+            Thread.sleep(5 * 1000);
+        } catch (InterruptedException e) {
+            throw new RuntimeException(e);
+        }
+
+        Lock lock2 = zhub.tryLock("lock-a", 5);
+        System.out.println("lock-3: " + lock2.success());
+        /*try {
+            Thread.sleep(100);
+        } catch (InterruptedException e) {
+            throw new RuntimeException(e);
+        }*/
+        Lock lock3 = zhub.tryLock("lock-a", 5);
+        System.out.println("lock-4: " + lock3.success());
+
+        try {
+            Thread.sleep(3000 * 30000);
+        } catch (InterruptedException e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    public void lockTest() {
+
+    }
+
+    /*RpcResult<FileToken> x = zhub.rpc("rpc:file:up-token", Map.of(), new TypeToken<>() {
+            });*/
+}

test/net.tccn/mq/AppTest.java

@@ -1,412 +0,0 @@
-package net.tccn.mq;
-
-import net.tccn.Event;
-import dev.zhub.timer.Timers;
-import org.junit.Test;
-import org.redkale.convert.json.JsonConvert;
-
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Map;
-import java.util.concurrent.DelayQueue;
-import java.util.concurrent.LinkedBlockingQueue;
-import java.util.concurrent.TimeUnit;
-import java.util.function.Function;
-import java.util.logging.Logger;
-import java.util.regex.Matcher;
-import java.util.regex.Pattern;
-
-/**
- * 消息发布订阅测试
- */
-public class AppTest {
-    Logger logger = Logger.getLogger("");
-
-    @Test
-    public void runConsumer() {
-        try {
-            // String str = ", response = {\"success\":true,\"retcode\":0,\"result\":{\"age\":0,\"explevel\":1,\"face\":\"https://aimg.woaihaoyouxi.com/haogame/202106/pic/20210629095545FmGt-v9NYqyNZ_Q6_y3zM_RMrDgd.jpg\",\"followed\":0,\"gender\":0,\"idenstatus\":0,\"matchcatelist\":[{\"catename\":\"足球\",\"catepic\":\"https://aimg.woaihaoyouxi.com/haogame/202107/pic/20210714103556FoG5ICf_7BFx6Idyo3TYpJQ7tmfG.png\",\"matchcateid\":1},{\"catename\":\"篮球\",\"catepic\":\"https://aimg.woaihaoyouxi.com/haogame/202107/pic/20210714103636FklsXTn1f6Jlsam8Jk-yFB7Upo3C.png\",\"matchcateid\":2}],\"matchcates\":\"2,1\",\"mobile\":\"18515190967\",\"regtime\":1624931714781,\"sessionid\":\"d1fc447753bd4700ad29674a753030fa\",\"status\":10,\"userid\":100463,\"username\":\"绝尘\",\"userno\":100463}}";
-            String str = "hello你好";
-
-            System.out.println(str.length());
-
-            /*consumer.timer("a", () -> {
-                System.out.println(Utility.now() + " timer a 执行了");
-                try {
-                    Thread.sleep(3000);
-                } catch (InterruptedException e) {
-                    e.printStackTrace();
-                }
-            });
-
-            consumer.timer("b", () -> {
-                System.out.println(Utility.now() + " ----------------- timer b 执行了");
-            });
-            //consumer.delay("a", "1", 200);
-            consumer.delay("a", "1", "2000");*/
-
-            /*Consumer<String> con = x -> {
-                logger.info("--->开始申请锁:" + System.currentTimeMillis());
-                Lock lock = consumer.tryLock("a", 20);
-                logger.info("===>成功申请锁:" + System.currentTimeMillis());
-                for (int i = 0; i < 20; i++) {
-                    try {
-                        Thread.sleep(500);
-                    } catch (InterruptedException e) {
-                        e.printStackTrace();
-                    }
-                    System.out.println(x + ":" + i);
-                }
-                lock.unLock();
-            };
-
-            new Thread(() -> con.accept("x")).start();
-            new Thread(() -> con.accept("y")).start();
-            new Thread(() -> con.accept("z")).start();*/
-
-
-            Thread.sleep(60_000 * 60);
-        } catch (Exception e) {
-            e.printStackTrace();
-        }
-    }
-
-    @Test
-    public void runProducer() {
-
-    }
-
-    private static LinkedBlockingQueue<String> queue = new LinkedBlockingQueue();
-
-    @Test
-    public void t() {
-        List<String> list = new ArrayList<>();
-        list.toArray(String[]::new);
-
-        new Thread(() -> {
-            while (true) {
-                System.out.println("accept:");
-                String peek = null;
-                try {
-                    System.out.println(!queue.isEmpty());
-                    peek = queue.poll(100, TimeUnit.MILLISECONDS);
-
-                } catch (InterruptedException e) {
-                    e.printStackTrace();
-                }
-                System.out.println(peek);
-            }
-        }).start();
-
-        for (int i = 0; i < 10; i++) {
-            try {
-                queue.put(i + "");
-                Thread.sleep(1000);
-            } catch (InterruptedException e) {
-                e.printStackTrace();
-            }
-        }
-
-
-        try {
-            System.out.println("---");
-            Thread.sleep(1000 * 5);
-        } catch (InterruptedException e) {
-            e.printStackTrace();
-        }
-    }
-
-    @Test
-    public void xx() {
-        Function<String, String> fun = x -> {
-            return x.toUpperCase();
-        };
-
-        System.out.println(fun.toString());
-    }
-
-
-    // (27+5*23)/(63-59)
-    // [27+5*23] [/] [63-59]
-    // [27] + [5*23] [/] [63-59]
-
-
-    /**
-     * 1. 按照优先级逐一拆解运算
-     * 括号, 乘除，加减
-     * 2. 逐一进行计算
-     */
-    class C {
-        C A;
-        C B;
-
-        int c1; // 如果 A 只剩数字，将c1 转为整数存放到 c1
-        int c2;
-        String x; // + - * /
-    }
-
-    @Test
-    public void x() {
-        // (27+5*23)/(63-59)
-
-        String str = "(27+5*23)/(63-59)";
-        str = "27+5*23";
-        str = "258/((35+17)/(5*3+18-29)+3)(138-134)*41-6+10+24";
-        str = "5*3+18-29";
-
-        //System.out.println("258/((35+17)/(5*3+18-29)+3)(138-134)*41-6+10+24".replaceAll("\\)\\(", ")*("));
-        List<String> parse = parse(str.replaceAll("\\)\\(", ")\\*("));
-
-
-        System.out.println(c(str));
-    }
-
-    // 因式分解：括号 -> 乘除 -> 加减
-    public List<String> parse(String str) {
-        // 找到一对
-        /*
-
-        258/()()
-        [258, /, (35+17)/(5*3+18-29)+3, 138-134, *, 41, -, 6, +, 10, +, 24]
-
-         */
-        // 258/((35+17)/(5*3+18-29)+3)(138-134)*41-6+10+24
-        //str = "258    /   (35+17)/(5*3+18-29)+3   138-134,   *   41-6+10+24";
-        String[] strArr = str.split("");
-
-        // 一级括号、加、减、乘、除分解
-        List<String> arr = new ArrayList<>();
-        String tmp = "";
-        int n1 = 0; // 括号层级开始
-        int m1 = 0; // 括号层级结尾
-        for (String s : strArr) {
-            if (n1 > 0) { // 一级括号分解
-                if (")".equals(s) && (++m1) == n1) { // 一级括号结束
-                    arr.add(tmp);
-                    tmp = "";
-                    n1 = 0;
-                    m1 = 0;
-                } else {
-                    if ("(".equals(s)) {
-                        n1++;
-                    }
-                    tmp += s;
-                }
-            } else { // 无括号
-                if ("+".equals(s) || "-".equals(s) || "*".equals(s) || "/".equals(s)) {
-                    if (!"".equals(tmp)) {
-                        arr.add(tmp);
-                    }
-                    arr.add(s);
-                    tmp = "";
-                } else if ("(".equals(s)) {
-                    n1 = 1;
-                } else {
-                    tmp += s;
-                }
-            }
-        }
-        if (!"".equals(tmp)) {
-            arr.add(tmp);
-        }
-        return arr;
-    }
-
-    public int c(String str) {
-        List<String> arr = parse(str); // 预期 length >= 3 基数的基数，如：[27+5*23, /, 63-59], [60, /, 2, *, 164-23*7]
-        System.out.println(arr);
-        if (arr == null || arr.size() < 3) {
-            return -1; // 错误码-1：错误的计算式
-        }
-
-        List<String> _arr = new ArrayList<>();
-        // 按照优先级做合并计算 乘除优先
-
-        // 乘除
-        for (int i = 1; i < arr.size() - 1; i += 2) {
-            /*if ("*".equals(arr.get(i)) || "/".equals(arr.get(i))) {
-                if (_arr.size() > 0) {
-                    _arr.remove(_arr.size() - 1);
-                }
-                int c = c(arr.get(i - 1), arr.get(i + 1), arr.get(i));
-                if (c < 0) {
-                    return -1;
-                }
-                _arr.add(c + "");
-            } else {
-                _arr.add(arr.get(i - 1));
-                _arr.add(arr.get(i));
-                _arr.add(arr.get(i + 1));
-            }*/
-
-            if ("*".equals(arr.get(i)) || "/".equals(arr.get(i))) {
-                int c = c(arr.get(i - 1), arr.get(i + 1), arr.get(i));
-                if (c < 0) {
-                    return c;
-                }
-                _arr.add(c + "");
-            } else {
-
-
-            }
-        }
-
-        if (_arr.size() == 1) { // 通过第一轮的 乘除计算，完成结果合并
-            return Integer.parseInt(_arr.get(0));
-        }
-        int c = 0;
-        for (int i = 1; i < _arr.size(); i += 2) {
-            int _c = c(_arr.get(i - 1), _arr.get(i + 1), _arr.get(i));
-            if (_c < 0) {
-                return _c;
-            }
-            c += _c;
-        }
-        return c;
-    }
-
-    public int c(String a, String b, String x) {
-        int _a = 0;
-        if (a.contains("(") || a.contains("+") || a.contains("-") || a.contains("*") || a.contains("/")) {
-            _a = c(a);
-        } else { // 预期 无 ( + - * / 的结果为标准数字
-            _a = Integer.parseInt(a);
-        }
-        int _b = 0;
-        if (b.contains("(") || b.contains("+") || b.contains("-") || b.contains("*") || b.contains("/")) {
-            _b = c(b);
-        } else { // 预期 无 ( + - * / 的结果为标准数字
-            _b = Integer.parseInt(b);
-        }
-
-        // 如果出现负数（错误码）直接返回对应的负数
-        if (_a < 0) {
-            return _a;
-        }
-        if (_b < 0) {
-            return _b;
-        }
-
-        // 定义错误标识: -1错误的计算式，-2除不尽，-3除数为0，-4大于200,
-        if ("+".equals(x)) {
-            return _a + _b;
-        } else if ("-".equals(x)) {
-            return _a - _b;
-        } else if ("*".equals(x)) {
-            return _a * _b;
-        } else if ("/".equals(x)) {
-            return _a % _b > 0 ? -2 : _a / _b; // 除不尽
-        }
-        return 0;
-    }
-
-    @Test
-    public void testxx() {
-        Event of = Event.of("A", Map.of("b", 1));
-
-        System.out.println(JsonConvert.root().convertTo(of));
-
-        String str = "❦别人家的女娃子🤞🏻ꚢ";
-
-        /*
-        System.out.println("别人家的女娃子🤞🏻".length());*/
-        System.out.println(strLength(str));
-        System.out.println(getWordCount(str));
-        /*try {
-            System.out.println("别人家的女娃子🤞🏻".getBytes("UTF-8").length);
-        } catch (UnsupportedEncodingException e) {
-            e.printStackTrace();
-        }
-        System.out.println("系统默认编码方式：" + System.getProperty("file.encoding"));*/
-    }
-
-
-    public static int strLength(String value) {
-        int valueLength = 0;
-        String chinese = "[\u4e00-\u9fa5]";
-        for (int i = 0; i < value.length(); i++) {
-            String temp = value.substring(i, i + 1);
-            if (temp.matches(chinese)) {
-                valueLength += 2;
-            } else {
-                valueLength += 1;
-            }
-        }
-        return valueLength;
-    }
-
-    public int getWordCount(String str) {
-        str = str.replaceAll("[^\\x00-\\xff]", "*");
-        return str.length();
-    }
-
-    @Test
-    public void delay() {
-        DelayQueue<com.zdemo.zhub.Delays> delayQueue = new DelayQueue<>();
-
-        logger.info("加入延时任务1");
-        delayQueue.add(new com.zdemo.zhub.Delays(5000, () -> {
-            logger.info("任务1 延时任务执行了!");
-        }));
-
-        try {
-            Thread.sleep(1000);
-        } catch (InterruptedException e) {
-            e.printStackTrace();
-        }
-
-        logger.info("加入延时任务2");
-        delayQueue.add(new com.zdemo.zhub.Delays(5000, () -> {
-            logger.info("任务2 延时任务执行了!");
-        }));
-
-        try {
-            while (true) {
-                com.zdemo.zhub.Delays delay = delayQueue.take();
-
-                delay.run();
-            }
-        } catch (InterruptedException e) {
-            e.printStackTrace();
-        }
-    }
-
-    @Test
-    public void regTest() {
-        // 按指定模式在字符串查找
-        String line = "This order was placed for QT3000! OK?";
-        String pattern = "(\\D*)(\\d+)(.*)";
-
-        // 创建 Pattern 对象
-        Pattern r = Pattern.compile(pattern);
-
-        // 现在创建 matcher 对象
-        Matcher m = r.matcher(line);
-        if (m.find()) {
-            System.out.println("Found value: " + m.group(0));
-            System.out.println("Found value: " + m.group(1));
-            System.out.println("Found value: " + m.group(2));
-            System.out.println("Found value: " + m.group(3));
-        } else {
-            System.out.println("NO MATCH");
-        }
-    }
-
-    @Test
-    public void timersTest() {
-        Timers.tryDelay(() -> {
-            logger.info("xx:" + System.currentTimeMillis());
-            return true;
-        }, 1000, 5);
-
-        Timers.delay(() -> {
-            System.out.println("11");
-        }, 3000);
-
-        try {
-            Thread.sleep(100000);
-        } catch (InterruptedException e) {
-            e.printStackTrace();
-        }
-    }
-}

test/net.tccn/mq/Delays.java

@@ -1,72 +0,0 @@
-package com.zdemo.zhub;
-
-import java.util.concurrent.DelayQueue;
-import java.util.concurrent.Delayed;
-import java.util.concurrent.TimeUnit;
-import java.util.function.Supplier;
-import java.util.logging.Logger;
-
-public class Delays implements Delayed, Runnable {
-    public Logger logger = Logger.getLogger(Delays.class.getSimpleName());
-
-    private long time;          // 执行时间
-    private Runnable runnable;  // 任务到时间执行 runnable
-
-    public Delays(long timeout, Runnable runnable) {
-        this.time = System.currentTimeMillis() + timeout;
-        this.runnable = runnable;
-    }
-
-    @Override
-    public long getDelay(TimeUnit unit) {
-        return unit.convert(time - System.currentTimeMillis(), TimeUnit.MILLISECONDS);
-    }
-
-    @Override
-    public int compareTo(Delayed other) {
-        if (other == this) { // compare zero ONLY if same object
-            return 0;
-        }
-        if (other instanceof Delays) {
-            Delays x = (Delays) other;
-            long diff = time - x.time;
-            if (diff < 0) {
-                return -1;
-            } else if (diff > 0) {
-                return 1;
-            }
-        }
-        long d = (getDelay(TimeUnit.NANOSECONDS) -
-                other.getDelay(TimeUnit.NANOSECONDS));
-        return (d == 0) ? 0 : ((d < 0) ? -1 : 1);
-    }
-
-    @Override
-    public void run() {
-        runnable.run();
-    }
-
-    // ===========
-    public static DelayQueue<Delays> delayQueue = new DelayQueue<>();
-
-    public static void addDelay(long timeout, Runnable runnable) {
-        delayQueue.add(new Delays(timeout, runnable));
-    }
-
-    public static void tryDelay(Supplier<Boolean> supplier, long delayMillis, int maxCount) {
-
-    }
-
-    static {
-        new Thread(() -> {
-            try {
-                while (true) {
-                    Delays delay = delayQueue.take();
-                    delay.run(); //异常会导致延时队列失败
-                }
-            } catch (InterruptedException e) {
-                e.printStackTrace();
-            }
-        }).start();
-    }
-}

test/net.tccn/mq/HelloService.java

@@ -1,244 +0,0 @@
-package net.tccn.mq;
-
-import net.tccn.IType;
-import net.tccn.zhub.RpcResult;
-import net.tccn.zhub.ZHubClient;
-import org.redkale.net.http.RestMapping;
-import org.redkale.net.http.RestService;
-import org.redkale.service.Service;
-import org.redkale.util.AnyValue;
-import org.redkale.util.TypeToken;
-
-import javax.annotation.Resource;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Map;
-import java.util.concurrent.CompletableFuture;
-
-@RestService(automapping = true, name = "hello")
-public class HelloService implements Service {
-
-    @Resource(name = "hub")
-    private ZHubClient zhub;
-
-    /*@Resource(name = "hubx")
-    private ZHubClient zhubx;
-
-    @Resource(name = "vvvvhub2")
-    private ZHubClient zhub2;*/
-
-    //private dev.zhub.client.ZHubClient zhubx = null;
-
-
-    @Override
-    public void init(AnyValue config) {
-
-        /*CompletableFuture.runAsync(() -> {
-            zhubx = new dev.zhub.client.ZHubClient("127.0.0.1", 1216, "g-dev", "DEV-LOCAL");
-            //zhubx = new dev.zhub.client.ZHubClient("47.111.150.118", 6066, "g-dev", "DEV-LOCAL");
-        });*/
-
-        // Function<Rpc<T>, RpcResult<R>> fun
-        /*zhub.rpcSubscribe("x", new TypeToken<String>() {
-        }, r -> {
-
-            return r.buildResp(Map.of("v", r.getValue().toUpperCase() + ": Ok"));
-        });*/
-        zhub.rpcSubscribe("y", new TypeToken<String>() {
-        }, r -> {
-
-            return r.render(Map.of("v", r.getValue().toUpperCase() + ": Ok"));
-        });
-
-        zhub.subscribe("sport:reqtime", x -> {
-            System.out.println(x);
-        });
-        zhub.subscribe("abx1", x -> {
-            System.out.println(x);
-        });
-
-        try {
-            Thread.sleep(010);
-        } catch (InterruptedException e) {
-            e.printStackTrace();
-        }
-
-        /*zhub.delay("sport:reqtime", "别✈人家的✦女娃子❤🤞🏻", 0);
-        zhub.delay("sport:reqtime", "别人家的女娃子➾🤞🏻", 0);
-        zhub.delay("sport:reqtime", "❤别人家✉<E5AEB6>的女娃子❤🤞🏻", 0);
-        zhub.delay("sport:reqtime", "中文特殊符号：『』 ＄ ￡ ♀ ‖ 「」\n" +
-                "英文：# + = & ﹉ .. ^ \"\" ·{ } % – ' €\n" +
-                "数学：＋× ＝ － ° ± ＜ ＞ ℃ ㎡ ∑ ≥ ∫ ㏄ ⊥ ≯ ∠ ∴ ∈ ∧ ∵ ≮ ∪ ㎝ ㏑ ≌ ㎞ № § ℉ ÷ ％ ‰ ㎎ ㎏ ㎜ ㏒ ⊙ ∮ ∝ ∞ º ¹ ² ³ ½ ¾ ¼ ≈ ≡ ≠ ≤ ≦ ≧ ∽ ∷ ／ ∨ ∏ ∩ ⌒ √Ψ ¤ ‖ ¶\n" +
-                "特殊：♤ ♧ ♡ ♢ ♪ ♬ ♭ ✔ ✘ ♞ ♟ ↪ ↣ ♚ ♛ ♝ ☞ ☜ ⇔ ☆ ★ □ ■ ○ ● △ ▲ ▽ ▼ ◇ ◆ ♀ ♂ ※ ↓ ↑ ↔ ↖ ↙ ↗ ↘ ← → ♣ ♠ ♥ ◎ ◣ ◢ ◤ ◥ 卍 ℡ ⊙ ㊣ ® © ™ ㈱ 囍\n" +
-                "序号：①②③④⑤⑥⑦⑧⑨⑩㈠㈡㈢㈣㈤㈥㈦㈧㈨㈩⑴ ⑵ ⑶ ⑷ ⑸ ⑹ ⑺ ⑻ ⑼ ⑽ ⒈ ⒉ ⒊ ⒋ ⒌ ⒍ ⒎ ⒏ ⒐ ⒑ Ⅰ Ⅱ Ⅲ Ⅳ Ⅴ Ⅵ Ⅶ Ⅷ ⅨⅩ\n" +
-                "日文：アイウエオァィゥェォカキクケコガギグゲゴサシスセソザジズゼゾタチツテトダヂヅデドッナニヌネノハヒフヘホバビブベボパピプペポマミムメモャヤュユョラリヨルレロワヰヱヲンヴヵヶヽヾ゛゜ー、。「「あいうえおぁぃぅぇぉかきくけこがぎぐげごさしすせそざじずぜぞたちつてでどっなにぬねのはひふへ」」ほばびぶべぼぱぴぷぺぽまみむめもやゆよゃゅょらりるれろわをんゎ゛゜ー、。「」\n" +
-                "部首：犭 凵 巛 冖 氵 廴 讠 亻 钅 宀 亠 忄 辶 弋 饣 刂 阝 冫 卩 疒 艹 疋 豸 冂 匸 扌 丬 屮衤 礻 勹 彳 彡", 0);
-*/
-    }
-
-    @RestMapping
-    public RpcResult x(String v) {
-        if (v == null) {
-            v = "";
-        }
-
-        List<CompletableFuture> list = new ArrayList();
-
-        for (int i = 0; i < 100; i++) {
-            long start = System.currentTimeMillis();
-            /*RpcResult<FileToken> x = zhub.rpc("rpc:file:up-token", Map.of(), new TypeToken<>() {
-            });*/
-
-            /*list.add(zhub.rpcAsync("x", v + i, new TypeToken<>() {
-            }));*/
-            zhub.publish("x", v + i);
-
-            System.out.println("time: " + (System.currentTimeMillis() - start) + " ms");
-            //System.out.println(x.getResult().get("v"));
-        }
-
-        return zhub.rpc("x", v, IType.STRING);
-    }
-
-    @RestMapping
-    public RpcResult<String> d(String v) {
-        RpcResult<String> rpc = zhub.rpc("x", v, IType.STRING);
-        return rpc;
-    }
-
-    @RestMapping
-    public String y(String v) {
-        if (v == null) {
-            v = "";
-        }
-
-        for (int i = 0; i < 100; i++) {
-            long start = System.currentTimeMillis();
-            /*RpcResult<FileToken> x = zhub.rpc("rpc:file:up-token", Map.of(), new TypeToken<>() {
-            });*/
-
-            /*RpcResult<Object> x = zhubx.rpc("y", v + i, new com.google.gson.reflect.TypeToken<>() {
-            });*/
-
-            System.out.println("time: " + (System.currentTimeMillis() - start) + " ms");
-
-            //System.out.println(x.getResult());
-        }
-
-        return "ok";
-    }
-
-    @RestMapping(name = "send")
-    public String send() {
-        zhub.publish("abx1", 1);
-        return "ok";
-    }
-
-
-    public static void main(String[] args) {
-        // "\"别人家的女娃子\uD83E\uDD1E\uD83C\uDFFB\""
-        /*String s = "别人家的女娃子\uD83E\uDD1E\uD83C\uDFFB";
-        System.out.println("别人家的女娃子🤞🏻".length());
-
-        byte[] bytes = "别人家的女娃子🤞🏻".getBytes();
-        System.out.println(bytes.length);
-
-        System.out.println(unicodeToUtf8(s));
-        System.out.println(utf8ToUnicode("别人家的女娃子🤞🏻"));
-        */
-
-        //ExecutorService pool = Executors.newFixedThreadPool(5);
-
-
-    }
-
-
-    public static String utf8ToUnicode(String inStr) {
-        char[] myBuffer = inStr.toCharArray();
-
-        StringBuffer sb = new StringBuffer();
-        for (int i = 0; i < inStr.length(); i++) {
-            Character.UnicodeBlock ub = Character.UnicodeBlock.of(myBuffer[i]);
-            if (ub == Character.UnicodeBlock.BASIC_LATIN) {
-                //英文及数字等
-                sb.append(myBuffer[i]);
-            } else if (ub == Character.UnicodeBlock.HALFWIDTH_AND_FULLWIDTH_FORMS) {
-                //全角半角字符
-                int j = (int) myBuffer[i] - 65248;
-                sb.append((char) j);
-            } else {
-                //汉字
-                short s = (short) myBuffer[i];
-                String hexS = Integer.toHexString(s);
-                String unicode = "\\u" + hexS;
-                sb.append(unicode.toLowerCase());
-            }
-        }
-        return sb.toString();
-    }
-
-    public static String unicodeToUtf8(String theString) {
-        char aChar;
-        int len = theString.length();
-        StringBuffer outBuffer = new StringBuffer(len);
-        for (int x = 0; x < len; ) {
-            aChar = theString.charAt(x++);
-            if (aChar == '\\') {
-                aChar = theString.charAt(x++);
-                if (aChar == 'u') {
-                    // Read the xxxx
-                    int value = 0;
-                    for (int i = 0; i < 4; i++) {
-                        aChar = theString.charAt(x++);
-                        switch (aChar) {
-                            case '0':
-                            case '1':
-                            case '2':
-                            case '3':
-                            case '4':
-                            case '5':
-                            case '6':
-                            case '7':
-                            case '8':
-                            case '9':
-                                value = (value << 4) + aChar - '0';
-                                break;
-                            case 'a':
-                            case 'b':
-                            case 'c':
-                            case 'd':
-                            case 'e':
-                            case 'f':
-                                value = (value << 4) + 10 + aChar - 'a';
-                                break;
-                            case 'A':
-                            case 'B':
-                            case 'C':
-                            case 'D':
-                            case 'E':
-                            case 'F':
-                                value = (value << 4) + 10 + aChar - 'A';
-                                break;
-                            default:
-                                throw new IllegalArgumentException(
-                                        "Malformed   \\uxxxx   encoding.");
-                        }
-                    }
-                    outBuffer.append((char) value);
-                } else {
-                    if (aChar == 't')
-                        aChar = '\t';
-                    else if (aChar == 'r')
-                        aChar = '\r';
-                    else if (aChar == 'n')
-                        aChar = '\n';
-                    else if (aChar == 'f')
-                        aChar = '\f';
-                    outBuffer.append(aChar);
-                }
-            } else
-                outBuffer.append(aChar);
-        }
-        return outBuffer.toString();
-    }
-}