lxy/zhub-client
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

新增：zhub-cli 独立分支

Browse Source
This commit is contained in:
梁显优 2021-12-31 11:53:25 +08:00
parent 6995d103f0
commit 3debcee7d2
27 changed files with 146 additions and 8680 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

24
pom.xml Normal file
View File

@ -0,0 +1,24 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
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>net.tccn</groupId>
<artifactId>zhub-client</artifactId>
<version>1.0-SNAPSHOT</version>
<properties>
<maven.compiler.source>17</maven.compiler.source>
<maven.compiler.target>17</maven.compiler.target>
</properties>
<dependencies>
<dependency>
<groupId>com.google.code.gson</groupId>
<artifactId>gson</artifactId>
<version>2.8.8</version>
</dependency>
</dependencies>
</project>

18
src/com/zdemo/AbstractConsumer.java
View File

@ -1,12 +1,12 @@
package com.zdemo;
import org.redkale.convert.json.JsonConvert;
import org.redkale.util.TypeToken;
import com.google.gson.Gson;
import com.google.gson.reflect.TypeToken;
import com.zdemo.zhub.Rpc;
import javax.annotation.Resource;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.Consumer;
/**
@ -15,12 +15,12 @@ import java.util.function.Consumer;
*/
public abstract class AbstractConsumer implements IConsumer {
protected JsonConvert convert = JsonConvert.root();
public Gson gson = Rpc.gson;
@Resource(name = "APP_NAME")
// @Resource(name = "APP_NAME")
protected String APP_NAME = "";
private Map<String, EventType> eventMap = new ConcurrentHashMap<>();
private Map<String, EventType> eventMap = new HashMap<>();
protected abstract String getGroupid();
@ -43,7 +43,7 @@ public abstract class AbstractConsumer implements IConsumer {
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);
@ -61,7 +61,7 @@ public abstract class AbstractConsumer implements IConsumer {
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

3
src/com/zdemo/EventType.java
View File

@ -1,6 +1,7 @@
package com.zdemo;
import org.redkale.util.TypeToken;
import com.google.gson.reflect.TypeToken;
import java.util.function.Consumer;

6
src/com/zdemo/IConsumer.java
View File

@ -1,10 +1,14 @@
package com.zdemo;
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>() {
};
/**
* 取消订阅

49
src/com/zdemo/ZhubListener.java
View File

@ -1,49 +0,0 @@
package com.zdemo;
import org.redkale.boot.Application;
import org.redkale.boot.ApplicationListener;
import org.redkale.service.Service;
import org.redkale.util.AnyValue;
import org.redkale.util.RedkaleClassLoader;
import org.redkale.util.ResourceFactory;
import java.lang.reflect.InvocationTargetException;
import java.util.concurrent.CompletableFuture;
/**
* 服务监听
*
* @author: liangxy.
*/
public class ZhubListener implements ApplicationListener {
@Override
public void preStart(Application application) {
CompletableFuture.runAsync(() -> {
ResourceFactory resourceFactory = application.getResourceFactory();
RedkaleClassLoader classLoader = application.getClassLoader();
AnyValue appConfig = application.getAppConfig();
AnyValue zhubs = appConfig.getAnyValue("zhubs");
AnyValue[] values = zhubs.getAnyValues("zhub");
for (AnyValue zhub : values) {
String className = zhub.getValue("value", "com.zdemo.zhub.ZHubClient");
try {
Class<?> clazz = classLoader.loadClass(className);
Service obj = (Service) clazz.getDeclaredConstructor().newInstance();
application.getResourceFactory().inject(obj);
obj.init(zhub);
resourceFactory.register(zhub.get("name"), clazz, obj);
} catch (InstantiationException | IllegalAccessException | InvocationTargetException | NoSuchMethodException | ClassNotFoundException e) {
e.printStackTrace();
}
}
});
}
@Override
public void preShutdown(Application application) {
}
}

2173
src/com/zdemo/cache_/RedisCacheSource.java
View File

File diff suppressed because it is too large Load Diff

262
src/com/zdemo/cache_/RedisTest.java
View File

@ -1,262 +0,0 @@
package com.zdemo.cache_;
import com.zdemo.cachex.MyRedisCacheSource;
import org.redkale.convert.json.JsonFactory;
import org.redkale.net.AsyncIOGroup;
import org.redkale.util.AnyValue;
import org.redkale.util.ResourceFactory;
import java.util.Map;
import static org.redkale.boot.Application.RESNAME_APP_GROUP;
public class RedisTest {
static MyRedisCacheSource<String> source;
static MyRedisCacheSource<Integer> sourceInt;
static {
AnyValue.DefaultAnyValue conf = new AnyValue.DefaultAnyValue().addValue("maxconns", "10");
conf.addValue("node", new AnyValue.DefaultAnyValue().addValue("addr", "47.111.150.118").addValue("port", "6064").addValue("password", "*Zhong9307!").addValue("db", 1));
final AsyncIOGroup asyncGroup = new AsyncIOGroup(8192, 16);
asyncGroup.start();
ResourceFactory.root().register(RESNAME_APP_GROUP, asyncGroup);
source = new MyRedisCacheSource();
ResourceFactory.root().inject(source);
source.init(null);
source.defaultConvert = JsonFactory.root().getConvert();
source.init(conf);
// int
sourceInt = new MyRedisCacheSource<Integer>();
ResourceFactory.root().inject(sourceInt);
sourceInt.init(null);
sourceInt.defaultConvert = JsonFactory.root().getConvert();
sourceInt.init(conf);
sourceInt.initValueType(Integer.class);
}
public static void main(String[] args) {
//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(1000);
for (int i = 0; i < 1000; i++) {
executor.submit(() -> {
try {
source.lock("c", 1000);
arr[0]++;
// System.out.println("Thread: " + Thread.currentThread().getName());
// Thread.sleep(10);
} catch (Exception e) {
e.printStackTrace();
} finally {
source.unlock("c");
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 测试
/*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 < 2000; 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();
}
}*/
}
}

430
src/com/zdemo/cachex/MyRedisCacheSource.java
View File

@ -1,430 +0,0 @@
package com.zdemo.cachex;
import org.redkale.convert.Convert;
import org.redkale.service.Local;
import org.redkale.source.CacheSource;
import org.redkale.util.AutoLoad;
import org.redkale.util.ResourceType;
import java.io.Serializable;
import java.lang.reflect.Type;
import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.stream.Stream;
@Local
@AutoLoad(false)
@ResourceType(CacheSource.class)
public class MyRedisCacheSource<V extends Object> extends RedisCacheSource<V> {
//--------------------- oth ------------------------------
public boolean setnx(String key, Object v) {
byte[][] bytes = Stream.of(key, v).map(x -> String.valueOf(x).getBytes(StandardCharsets.UTF_8)).toArray(byte[][]::new);
Serializable rs = send("SETNX", CacheEntryType.OBJECT, (Type) null, key, bytes).join();
return rs == null ? false : (long) rs == 1;
}
//--------------------- oth ------------------------------
//--------------------- bit ------------------------------
public boolean getBit(String key, int offset) {
byte[][] bytes = Stream.of(key, offset).map(x -> String.valueOf(x).getBytes(StandardCharsets.UTF_8)).toArray(byte[][]::new);
Serializable v = send("GETBIT", CacheEntryType.OBJECT, (Type) null, key, bytes).join();
return v == null ? false : (long) v == 1;
}
public void setBit(String key, int offset, boolean bool) {
byte[][] bytes = Stream.of(key, offset, bool ? 1 : 0).map(x -> String.valueOf(x).getBytes(StandardCharsets.UTF_8)).toArray(byte[][]::new);
send("SETBIT", CacheEntryType.OBJECT, (Type) null, key, bytes).join();
}
//--------------------- bit ------------------------------
//--------------------- lock ------------------------------
// 尝试加锁，成功返回0，否则返回上一锁剩余毫秒值
public int tryLock(String key, int millis) {
byte[][] bytes = Stream.of("" +
"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).map(x -> String.valueOf(x).getBytes(StandardCharsets.UTF_8)).toArray(byte[][]::new);
int n = (int) send("EVAL", CacheEntryType.OBJECT, (Type) null, null, bytes).join();
return n;
}
// 加锁
public void lock(String key, int millis) {
int 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 long getTtl(String key) {
return (long) send("TTL", CacheEntryType.OBJECT, (Type) null, key, key.getBytes(StandardCharsets.UTF_8)).join();
}
public long getPttl(String key) {
return (long) send("PTTL", CacheEntryType.OBJECT, (Type) null, key, key.getBytes(StandardCharsets.UTF_8)).join();
}
public int remove(String... keys) {
if (keys == null || keys.length == 0) {
return 0;
}
List<String> para = new ArrayList<>();
para.add("" +
" local args = ARGV;" +
" local x = 0;" +
" for i,v in ipairs(args) do" +
" local inx = redis.call('del', v);" +
" if(inx > 0) then" +
" x = x + 1;" +
" end" +
" end" +
" return x;");
para.add("0");
for (Object field : keys) {
para.add(String.valueOf(field));
}
byte[][] bytes = para.stream().map(x -> String.valueOf(x).getBytes(StandardCharsets.UTF_8)).toArray(byte[][]::new);
return (int) send("EVAL", CacheEntryType.OBJECT, (Type) null, null, bytes).join();
}
//--------------------- hmget ------------------------------
public <T extends Object> V getHm(String key, T field) {
// return (V) send("HMGET", CacheEntryType.OBJECT, (Type) null, key, key.getBytes(StandardCharsets.UTF_8), field.getBytes(StandardCharsets.UTF_8)).join();
Map<Object, V> map = getHms(key, field);
return map.get(field);
}
public <T extends Object> Map<T, V> getHms(String key, T... field) {
if (field == null || field.length == 0) {
return new HashMap<>();
}
byte[][] bytes = Stream.concat(Stream.of(key), Stream.of(field)).map(x -> String.valueOf(x).getBytes(StandardCharsets.UTF_8)).toArray(byte[][]::new);
Map<T, V> result = new HashMap<>();
List<V> vs = (List) send("HMGET", CacheEntryType.OBJECT, (Type) null, key, bytes).join();
for (int i = 0; i < field.length; i++) { // /*vs != null && vs.size() > i &&*/
if (vs.get(i) == null) {
continue;
}
result.put(field[i], vs.get(i));
}
return result;
}
public Map<String, V> getHmall(String key) {
List<V> vs = (List) send("HGETALL", CacheEntryType.OBJECT, (Type) null, key, key.getBytes(StandardCharsets.UTF_8)).join();
Map<String, V> result = new HashMap<>(vs.size() / 2);
for (int i = 0; i < vs.size(); i += 2) {
result.put(String.valueOf(vs.get(i)), vs.get(i + 1));
}
return result;
}
//--------------------- hmset、hmdel、incr ------------------------------
public <T> void setHm(String key, T field, V value) {
byte[][] bytes = Stream.of(key, field, value).map(x -> x.toString().getBytes(StandardCharsets.UTF_8)).toArray(byte[][]::new);
send("HMSET", CacheEntryType.OBJECT, (Type) null, key, bytes).join();
}
public <T> void setHms(String key, Map<T, V> kv) {
List<String> args = new ArrayList();
args.add(key);
kv.forEach((k, v) -> {
args.add(String.valueOf(k));
args.add(String.valueOf(v));
});
byte[][] bytes = args.stream().map(x -> x.getBytes(StandardCharsets.UTF_8)).toArray(byte[][]::new);
send("HMSET", CacheEntryType.OBJECT, (Type) null, key, bytes).join();
}
public <T> Long incrHm(String key, T field, long n) {
byte[][] bytes = Stream.of(key, String.valueOf(field), String.valueOf(n)).map(x -> x.getBytes(StandardCharsets.UTF_8)).toArray(byte[][]::new);
return (Long) send("HINCRBY", CacheEntryType.OBJECT, (Type) null, key, bytes).join();
}
public <T> Double incrHm(String key, T field, double n) {
byte[][] bytes = Stream.of(key, String.valueOf(field), String.valueOf(n)).map(x -> x.getBytes(StandardCharsets.UTF_8)).toArray(byte[][]::new);
Serializable v = send("HINCRBYFLOAT", CacheEntryType.OBJECT, (Type) null, key, bytes).join();
if (v == null) {
return null;
}
return Double.parseDouble(String.valueOf(v));
}
public <T> void hdel(String key, T... field) {
byte[][] bytes = Stream.concat(Stream.of(key), Stream.of(field)).map(x -> String.valueOf(x).getBytes(StandardCharsets.UTF_8)).toArray(byte[][]::new);
send("HDEL", null, (Type) null, key, bytes).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));
}
byte[][] bytes = para.stream().map(x -> String.valueOf(x).getBytes(StandardCharsets.UTF_8)).toArray(byte[][]::new);
return (List<T>) send("EVAL", CacheEntryType.OBJECT, (Type) null, null, bytes).join();
}
//--------------------- set ------------------------------
public <T> T srandomItem(String key) {
byte[][] bytes = Stream.of(key, 1).map(x -> formatValue(CacheEntryType.OBJECT, (Convert) null, (Type) null, x)).toArray(byte[][]::new);
List<T> list = (List) send("SRANDMEMBER", null, (Type) null, key, bytes).join();
return list != null && !list.isEmpty() ? list.get(0) : null;
}
public <T> List<T> srandomItems(String key, int n) {
byte[][] bytes = Stream.of(key, n).map(x -> formatValue(CacheEntryType.OBJECT, (Convert) null, (Type) null, x)).toArray(byte[][]::new);
return (List) send("SRANDMEMBER", null, (Type) null, key, bytes).join();
}
//--------------------- list ------------------------------
public CompletableFuture<Void> appendListItemsAsync(String key, V... values) {
byte[][] bytes = Stream.concat(Stream.of(key), Stream.of(values)).map(x -> String.valueOf(x).getBytes(StandardCharsets.UTF_8)).toArray(byte[][]::new);
return (CompletableFuture) send("RPUSH", null, (Type) null, key, bytes);
}
public CompletableFuture<Void> lpushListItemAsync(String key, V value) {
return (CompletableFuture) send("LPUSH", null, (Type) null, key, key.getBytes(StandardCharsets.UTF_8), formatValue(CacheEntryType.OBJECT, (Convert) null, (Type) null, value));
}
public void lpushListItem(String key, V value) {
lpushListItemAsync(key, value).join();
}
public void appendListItems(String key, V... values) {
appendListItemsAsync(key, values).join();
}
public void appendSetItems(String key, V... values) {
// todo:
for (V v : values) {
appendSetItem(key, v);
}
}
// 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15
public CompletableFuture<Collection<V>> getCollectionAsync(String key, int offset, int limit) {
return (CompletableFuture) send("OBJECT", null, (Type) null, key, "ENCODING".getBytes(StandardCharsets.UTF_8), key.getBytes(StandardCharsets.UTF_8)).thenCompose(t -> {
if (t == null) return CompletableFuture.completedFuture(null);
if (new String((byte[]) t).contains("list")) { //list
return send("LRANGE", CacheEntryType.OBJECT, (Type) null, false, key, key.getBytes(StandardCharsets.UTF_8), String.valueOf(offset).getBytes(StandardCharsets.UTF_8), String.valueOf(offset + limit - 1).getBytes(StandardCharsets.UTF_8));
} else {
return send("SMEMBERS", CacheEntryType.OBJECT, (Type) null, true, key, key.getBytes(StandardCharsets.UTF_8));
}
});
}
public Collection<V> getCollection(String key, int offset, int limit) {
return getCollectionAsync(key, offset, limit).join();
}
public V brpop(String key, int seconds) {
byte[][] bytes = Stream.concat(Stream.of(key), Stream.of(seconds)).map(x -> String.valueOf(x).getBytes(StandardCharsets.UTF_8)).toArray(byte[][]::new);
return (V) send("BRPOP", null, (Type) null, key, bytes).join();
}
//--------------------- zset ------------------------------
public <N extends Number> void zadd(String key, Map<V, N> kv) {
if (kv == null || kv.isEmpty()) {
return;
}
List<String> args = new ArrayList();
args.add(key);
kv.forEach((k, v) -> {
args.add(String.valueOf(v));
args.add(String.valueOf(k));
});
byte[][] bytes = args.stream().map(x -> x.getBytes(StandardCharsets.UTF_8)).toArray(byte[][]::new);
send("ZADD", CacheEntryType.OBJECT, (Type) null, key, bytes).join();
}
public <N extends Number> double zincr(String key, Object number, N n) {
byte[][] bytes = Stream.of(key, n, number).map(x -> String.valueOf(x).getBytes(StandardCharsets.UTF_8)).toArray(byte[][]::new);
Serializable v = send("ZINCRBY", CacheEntryType.OBJECT, (Type) null, key, bytes).join();
return Double.parseDouble(String.valueOf(v));
}
public void zrem(String key, V... vs) {
List<String> args = new ArrayList();
args.add(key);
for (V v : vs) {
args.add(String.valueOf(v));
}
byte[][] bytes = args.stream().map(x -> x.getBytes(StandardCharsets.UTF_8)).toArray(byte[][]::new);
send("ZREM", CacheEntryType.OBJECT, (Type) null, key, bytes).join();
}
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;
}
// ----------
protected byte[] formatValue(CacheEntryType cacheType, Convert convert0, Type resultType, Object value) {
if (value == null) return "null".getBytes(StandardCharsets.UTF_8);
if (convert0 == null) convert0 = convert;
if (cacheType == CacheEntryType.LONG || cacheType == CacheEntryType.ATOMIC)
return String.valueOf(value).getBytes(StandardCharsets.UTF_8);
if (cacheType == CacheEntryType.STRING) return convert0.convertToBytes(String.class, value);
if (value instanceof String) return String.valueOf(value).getBytes(StandardCharsets.UTF_8);
if (value instanceof Number) return String.valueOf(value).getBytes(StandardCharsets.UTF_8);
return convert0.convertToBytes(resultType == null ? objValueType : resultType, value);
}
}

2240
src/com/zdemo/cachex/RedisCacheSource.java
View File

File diff suppressed because it is too large Load Diff

244
src/com/zdemo/cachex/RedisTest.java
View File

@ -1,244 +0,0 @@
package com.zdemo.cachex;
import org.redkale.convert.json.JsonFactory;
import org.redkale.util.AnyValue;
import java.util.Map;
public class RedisTest {
static MyRedisCacheSource<String> source = new MyRedisCacheSource();
static {
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) {
//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();
}
}*/
}
}

1944
src/com/zdemo/cachex/RedissionCacheSource.java
View File

File diff suppressed because it is too large Load Diff

93
src/com/zdemo/kafak/KafakConsumer.java
View File

@ -1,93 +0,0 @@
package com.zdemo.kafak;
import com.zdemo.AbstractConsumer;
import com.zdemo.IConsumer;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.errors.WakeupException;
import org.redkale.service.Service;
import org.redkale.util.AnyValue;
import org.redkale.util.AutoLoad;
import javax.annotation.Resource;
import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.time.Duration;
import java.util.Properties;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.logging.Level;
import java.util.logging.Logger;
/**
* 消费
*/
@AutoLoad(false)
public abstract class KafakConsumer extends AbstractConsumer implements IConsumer, Service {
public Logger logger = Logger.getLogger(this.getClass().getSimpleName());
@Resource(name = "APP_HOME")
protected File APP_HOME;
private final LinkedBlockingQueue<Runnable> queue = new LinkedBlockingQueue<>();
@Override
public final void init(AnyValue config) {
if (!preInit()) {
return;
}
try (FileInputStream fis = new FileInputStream(new File(APP_HOME, "conf/kafak.properties"));) {
Properties props = new Properties();
props.load(fis);
new Thread(() -> {
try {
props.put("group.id", getGroupid());
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(getTopics());
while (true) {
ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1_000));
records.forEach(record -> {
String topic = record.topic();
long offset = record.offset();
String value = record.value();
try {
accept(topic, value);
} catch (Exception e) {
logger.log(Level.WARNING, String.format("topic[%s] event accept error, offset=%s,value:%s", topic, offset, value), e);
}
});
if (!queue.isEmpty()) {
Runnable runnable;
while ((runnable = queue.poll()) != null) {
runnable.run();
}
consumer.unsubscribe();
consumer.subscribe(getTopics());
}
}
} catch (WakeupException ex) {
System.out.println("WakeupException !!!!");
}
}, "thread-consumer-[" + getGroupid() + "]").start();
} catch (IOException e) {
logger.log(Level.WARNING, "", e);
}
}
@Override
public void unsubscribe(String topic) {
queue.add(() -> super.removeEventType(topic)); // 加入延时执行队列（下一次订阅变更检查周期执行）
}
@Override
protected void subscribe(String topic) {
queue.add(() -> {
// just set flag, nothing to do
});
}
}

