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

修改：删除多余依赖调整,代码风格调整

Browse Source
This commit is contained in:
梁显优 2021-12-31 12:55:01 +08:00
parent 3debcee7d2
commit 0df9e254e6
7 changed files with 57 additions and 108 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

26
conf/application.xml
View File

@ -1,26 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<application port="2001" name="xxx">
<zhubs>
<!--<zhub name="zhub" addr="47.111.150.118" port="6066" groupid="group-zhub"/>-->
<zhub name="zhub" addr="127.0.0.1:1216" groupid="group-zhub"/>
</zhubs>
<resources>
<properties load="config.properties"></properties>
<listener value="com.zdemo.ZhubListener"/>
</resources>
<server protocol="HTTP" port="80">
<request>
<remoteaddr value="request.headers.X-Real-IP"/>
</request>
<rest autoload="true" path="/"/>
<services autoload="true"/>
<servlets path="/"/>
</server>
</application>

19
conf/kafak.properties
View File

@ -1,19 +0,0 @@
# Producer
#bootstrap.servers=47.111.150.118:6062
#bootstrap.servers=121.196.17.55:6062
bootstrap.servers=39.108.56.246:9092
#bootstrap.servers=122.112.180.156:6062
acks=all
retries=0
batch.size=16384
linger.ms=1
buffer.memory=33554432
key.serializer=org.apache.kafka.common.serialization.StringSerializer
value.serializer=org.apache.kafka.common.serialization.StringSerializer
# Consumer
enable.auto.commit=true
auto.commit.interval.ms=1000
group.id=
key.deserializer=org.apache.kafka.common.serialization.StringDeserializer
value.deserializer=org.apache.kafka.common.serialization.StringDeserializer

16
pom.xml
View File

@ -6,7 +6,7 @@
<groupId>net.tccn</groupId> <groupId>net.tccn</groupId>
<artifactId>zhub-client</artifactId> <artifactId>zhub-client</artifactId>
<version>1.0-SNAPSHOT</version> <version>1.0</version>
<properties> <properties>
<maven.compiler.source>17</maven.compiler.source> <maven.compiler.source>17</maven.compiler.source>
@ -21,4 +21,18 @@
</dependency> </dependency>
</dependencies> </dependencies>
<build>
<resources>
<resource>
<directory>src</directory>
</resource>
</resources>
<testResources>
<testResource>
<directory>test</directory>
</testResource>
</testResources>
</build>
</project> </project>

10
src/com/zdemo/IType.java
View File

@ -1,21 +1,21 @@
package com.zdemo; package com.zdemo;
import org.redkale.util.TypeToken; import com.google.gson.reflect.TypeToken;
import java.util.List; import java.util.List;
import java.util.Map; import java.util.Map;
public interface IType { public interface IType {
TypeToken<String> STRING = new TypeToken<String>() { TypeToken<String> STRING = new TypeToken<>() {
}; };
TypeToken<Integer> INT = new TypeToken<Integer>() { TypeToken<Integer> INT = new TypeToken<>() {
}; };
TypeToken<Map<String, String>> MAP = new TypeToken<Map<String, String>>() { TypeToken<Map<String, String>> MAP = new TypeToken<>() {
}; };
TypeToken<List<Map<String, String>>> LMAP = new TypeToken<List<Map<String, String>>>() { TypeToken<List<Map<String, String>>> LMAP = new TypeToken<>() {
}; };
} }

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

