新增:1.redtimer 代码包暂时合并到此工程

2.新增Timers.tryDelay、Timers.delay 方法
This commit is contained in:
2021-11-03 19:56:27 +08:00
parent f1e1974aa2
commit c6a24098ff
17 changed files with 3774 additions and 16 deletions

View File

@@ -2,6 +2,8 @@ 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;
@@ -9,10 +11,13 @@ 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;
/**
* 消息发布订阅测试
@@ -23,10 +28,15 @@ public class AppTest {
@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", str -> {
consumer.subscribe("a", strx -> {
logger.info("我收到了消息 a 事件:" + str);
});
@@ -333,5 +343,108 @@ public class AppTest {
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();
}
}
}

View File

@@ -0,0 +1,72 @@
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();
}
}

View File

@@ -3,12 +3,17 @@ package com.zdemo.test;
import com.zdemo.IConsumer;
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 {
@@ -16,22 +21,212 @@ 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>() {
/*zhub.rpcSubscribe("x", new TypeToken<String>() {
}, r -> {
return r.buildResp(r.getValue().toUpperCase() + ": Ok");
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", "❤别人家✉<E5AEB6>的女娃子❤🤞🏻", 0);
zhub.delay("sport:reqtime", "中文特殊符号:『』 £ ♀ ‖ 「」\n" +
"英文:# + = & ﹉ .. ^ \"\" ·{ } % ' €\n" +
"数学:+× ° ± ℃ ㎡ ∑ ≥ ∫ ㏄ ⊥ ≯ ∠ ∴ ∈ ∧ ∵ ≮ ㎝ ㏑ ≌ ㎞ № § ℉ ÷ ‰ ㎎ ㎏ ㎜ ㏒ ⊙ ∮ ∝ ∞ º ¹ ² ³ ½ ¾ ¼ ≈ ≡ ≠ ≤ ≦ ≧ ∽ ∷ ∏ ∩ ⌒ √Ψ ¤ ‖ ¶\n" +
"特殊:♤ ♧ ♡ ♢ ♪ ♬ ♭ ✔ ✘ ♞ ♟ ↪ ↣ ♚ ♛ ♝ ☞ ☜ ⇔ ☆ ★ □ ■ ○ ● △ ▲ ▽ ▼ ◇ ◆ ♀ ♂ ※ ↓ ↑ ↔ ↖ ↙ ↗ ↘ ← → ♣ ♠ ♥ ◎ ◣ ◢ ◤ ◥ 卍 ℡ ⊙ ㊣ ® © ™ ㈱ 囍\n" +
"序号:①②③④⑤⑥⑦⑧⑨⑩㈠㈡㈢㈣㈤㈥㈦㈧㈨㈩⑴ ⑵ ⑶ ⑷ ⑸ ⑹ ⑺ ⑻ ⑼ ⑽ ⒈ ⒉ ⒊ ⒋ ⒌ ⒍ ⒎ ⒏ ⒐ ⒑ Ⅱ Ⅲ Ⅳ Ⅵ Ⅶ Ⅷ ⅨⅩ\n" +
"日文:アイウエオァィゥェォカキクケコガギグゲゴサシスセソザジズゼゾタチツテトダヂヅデドッナニヌネノハヒフヘホバビブベボパピプペポマミムメモャヤュユョラリヨルレロワヰヱヲンヴヵヶヽヾ゛゜ー、。「「あいうえおぁぃぅぇぉかきくけこがぎぐげごさしすせそざじずぜぞたちつてでどっなにぬねのはひふへ」」ほばびぶべぼぱぴぷぺぽまみむめもやゆよゃゅょらりるれろわをんゎ゛゜ー、。「」\n" +
"部首:犭 凵 巛 冖 氵 廴 讠 亻 钅 宀 亠 忄 辶 弋 饣 刂 阝 冫 卩 疒 艹 疋 豸 冂 匸 扌 丬 屮衤 礻 勹 彳 彡", 0);
*/
}
public RpcResult<String> x(String v) {
@RestMapping
public RpcResult x(String v) {
if (v == null) {
v = "";
}
RpcResult<String> x = zhub.rpc("x", v, IConsumer.TYPE_TOKEN_STRING).join();
return x;
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, IConsumer.TYPE_TOKEN_STRING);
}
@RestMapping
public RpcResult<String> d(String v) {
RpcResult<String> rpc = zhub.rpc("x", v, IConsumer.TYPE_TOKEN_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();
}
}