59
src/com/zdemo/kafak/KafakProducer.java
View File

@ -1,59 +0,0 @@
package com.zdemo.kafak;
import com.zdemo.IProducer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.redkale.convert.json.JsonConvert;
import org.redkale.service.Service;
import org.redkale.util.AnyValue;
import org.redkale.util.AutoLoad;
import javax.annotation.Resource;
import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.util.Properties;
import java.util.logging.Level;
/**
* 生产
*
* @param
*/
@AutoLoad(false)
public class KafakProducer implements IProducer, Service {
private KafkaProducer<String, String> producer;
@Resource(name = "APP_HOME")
protected File APP_HOME;
@Override
public void init(AnyValue config) {
File file = new File(APP_HOME, "conf/kafak.properties");
try (FileInputStream fis = new FileInputStream(file)) {
Properties props = new Properties();
props.load(fis);
producer = new KafkaProducer(props);
} catch (IOException e) {
logger.log(Level.WARNING, "未初始化kafak 生产者，kafak发布消息不可用", e);
}
}
@Override
public boolean publish(String topic, Object v) {
producer.send(new ProducerRecord(topic, toStr(v)));
return true;
}
@Override
public void destroy(AnyValue config) {
producer.close();
}
private <V> String toStr(V v) {
if (v instanceof String) {
return (String) v;
}
return JsonConvert.root().convertTo(v);
}
}