@ -6,9 +6,6 @@ import com.zdemo.Event;
import com.zdemo.IConsumer; import com.zdemo.IConsumer;
import com.zdemo.IProducer; import com.zdemo.IProducer;
import net.tccn.timer.Timers; import net.tccn.timer.Timers;
import org.redkale.util.AnyValue;
import org.redkale.util.Comment;
import org.redkale.util.Utility;
import java.io.BufferedReader; import java.io.BufferedReader;
import java.io.IOException; import java.io.IOException;
@ -17,7 +14,10 @@ import java.io.OutputStream;
import java.net.InetSocketAddress; import java.net.InetSocketAddress;
import java.net.Socket; import java.net.Socket;
import java.net.SocketException; import java.net.SocketException;
import java.util.*; import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.CompletableFuture; import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.LinkedBlockingQueue;
@ -35,8 +35,6 @@ public class ZHubClient extends AbstractConsumer implements IConsumer, IProducer
//private String password = ""; //private String password = "";
private String groupid = ""; private String groupid = "";
//private ReentrantLock lock = new ReentrantLock();
private Socket client;
private OutputStream writer; private OutputStream writer;
private BufferedReader reader; private BufferedReader reader;
@ -46,7 +44,7 @@ public class ZHubClient extends AbstractConsumer implements IConsumer, IProducer
private final LinkedBlockingQueue<Event<String>> rpcCallQueue = new LinkedBlockingQueue<>(); // RPC CALL MSG private final LinkedBlockingQueue<Event<String>> rpcCallQueue = new LinkedBlockingQueue<>(); // RPC CALL MSG
private final LinkedBlockingQueue<String> sendMsgQueue = new LinkedBlockingQueue<>(); // SEND MSG private final LinkedBlockingQueue<String> sendMsgQueue = new LinkedBlockingQueue<>(); // SEND MSG
private BiConsumer<Runnable, Integer> threadBuilder = (r, n) -> { private final BiConsumer<Runnable, Integer> threadBuilder = (r, n) -> {
for (int i = 0; i < n; i++) { for (int i = 0; i < n; i++) {
new Thread(() -> r.run()).start(); new Thread(() -> r.run()).start();
} }
@ -54,7 +52,7 @@ public class ZHubClient extends AbstractConsumer implements IConsumer, IProducer
/*private static boolean isFirst = true; /*private static boolean isFirst = true;
private boolean isMain = false;*/ private boolean isMain = false;*/
private static Map<String, ZHubClient> mainHub = new HashMap<>(); // 127.0.0.1:1216 - ZHubClient private static final Map<String, ZHubClient> mainHub = new HashMap<>(); // 127.0.0.1:1216 - ZHubClient
public ZHubClient(String addr, String groupid, String appname) { public ZHubClient(String addr, String groupid, String appname) {
this.addr = addr; this.addr = addr;
@ -63,22 +61,16 @@ public class ZHubClient extends AbstractConsumer implements IConsumer, IProducer
init(null); init(null);
} }
public void init(AnyValue config) { public void init(Map<String, String> config) {
if (!preInit()) { if (!preInit()) {
return; return;
} }
// 自动注入 // 自动注入
if (config != null) { if (config != null) {
addr = config.getValue("addr", addr); addr = config.getOrDefault("addr", addr);
groupid = config.getOrDefault("groupid", groupid);
// 合并 addr = host:port, 做历史兼容 APP_NAME = config.getOrDefault("appname", APP_NAME);
int port = config.getIntValue("port", 0);
if (port != 0 && !addr.contains(":")) {
addr = addr + ":" + port;
}
groupid = config.getValue("groupid", groupid);
} }
// 设置第一个启动的 实例为主实例 // 设置第一个启动的 实例为主实例
@ -102,7 +94,7 @@ public class ZHubClient extends AbstractConsumer implements IConsumer, IProducer
continue; continue;
} }
String type = ""; String type;
// +ping // +ping
if ("+ping".equals(readLine)) { if ("+ping".equals(readLine)) {
@ -112,7 +104,8 @@ public class ZHubClient extends AbstractConsumer implements IConsumer, IProducer
// 主题订阅消息 // 主题订阅消息
if ("*3".equals(readLine)) { if ("*3".equals(readLine)) {
readLine = reader.readLine(); // $7 len() reader.readLine(); // $7 len()
type = reader.readLine(); // message type = reader.readLine(); // message
if (!"message".equals(type)) { if (!"message".equals(type)) {
continue; continue;
@ -165,7 +158,8 @@ public class ZHubClient extends AbstractConsumer implements IConsumer, IProducer
// timer 消息 // timer 消息
if ("*2".equals(readLine)) { if ("*2".equals(readLine)) {
readLine = reader.readLine(); // $7 len() reader.readLine(); // $7 len()
type = reader.readLine(); // message type = reader.readLine(); // message
if (!"timer".equals(type)) { if (!"timer".equals(type)) {
continue; continue;
@ -284,7 +278,7 @@ public class ZHubClient extends AbstractConsumer implements IConsumer, IProducer
if (data.length == 1) { if (data.length == 1) {
sendMsgQueue.add(data[0] + "\r\n"); sendMsgQueue.add(data[0] + "\r\n");
} else if (data.length > 1) { } else if (data.length > 1) {
StringBuffer buf = new StringBuffer(); StringBuilder buf = new StringBuilder();
buf.append("*" + data.length + "\r\n"); buf.append("*" + data.length + "\r\n");
for (String d : data) { for (String d : data) {
buf.append("$" + strLength(d) + "\r\n"); buf.append("$" + strLength(d) + "\r\n");
@ -339,7 +333,8 @@ public class ZHubClient extends AbstractConsumer implements IConsumer, IProducer
String host = hostPort[0]; String host = hostPort[0];
int port = Integer.parseInt(hostPort[1]); int port = Integer.parseInt(hostPort[1]);
client = new Socket(); //private ReentrantLock lock = new ReentrantLock();
Socket client = new Socket();
client.connect(new InetSocketAddress(host, port)); client.connect(new InetSocketAddress(host, port));
client.setKeepAlive(true); client.setKeepAlive(true);
@ -356,7 +351,7 @@ public class ZHubClient extends AbstractConsumer implements IConsumer, IProducer
/*if (isMain) { /*if (isMain) {
}*/ }*/
if (mainHub.containsValue(this)) { if (mainHub.containsValue(this)) {
buf.append(" " + APP_NAME); buf.append(" ").append(APP_NAME);
} }
for (String topic : getTopics()) { for (String topic : getTopics()) {
buf.append(" ").append(topic); buf.append(" ").append(topic);
@ -364,9 +359,7 @@ public class ZHubClient extends AbstractConsumer implements IConsumer, IProducer
send(buf.toString()); send(buf.toString());
// 重连 timer 订阅 // 重连 timer 订阅
timerMap.forEach((name, timer) -> { timerMap.forEach((name, timer) -> send("timer", name));
send("timer", name);
});
if (retry > 0) { if (retry > 0) {
logger.warning(String.format("ZHubClient[%s][%s] %s Succeed！", getGroupid(), i + 1, retry > 0 ? "reconnection" : "init")); logger.warning(String.format("ZHubClient[%s][%s] %s Succeed！", getGroupid(), i + 1, retry > 0 ? "reconnection" : "init"));
} else { } else {
@ -426,16 +419,12 @@ public class ZHubClient extends AbstractConsumer implements IConsumer, IProducer
} }
} }
if ("M".equals(endchar)) { switch (endchar) {
delay *= (1000 * 60 * 60 * 24 * 30); case "M" -> delay *= (1000 * 60 * 60 * 24 * 30);
} else if ("d".equals(endchar)) { case "d" -> delay *= (1000 * 60 * 60 * 24);
delay *= (1000 * 60 * 60 * 24); case "H" -> delay *= (1000 * 60 * 60);
} else if ("H".equals(endchar)) { case "m" -> delay *= (1000 * 60);
delay *= (1000 * 60 * 60); case "s" -> delay *= 1000;
} else if ("m".equals(endchar)) {
delay *= (1000 * 60);
} else if ("s".equals(endchar)) {
delay *= 1000;
} }
delay(topic, v, delay); delay(topic, v, delay);
@ -452,7 +441,7 @@ public class ZHubClient extends AbstractConsumer implements IConsumer, IProducer
} }
// ================================================== lock ================================================== // ================================================== lock ==================================================
private Map<String, Lock> lockTag = new ConcurrentHashMap<>(); private final Map<String, Lock> lockTag = new ConcurrentHashMap<>();
public Lock tryLock(String key, int duration) { public Lock tryLock(String key, int duration) {
String uuid = UUID.randomUUID().toString().replaceAll("-", ""); String uuid = UUID.randomUUID().toString().replaceAll("-", "");
@ -472,7 +461,7 @@ public class ZHubClient extends AbstractConsumer implements IConsumer, IProducer
} }
// ================================================== timer ================================================== // ================================================== timer ==================================================
private ConcurrentHashMap<String, Timer> timerMap = new ConcurrentHashMap(); private final ConcurrentHashMap<String, Timer> timerMap = new ConcurrentHashMap();
class Timer { class Timer {
String name; String name;
@ -506,8 +495,8 @@ public class ZHubClient extends AbstractConsumer implements IConsumer, IProducer
// ================================================== rpc ================================================== // ================================================== rpc ==================================================
// -- 调用端 -- // -- 调用端 --
private static Map<String, Rpc> rpcMap = new ConcurrentHashMap<>(); private static final Map<String, Rpc> rpcMap = new ConcurrentHashMap<>();
private static Map<String, TypeToken> rpcRetType = new ConcurrentHashMap<>(); private static final Map<String, TypeToken> rpcRetType = new ConcurrentHashMap<>();
// rpc call // rpc call
public RpcResult<Void> rpc(String topic, Object v) { public RpcResult<Void> rpc(String topic, Object v) {
@ -519,9 +508,9 @@ public class ZHubClient extends AbstractConsumer implements IConsumer, IProducer
return rpc(topic, v, typeToken, 0); return rpc(topic, v, typeToken, 0);
} }
@Comment("rpc call") // rpc call
public <T, R> RpcResult<R> rpc(String topic, T v, TypeToken<R> typeToken, long timeout) { public <T, R> RpcResult<R> rpc(String topic, T v, TypeToken<R> typeToken, long timeout) {
Rpc rpc = new Rpc<>(APP_NAME, Utility.uuid(), topic, v); Rpc rpc = new Rpc<>(APP_NAME, UUID.randomUUID().toString().replaceAll("-", ""), topic, v);
String ruk = rpc.getRuk(); String ruk = rpc.getRuk();
rpcMap.put(ruk, rpc); rpcMap.put(ruk, rpc);
if (typeToken != null) { if (typeToken != null) {
@ -602,7 +591,7 @@ public class ZHubClient extends AbstractConsumer implements IConsumer, IProducer
} }
// -- 订阅端 -- // -- 订阅端 --
private Set<String> rpcTopics = new HashSet(); private final HashSet rpcTopics = new HashSet();
// rpc call consumer // rpc call consumer
public <T, R> void rpcSubscribe(String topic, TypeToken<T> typeToken, Function<Rpc<T>, RpcResult<R>> fun) { public <T, R> void rpcSubscribe(String topic, TypeToken<T> typeToken, Function<Rpc<T>, RpcResult<R>> fun) {

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

@ -1,23 +1,19 @@
package net.tccn.timer; package net.tccn.timer;
import net.tccn.timer.scheduled.ScheduledCycle; import net.tccn.timer.scheduled.ScheduledCycle;
import org.redkale.util.Utility;
import java.util.UUID;
import java.util.function.Supplier; import java.util.function.Supplier;
public class Timers { 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) { 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) { if (supplier.get() || task.getExecCount() == maxCount) {
task.setComplete(true); task.setComplete(true);
} }
@ -32,12 +28,9 @@ public class Timers {
/** /**
* 本地延时：延时时间极短的场景下使用 （如：1分钟内） * 本地延时：延时时间极短的场景下使用 （如：1分钟内）
*
* @param runnable
* @param millis
*/ */
public static void delay(Runnable runnable, long 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(); runnable.run();
task.setComplete(true); task.setComplete(true);
})); }));

2
test/HelloService.java
View File

@ -58,6 +58,4 @@ public class HelloService {
/*RpcResult<FileToken> x = zhub.rpc("rpc:file:up-token", Map.of(), new TypeToken<>() { /*RpcResult<FileToken> x = zhub.rpc("rpc:file:up-token", Map.of(), new TypeToken<>() {
});*/ });*/
} }