29
src/com/zdemo/pulsar/AService.java
View File

@ -1,29 +0,0 @@
package com.zdemo.pulsar;
import com.zdemo.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.Utility;
import javax.annotation.Resource;
@RestService
public class AService implements Service {
@Resource(name = "zhub")
private ZHubClient zhub;
@Override
public void init(AnyValue config) {
zhub.timer("a", () -> {
System.out.println(Utility.now() + " timer RANK-DATA-RELOADALL 执行了");
});
}
@RestMapping
public void x() {
}
}

91
src/com/zdemo/pulsar/PulsarConsumer.java
View File

@ -1,91 +0,0 @@
package com.zdemo.pulsar;
import com.zdemo.AbstractConsumer;
import com.zdemo.EventType;
import com.zdemo.IConsumer;
import org.apache.pulsar.client.api.*;
import org.redkale.service.Service;
import org.redkale.util.AnyValue;
import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.logging.Level;
public abstract class PulsarConsumer extends AbstractConsumer implements IConsumer, Service {
@Resource(name = "property.pulsar.serviceurl")
private String serviceurl = "pulsar://127.0.0.1:6650";
private PulsarClient client;
private Consumer consumer;
public abstract String getGroupid();
private final LinkedBlockingQueue<Runnable> queue = new LinkedBlockingQueue<>();
@Override
public void addEventType(EventType... eventTypes) {
super.addEventType(eventTypes);
// 增加变更标记
queue.add(() -> logger.info("PulsarConsumer add new topic!"));
}
@Override
public void init(AnyValue config) {
if (!preInit()) {
return;
}
queue.add(() -> logger.info("PulsarConsumer starting ..."));
new Thread(() -> {
try {
client = PulsarClient.builder()
.serviceUrl(serviceurl)
.build();
while (true) {
// 动态新增订阅
if (!queue.isEmpty()) {
Runnable runnable;
while ((runnable = queue.poll()) != null) {
runnable.run();
}
consumer.unsubscribe();
consumer = client.newConsumer()
.topics(new ArrayList<>(getTopics()))
.subscriptionName(getGroupid())
.subscriptionType(SubscriptionType.Shared)
.subscribe();
}
// Wait for a message
Message msg = consumer.receive(10, TimeUnit.SECONDS);
if (msg == null) {
continue;
}
String topic = msg.getTopicName().replace("persistent://public/default/", "");
long offset = 0;
String value = new String(msg.getData());
try {
accept(topic, value);
consumer.acknowledge(msg); // Acknowledge the message so that it can be deleted by the message broker
} catch (Exception e) {
logger.log(Level.WARNING, String.format("topic[%s] event accept error, offset=%s,value:%s", topic, offset, value), e);
consumer.negativeAcknowledge(msg); // Message failed to process, redeliver later
}
}
} catch (PulsarClientException e) {
e.printStackTrace();
}
}).start();
}
@Override
public void unsubscribe(String topic) {
}
}

82
src/com/zdemo/pulsar/PulsarProducer.java
View File

@ -1,82 +0,0 @@
package com.zdemo.pulsar;
import com.zdemo.Event;
import com.zdemo.IProducer;
import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.PulsarClientException;
import org.redkale.convert.json.JsonConvert;
import org.redkale.service.Service;
import org.redkale.util.AnyValue;
import org.redkale.util.Comment;
import javax.annotation.Resource;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.logging.Level;
public class PulsarProducer<T extends Event> implements IProducer<T>, Service {
@Resource(name = "property.pulsar.serviceurl")
private String serviceurl = "pulsar://127.0.0.1:6650";
@Comment("消息生产者")
private Map<String, Producer<byte[]>> producerMap = new HashMap();
private PulsarClient client;
@Override
public void init(AnyValue config) {
try {
client = PulsarClient.builder()
.serviceUrl(serviceurl)
.build();
} catch (PulsarClientException e) {
e.printStackTrace();
}
}
public Producer<byte[]> getProducer(String topic) {
Producer<byte[]> producer = producerMap.get(topic);
if (producer != null) {
return producer;
}
synchronized (this) {
if ((producer = producerMap.get(topic)) == null) {
try {
producer = client.newProducer()
.topic(topic)
.batchingMaxPublishDelay(10, TimeUnit.MILLISECONDS)
.sendTimeout(10, TimeUnit.SECONDS)
.blockIfQueueFull(true)
.create();
producerMap.put(topic, producer);
return producer;
} catch (PulsarClientException e) {
e.printStackTrace();
}
}
}
return producer;
}
@Override
public void send(T t) {
try {
Producer<byte[]> producer = getProducer(t.topic);
String v = JsonConvert.root().convertTo(t.value);
if (v.startsWith("\"") && v.endsWith("\"")) {
v = v.substring(1, v.length() - 1);
}
producer.newMessage()
.key("")
.value(v.getBytes())
.send();
} catch (Exception e) {
logger.log(Level.WARNING, "", e);
}
}
}

114
src/com/zdemo/redis/RedisConsumer.java
View File

@ -1,114 +0,0 @@
package com.zdemo.redis;
import com.zdemo.AbstractConsumer;
import com.zdemo.IConsumer;
import org.redkale.service.Service;
import org.redkale.util.AnyValue;
import org.redkale.util.AutoLoad;
import javax.annotation.Resource;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.OutputStreamWriter;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.util.logging.Level;
import java.util.logging.Logger;
@AutoLoad(false)
public class RedisConsumer extends AbstractConsumer implements IConsumer, Service {
public Logger logger = Logger.getLogger(this.getClass().getSimpleName());
@Resource(name = "property.redis.host")
private String host = "127.0.0.1";
@Resource(name = "property.redis.password")
private String password = "";
@Resource(name = "property.redis.port")
private int port = 6379;
private Socket client;
private OutputStreamWriter writer;
private BufferedReader reader;
@Override
public void init(AnyValue config) {
try {
client = new Socket();
client.connect(new InetSocketAddress(host, port));
client.setKeepAlive(true);
writer = new OutputStreamWriter(client.getOutputStream());
writer.write("AUTH " + password + "\r\n");
writer.flush();
StringBuffer buf = new StringBuffer("SUBSCRIBE");
for (String topic : getTopics()) {
buf.append(" ").append(topic);
}
buf.append("\r\n");
writer.write(buf.toString());
writer.flush();
reader = new BufferedReader(new InputStreamReader(client.getInputStream()));
} catch (IOException e) {
logger.log(Level.WARNING, "Redis Consumer 初始化失败！", e);
}
new Thread(() -> {
try {
while (true) {
String readLine = reader.readLine();
String type = "";
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
reader.readLine(); //$n len(value)
String value = reader.readLine(); // value
try {
accept(topic, value);
} catch (Exception e) {
logger.log(Level.WARNING, "topic[" + topic + "] event accept error :" + value, e);
}
}
}
} catch (IOException e) {
logger.log(Level.WARNING, "", e);
}
}).start();
}
@Override
protected String getGroupid() {
return null;
}
@Override
public void unsubscribe(String topic) {
try {
writer.write("UNSUBSCRIBE " + topic + "\r\n");
writer.flush();
} catch (IOException e) {
logger.log(Level.WARNING, "", e);
}
super.removeEventType(topic);
}
@Override
protected void subscribe(String topic) {
//新增订阅
try {
writer.write("SUBSCRIBE " + topic + "\r\n");
writer.flush();
} catch (IOException e) {
logger.log(Level.WARNING, "", e);
}
}
}

61
src/com/zdemo/redis/RedisProducer.java
View File

@ -1,61 +0,0 @@
package com.zdemo.redis;
import com.zdemo.IProducer;
import org.redkale.convert.json.JsonConvert;
import org.redkale.service.Service;
import org.redkale.util.AnyValue;
import org.redkale.util.AutoLoad;
import javax.annotation.Resource;
import java.io.IOException;
import java.io.OutputStreamWriter;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.util.logging.Level;
@AutoLoad(false)
public class RedisProducer implements IProducer, Service {
@Resource(name = "property.redis.host")
private String host = "127.0.0.1";
@Resource(name = "property.redis.password")
private String password = "";
@Resource(name = "property.redis.port")
private int port = 6379;
private OutputStreamWriter osw;
@Override
public void init(AnyValue config) {
try {
Socket client = new Socket();
client.connect(new InetSocketAddress(host, port));
client.setKeepAlive(true);
osw = new OutputStreamWriter(client.getOutputStream());
osw.write("AUTH " + password + "\r\n");
osw.flush();
} catch (IOException e) {
logger.log(Level.WARNING, "", e);
}
}
@Override
public boolean publish(String topic, Object v) {
try {
osw.write("PUBLISH " + topic + " '" + toStr(v) + "' \r\n");
osw.flush();
return true;
} catch (IOException e) {
logger.log(Level.WARNING, "", e);
}
return false;
}
private String toStr(Object v) {
if (v instanceof String) {
return (String) v;
}
return JsonConvert.root().convertTo(v);
}
}

15
src/com/zdemo/zhub/Rpc.java
View File

@ -1,13 +1,19 @@
package com.zdemo.zhub;
import org.redkale.convert.ConvertColumn;
import org.redkale.convert.json.JsonConvert;
import com.google.gson.Gson;
import com.google.gson.annotations.Expose;
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;
public Rpc() {
@ -16,7 +22,7 @@ public class Rpc<T> {
public Rpc(String appname, String ruk, String topic, Object value) {
this.ruk = appname + "::" + ruk;
this.topic = topic;
this.value = (T) JsonConvert.root().convertTo(value);
this.value = (T) gson.toJson(value);
}
public String getRuk() {
@ -43,7 +49,7 @@ public class Rpc<T> {
this.value = value;
}
@ConvertColumn(ignore = true)
public RpcResult getRpcResult() {
return rpcResult;
}
@ -52,7 +58,6 @@ public class Rpc<T> {
this.rpcResult = rpcResult;
}
@ConvertColumn(ignore = true)
public String getBackTopic() {
return ruk.split("::")[0];
}

54
src/com/zdemo/zhub/ZHubClient.java
View File

@ -1,12 +1,14 @@
package com.zdemo.zhub;
import com.google.gson.reflect.TypeToken;
import com.zdemo.AbstractConsumer;
import com.zdemo.Event;
import com.zdemo.IConsumer;
import com.zdemo.IProducer;
import net.tccn.timer.Timers;
import org.redkale.service.Service;
import org.redkale.util.*;
import org.redkale.util.AnyValue;
import org.redkale.util.Comment;
import org.redkale.util.Utility;
import java.io.BufferedReader;
import java.io.IOException;
@ -15,10 +17,7 @@ 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.Set;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;
@ -28,8 +27,8 @@ import java.util.function.Function;
import java.util.logging.Level;
import java.util.logging.Logger;
@AutoLoad(value = false)
public class ZHubClient extends AbstractConsumer implements IConsumer, IProducer, Service {
public class ZHubClient extends AbstractConsumer implements IConsumer, IProducer {
public Logger logger = Logger.getLogger(ZHubClient.class.getSimpleName());
private String addr = "127.0.0.1:1216";
@ -57,7 +56,13 @@ public class ZHubClient extends AbstractConsumer implements IConsumer, IProducer
private boolean isMain = false;*/
private static Map<String, ZHubClient> mainHub = new HashMap<>(); // 127.0.0.1:1216 - ZHubClient
@Override
public ZHubClient(String addr, String groupid, String appname) {
this.addr = addr;
this.groupid = groupid;
this.APP_NAME = appname;
init(null);
}
public void init(AnyValue config) {
if (!preInit()) {
return;
@ -225,7 +230,7 @@ public class ZHubClient extends AbstractConsumer implements IConsumer, IProducer
continue;
}
//if (event)
logger.finest(String.format("rpc-back:[%s]: %s", event.topic, event.value));
logger.info(String.format("rpc-back:[%s]: %s", event.topic, event.value));
rpcAccept(event.value);
} catch (InterruptedException e) {
e.printStackTrace();
@ -243,7 +248,7 @@ public class ZHubClient extends AbstractConsumer implements IConsumer, IProducer
if ((event = rpcCallQueue.take()) == null) {
continue;
}
logger.finest(String.format("rpc-call:[%s] %s", event.topic, event.value));
logger.info(String.format("rpc-call:[%s] %s", event.topic, event.value));
accept(event.topic, event.value);
} catch (InterruptedException e) {
e.printStackTrace();
@ -323,7 +328,8 @@ public class ZHubClient extends AbstractConsumer implements IConsumer, IProducer
} else if (v == null) {
return null;
}
return convert.convertTo(v);
return gson.toJson(v);
}
protected boolean initSocket(int retry) {
@ -449,7 +455,7 @@ public class ZHubClient extends AbstractConsumer implements IConsumer, IProducer
private Map<String, Lock> lockTag = new ConcurrentHashMap<>();
public Lock tryLock(String key, int duration) {
String uuid = Utility.uuid();
String uuid = UUID.randomUUID().toString().replaceAll("-", "");
Lock lock = new Lock(key, uuid, duration, this);
lockTag.put(uuid, lock);
@ -503,12 +509,12 @@ public class ZHubClient extends AbstractConsumer implements IConsumer, IProducer
private static Map<String, Rpc> rpcMap = new ConcurrentHashMap<>();
private static Map<String, TypeToken> rpcRetType = new ConcurrentHashMap<>();
@Comment("rpc call")
// rpc call
public RpcResult<Void> rpc(String topic, Object v) {
return rpc(topic, v, null);
}
@Comment("rpc call")
// rpc call
public <T, R> RpcResult<R> rpc(String topic, T v, TypeToken<R> typeToken) {
return rpc(topic, v, typeToken, 0);
}
@ -537,7 +543,7 @@ public class ZHubClient extends AbstractConsumer implements IConsumer, IProducer
RpcResult rpcResult = rpc.buildResp(505, "请求超时");
rpc.setRpcResult(rpcResult);
logger.warning("rpc timeout: " + convert.convertTo(rpc));
logger.warning("rpc timeout: " + gson.toJson(rpc));
rpc.notify();
}
}, timeout);
@ -571,10 +577,10 @@ public class ZHubClient extends AbstractConsumer implements IConsumer, IProducer
}
// RpcResult: {ruk:xxx-xxxx, retcode:0}
@Comment("rpc call back consumer")
// rpc call back consumer
private void rpcAccept(String value) {
RpcResult resp = convert.convertFrom(new TypeToken<RpcResult<String>>() {
}.getType(), value);
RpcResult resp = gson.fromJson(value, new TypeToken<RpcResult<String>>() {
}.getType());
String ruk = resp.getRuk();
Rpc rpc = rpcMap.remove(ruk);
@ -585,7 +591,7 @@ public class ZHubClient extends AbstractConsumer implements IConsumer, IProducer
Object result = resp.getResult();
if (result != null && typeToken != null && !"java.lang.String".equals(typeToken.getType().getTypeName())) {
result = convert.convertFrom(typeToken.getType(), (String) resp.getResult());
result = gson.fromJson((String) resp.getResult(), typeToken.getType());
}
resp.setResult(result);
@ -598,16 +604,16 @@ public class ZHubClient extends AbstractConsumer implements IConsumer, IProducer
// -- 订阅端 --
private Set<String> rpcTopics = new HashSet();
@Comment("rpc call consumer")
// 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 = convert.convertFrom(new TypeToken<Rpc<String>>() {
}.getType(), v);
rpc = gson.fromJson(v, new TypeToken<Rpc<String>>() {
}.getType());
// 参数转换
T paras = convert.convertFrom(typeToken.getType(), (String) rpc.getValue());
T paras = gson.fromJson((String) rpc.getValue(), typeToken.getType());
rpc.setValue(paras);
RpcResult result = fun.apply(rpc);
result.setResult(toStr(result.getResult()));

2
src/net/tccn/timer/Timers.java
View File

@ -11,6 +11,7 @@ public class Timers {
/**
* 本地延时重试
*
* @param supplier
* @param millis
* @param maxCount
@ -31,6 +32,7 @@ public class Timers {
/**
* 本地延时：延时时间极短的场景下使用 （如：1分钟内）
*
* @param runnable
* @param millis
*/

1
src/net/tccn/timer/task/Task.java
View File

@ -63,6 +63,7 @@ public interface Task extends Runnable {
/**
* 得到总执行次数
*
* @return
*/
int getExecCount();

63
test/HelloService.java Normal file
View File

@ -0,0 +1,63 @@
import com.zdemo.IConsumer;
import com.zdemo.zhub.RpcResult;
import com.zdemo.zhub.ZHubClient;
import org.junit.Before;
import org.junit.Test;
// @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");
//zhub.init(Map.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", "❤别人家✉<EFBFBD>的女娃子❤🤞🏻", 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);
System.out.println(rpc.getResult());
}
/*RpcResult<FileToken> x = zhub.rpc("rpc:file:up-token", Map.of(), new TypeToken<>() {
});*/
}

450
test/com/zdemo/test/AppTest.java
View File

@ -1,450 +0,0 @@
package com.zdemo.test;
import com.zdemo.Event;
import com.zdemo.IProducer;
import com.zdemo.zhub.Delays;
import net.tccn.timer.Timers;
import org.junit.Test;
import org.redkale.boot.Application;
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());
//启动并开启消费监听
MyConsumer consumer = Application.singleton(MyConsumer.class);
consumer.subscribe("a", strx -> {
logger.info("我收到了消息 a 事件：" + str);
});
/*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() {
try {
MyConsumer producer = Application.singleton(MyConsumer.class);
for (int i = 0; i < 10_0000; i++) {
producer.publish("a-1", i);
}
try {
Thread.sleep(1_000);
} catch (InterruptedException e) {
e.printStackTrace();
}
} catch (Exception e) {
e.printStackTrace();
}
}
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());
}
@Test
public void yy() {
IProducer producer = null;
try {
producer = Application.singleton(MyConsumer.class);
for (int i = 0; i < 100; i++) {
producer.publish("x", "x");
Thread.sleep(1000);
}
} catch (Exception e) {
e.printStackTrace();
}
}
// (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<Delays> delayQueue = new DelayQueue<>();
logger.info("加入延时任务1");
delayQueue.add(new Delays(5000, () -> {
logger.info("任务1 延时任务执行了!");
}));
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
logger.info("加入延时任务2");
delayQueue.add(new Delays(5000, () -> {
logger.info("任务2 延时任务执行了!");
}));
try {
while (true) {
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();
}
}
}

72
test/com/zdemo/test/Delays.java
View File

@ -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();
}
}

232
test/com/zdemo/test/HelloService.java
View File

@ -1,232 +0,0 @@
package com.zdemo.test;
import com.zdemo.IType;
import com.zdemo.zhub.RpcResult;
import com.zdemo.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)
public class HelloService implements Service {
@Resource(name = "zhub")
private ZHubClient zhub;
private net.tccn.zhub.ZHubClient zhubx = null;
@Override
public void init(AnyValue config) {
CompletableFuture.runAsync(() -> {
zhubx = new net.tccn.zhub.ZHubClient("127.0.0.1", 1216, "g-dev", "DEV-LOCAL");
//zhubx = new net.tccn.zhub.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.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", "❤别人家✉<EFBFBD>的女娃子❤🤞🏻", 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<>() {
});*/
net.tccn.zhub.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";
}
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();
}
}

15
test/com/zdemo/test/MyConsumer.java
View File

@ -1,15 +0,0 @@
package com.zdemo.test;
import com.zdemo.zhub.ZHubClient;
public class MyConsumer extends ZHubClient {
public String getGroupid() {
return "group-test"; //消费组名称
}
@Override
protected boolean preInit() {
return true;
}
}