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

优化：rpc调用优先调用本地订阅

Browse Source
This commit is contained in:
绝尘 2024-03-28 00:22:43 +08:00
parent 2e0d9af014
commit ff3d877a37
7 changed files with 106 additions and 86 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

35
pom.xml
View File

@ -14,11 +14,26 @@
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding> <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
</properties> </properties>
<repositories>
<repository>
<id>maven-release</id>
<name>maven-nexus</name>
<url>https://nexus.1216.top/repository/maven-public/</url>
</repository>
</repositories>
<distributionManagement>
<repository>
<id>mvn-release</id>
<name>mvn-release</name>
<url>https://nexus.1216.top/repository/maven-releases/</url>
</repository>
</distributionManagement>
<dependencies> <dependencies>
<dependency> <dependency>
<groupId>org.redkale</groupId> <groupId>org.redkale</groupId>
<artifactId>redkale</artifactId> <artifactId>redkale</artifactId>
<version>2.8.0-dev</version> <version>2.8.0.dev</version>
<scope>compile</scope> <scope>compile</scope>
</dependency> </dependency>
<dependency> <dependency>
@ -28,22 +43,4 @@
<scope>compile</scope> <scope>compile</scope>
</dependency> </dependency>
</dependencies> </dependencies>
<repositories>
<repository>
<id>maven-nexus</id>
<name>maven-nexus</name>
<url>http://47.106.237.198:8081/repository/maven-public/</url>
</repository>
</repositories>
<distributionManagement>
<repository>
<id>mvn-release</id>
<name>mvn-release</name>
<url>http://47.106.237.198:8081/repository/maven-releases/</url>
</repository>
</distributionManagement>
</project> </project>

18
src/main/java/net/tccn/AbstractConsumer.java
View File

@ -19,7 +19,7 @@ public abstract class AbstractConsumer extends ZhubAgentProvider implements ICon
protected static String APP_NAME = ""; protected static String APP_NAME = "";
private Map<String, EventType> eventMap = new ConcurrentHashMap<>(); protected Map<String, EventType<?>> eventMap = new ConcurrentHashMap<>();
protected abstract String getGroupid(); protected abstract String getGroupid();
@ -31,6 +31,7 @@ public abstract class AbstractConsumer extends ZhubAgentProvider implements ICon
return Set.of("-"); return Set.of("-");
} }
// topic 消息消费前处理
protected void accept(String topic, String value) { protected void accept(String topic, String value) {
EventType eventType = eventMap.get(topic); EventType eventType = eventMap.get(topic);
@ -44,6 +45,12 @@ public abstract class AbstractConsumer extends ZhubAgentProvider implements ICon
eventType.accept(data); eventType.accept(data);
} }
// rpc 被调用端
protected <T> void rpcAccept(String topic, T value) {
EventType eventType = eventMap.get(topic);
eventType.accept(value);
}
protected final void removeEventType(String topic) { protected final void removeEventType(String topic) {
eventMap.remove(topic); eventMap.remove(topic);
} }
@ -77,4 +84,13 @@ public abstract class AbstractConsumer extends ZhubAgentProvider implements ICon
public String resourceName() { public String resourceName() {
return super.getName(); return super.getName();
} }
protected String toStr(Object v) {
if (v instanceof String) {
return (String) v;
} else if (v == null) {
return null;
}
return convert.convertTo(v);
}
} }

4
src/main/java/net/tccn/Event.java
View File

@ -15,8 +15,8 @@ public class Event<V> {
this.value = value; this.value = value;
} }
public static <V> Event of(String topic, V value) { public static <V> Event<V> of(String topic, V value) {
return new Event<V>(topic, value); return new Event<>(topic, value);
} }

2
src/main/java/net/tccn/timer/TimerExecutor.java
View File

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

6
src/main/java/net/tccn/timer/TimerTask.java
View File

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

5
src/main/java/net/tccn/zhub/Rpc.java
View File

@ -1,7 +1,6 @@
package net.tccn.zhub; package net.tccn.zhub;
import org.redkale.convert.ConvertColumn; import org.redkale.convert.ConvertColumn;
import org.redkale.convert.json.JsonConvert;
import org.redkale.service.RetResult; import org.redkale.service.RetResult;
public class Rpc<T> { public class Rpc<T> {
@ -14,10 +13,10 @@ public class Rpc<T> {
public Rpc() { public Rpc() {
} }
protected Rpc(String appname, String ruk, String topic, Object value) { protected Rpc(String appname, String ruk, String topic, T value) {
this.ruk = appname + "::" + ruk; this.ruk = appname + "::" + ruk;
this.topic = topic; this.topic = topic;
this.value = (T) JsonConvert.root().convertTo(value); this.value = value;
} }
public String getRuk() { public String getRuk() {

122
src/main/java/net/tccn/zhub/ZHubClient.java
View File

@ -41,17 +41,11 @@ public class ZHubClient extends AbstractConsumer implements IConsumer, IProducer
private BufferedReader reader; private BufferedReader reader;
private final LinkedBlockingQueue<Timer> timerQueue = new LinkedBlockingQueue<>(); private final LinkedBlockingQueue<Timer> timerQueue = new LinkedBlockingQueue<>();
private final LinkedBlockingQueue<Event<String>> topicQueue = new LinkedBlockingQueue<>(); private final LinkedBlockingQueue<Event<String>> topicQueue = new LinkedBlockingQueue<>(); // [=> Object]
private final LinkedBlockingQueue<Event<String>> rpcBackQueue = new LinkedBlockingQueue<>(); // RPC BACK MSG private final LinkedBlockingQueue<Event<Object>> rpcBackQueue = new LinkedBlockingQueue<>(); // RPC BACK MSG [=> Object]
private final LinkedBlockingQueue<Event<String>> rpcCallQueue = new LinkedBlockingQueue<>(); // RPC CALL MSG private final LinkedBlockingQueue<Event<Object>> rpcCallQueue = new LinkedBlockingQueue<>(); // RPC CALL MSG [=> Object]
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) -> {
for (int i = 0; i < n; i++) {
new Thread(() -> r.run()).start();
}
};*/
private static Map<String, ZHubClient> mainHub = new HashMap<>(); // 127.0.0.1:1216 - ZHubClient private static Map<String, ZHubClient> mainHub = new HashMap<>(); // 127.0.0.1:1216 - ZHubClient
public ZHubClient() { public ZHubClient() {
@ -145,7 +139,7 @@ public class ZHubClient extends AbstractConsumer implements IConsumer, IProducer
String value = ""; String value = "";
do { do {
if (value.length() > 0) { if (!value.isEmpty()) {
value += "\r\n"; value += "\r\n";
} }
String s = reader.readLine(); String s = reader.readLine();
@ -225,9 +219,8 @@ public class ZHubClient extends AbstractConsumer implements IConsumer, IProducer
while (true) { while (true) {
Timer timer = null; Timer timer = null;
try { try {
if ((timer = timerQueue.take()) == null) { timer = timerQueue.take();
return;
}
long start = System.currentTimeMillis(); long start = System.currentTimeMillis();
pool.submit(timer.runnable).get(5, TimeUnit.SECONDS); pool.submit(timer.runnable).get(5, TimeUnit.SECONDS);
long end = System.currentTimeMillis(); long end = System.currentTimeMillis();
@ -249,9 +242,7 @@ public class ZHubClient extends AbstractConsumer implements IConsumer, IProducer
while (true) { while (true) {
Event<String> event = null; Event<String> event = null;
try { try {
if ((event = topicQueue.take()) == null) { event = topicQueue.take();
continue;
}
String topic = event.topic; String topic = event.topic;
String value = event.value; String value = event.value;
@ -259,10 +250,10 @@ public class ZHubClient extends AbstractConsumer implements IConsumer, IProducer
} catch (InterruptedException e) { } catch (InterruptedException e) {
e.printStackTrace(); e.printStackTrace();
} catch (TimeoutException e) { } catch (TimeoutException e) {
logger.log(Level.SEVERE, "topic[" + event.topic + "] event deal time out: " + 5 + " S, value: " + event.value, e); logger.log(Level.SEVERE, "topic[" + event.topic + "] event deal time out: " + 5 + " S, value: " + toStr(event.value), e);
pool = Executors.newFixedThreadPool(1); pool = Executors.newFixedThreadPool(1);
} catch (Exception e) { } catch (Exception e) {
logger.log(Level.WARNING, "topic[" + event.topic + "] event accept error :" + event.value, e); logger.log(Level.WARNING, "topic[" + event.topic + "] event accept error :" + toStr(event.value), e);
} }
} }
}).start(); }).start();
@ -270,18 +261,16 @@ public class ZHubClient extends AbstractConsumer implements IConsumer, IProducer
// rpc back ,仅做数据解析，暂无耗时监控 // rpc back ,仅做数据解析，暂无耗时监控
new Thread(() -> { new Thread(() -> {
while (true) { while (true) {
Event<String> event = null; Event<Object> event = null;
try { try {
if ((event = rpcBackQueue.take()) == null) { event = rpcBackQueue.take();
continue;
}
//if (event) //if (event)
logger.finest(String.format("rpc-back:[%s]: %s", event.topic, event.value)); logger.finest(String.format("rpc-back:[%s]: %s", event.topic, toStr(event.value)));
rpcAccept(event.value); rpcAccept(event.value);
} catch (InterruptedException e) { } catch (InterruptedException e) {
e.printStackTrace(); e.printStackTrace();
} catch (Exception e) { } catch (Exception e) {
logger.log(Level.WARNING, "rpc-back[" + event.topic + "] event accept error :" + event.value, e); logger.log(Level.WARNING, "rpc-back[" + event.topic + "] event accept error :" + toStr(event.value), e);
} }
} }
}).start(); }).start();
@ -290,22 +279,21 @@ public class ZHubClient extends AbstractConsumer implements IConsumer, IProducer
new Thread(() -> { new Thread(() -> {
ExecutorService pool = Executors.newFixedThreadPool(1); ExecutorService pool = Executors.newFixedThreadPool(1);
while (true) { while (true) {
Event<String> event = null; Event<Object> event = null;
try { try {
if ((event = rpcCallQueue.take()) == null) { event = rpcCallQueue.take();
continue;
} logger.finest(String.format("rpc-call:[%s] %s", event.topic, toStr(event.value)));
logger.finest(String.format("rpc-call:[%s] %s", event.topic, event.value));
String topic = event.topic; String topic = event.topic;
String value = event.value; Object value = event.value;
pool.submit(() -> accept(topic, value)).get(5, TimeUnit.SECONDS); pool.submit(() -> rpcAccept(topic, value)).get(5, TimeUnit.SECONDS);
} catch (InterruptedException e) { } catch (InterruptedException e) {
e.printStackTrace(); e.printStackTrace();
} catch (TimeoutException e) { } catch (TimeoutException e) {
logger.log(Level.SEVERE, "topic[" + event.topic + "] event deal time out: " + 5 + " S, value: " + event.value, e); logger.log(Level.SEVERE, "topic[" + event.topic + "] event deal time out: " + 5 + " S, value: " + toStr(event.value), e);
pool = Executors.newFixedThreadPool(1); pool = Executors.newFixedThreadPool(1);
} catch (Exception e) { } catch (Exception e) {
logger.log(Level.WARNING, "rpc-call[" + event.topic + "] event accept error :" + event.value, e); logger.log(Level.WARNING, "rpc-call[" + event.topic + "] event accept error :" + toStr(event.value), e);
} }
} }
}).start(); }).start();
@ -315,9 +303,8 @@ public class ZHubClient extends AbstractConsumer implements IConsumer, IProducer
while (true) { while (true) {
String msg = null; String msg = null;
try { try {
if ((msg = sendMsgQueue.take()) == null) { msg = sendMsgQueue.take();
continue;
}
// logger.log(Level.FINEST, "send-msg: [" + msg + "]"); // logger.log(Level.FINEST, "send-msg: [" + msg + "]");
writer.write(msg.getBytes()); writer.write(msg.getBytes());
writer.flush(); writer.flush();
@ -402,15 +389,6 @@ public class ZHubClient extends AbstractConsumer implements IConsumer, IProducer
return str.length(); return str.length();
} }
private String toStr(Object v) {
if (v instanceof String) {
return (String) v;
} else if (v == null) {
return null;
}
return convert.convertTo(v);
}
protected boolean initSocket(int retry) { protected boolean initSocket(int retry) {
for (int i = 0; i <= retry; i++) { for (int i = 0; i <= retry; i++) {
try { try {
@ -479,11 +457,15 @@ public class ZHubClient extends AbstractConsumer implements IConsumer, IProducer
} }
public boolean publish(String topic, Object v) { public boolean publish(String topic, Object v) {
/*if (eventMap.containsKey(topic)) { // 本地调用
topicQueue.add(Event.of(topic, v));
return true;
}*/
return send("publish", topic, toStr(v)); return send("publish", topic, toStr(v));
} }
public void broadcast(String topic, Object v) { public void broadcast(String topic, Object v) {
send("broadcast", topic, toStr(v)); send("broadcast", topic, toStr(v)); // 广播必须走远端模式
} }
// 发送 publish 主题消息，若多次发送的 topic + "-" + value 相同，将会做延时重置 // 发送 publish 主题消息，若多次发送的 topic + "-" + value 相同，将会做延时重置
@ -633,7 +615,12 @@ public class ZHubClient extends AbstractConsumer implements IConsumer, IProducer
rpcRetType.put(ruk, typeToken); rpcRetType.put(ruk, typeToken);
} }
try { try {
publish(topic, rpc); // send("rpc", topic, toStr(rpc)); if (eventMap.containsKey(topic)) { // 本地调用
rpcCallQueue.add(Event.of(topic, rpc));
} else {
rpc.setValue(toStr(rpc.getValue()));
publish(topic, rpc); // send("rpc", topic, toStr(rpc));
}
synchronized (rpc) { synchronized (rpc) {
if (timeout <= 0) { if (timeout <= 0) {
timeout = 1000 * 15; timeout = 1000 * 15;
@ -683,9 +670,20 @@ public class ZHubClient extends AbstractConsumer implements IConsumer, IProducer
// RpcResult: {ruk:xxx-xxxx, retcode:0} // RpcResult: {ruk:xxx-xxxx, retcode:0}
@Comment("rpc call back consumer") @Comment("rpc call back consumer")
private void rpcAccept(String value) { private <T> void rpcAccept(T value) {
// 接收到 本地调用返回的 RpcResult TODO, 本地模式下返回的数据对象类型需要和处理端一致，不然会出现类型转换异常 - 解决办法，当出现不一致的情况取数据做转换
if (value instanceof RpcResult) {
String ruk = ((RpcResult) value).getRuk();
Rpc rpc = rpcMap.remove(ruk);
rpc.setRpcResult((RpcResult) value);
synchronized (rpc) {
rpc.notify();
}
return;
}
RpcResult resp = convert.convertFrom(new TypeToken<RpcResult<String>>() { RpcResult resp = convert.convertFrom(new TypeToken<RpcResult<String>>() {
}.getType(), value); }.getType(), (String) value);
String ruk = resp.getRuk(); String ruk = resp.getRuk();
Rpc rpc = rpcMap.remove(ruk); Rpc rpc = rpcMap.remove(ruk);
@ -716,18 +714,28 @@ public class ZHubClient extends AbstractConsumer implements IConsumer, IProducer
@Comment("rpc call consumer") @Comment("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) {
Consumer<String> consumer = v -> { Consumer<T> consumer = v -> {
Rpc<T> rpc = null; Rpc<T> rpc = null;
try { try {
rpc = convert.convertFrom(new TypeToken<Rpc<String>>() { if (v instanceof String) {
}.getType(), v); rpc = convert.convertFrom(IType.STRING.getType(), (String) v);
} else {
rpc = (Rpc<T>) v;
}
// 参数转换 // 参数转换
T paras = convert.convertFrom(typeToken.getType(), (String) rpc.getValue()); if (rpc.getValue() instanceof String && !"java.lang.String".equals(typeToken.getType().getTypeName())) {
rpc.setValue(paras); T paras = convert.convertFrom(typeToken.getType(), (String) rpc.getValue());
rpc.setValue(paras);
}
RpcResult result = fun.apply(rpc); RpcResult result = fun.apply(rpc);
result.setResult(toStr(result.getResult())); if (APP_NAME.equals(rpc.getBackTopic())) {
publish(rpc.getBackTopic(), result); rpcBackQueue.add(Event.of(topic, result));
} else {
result.setResult(toStr(result.getResult())); // 远程模式 结果转换
publish(rpc.getBackTopic(), result);
}
} catch (Exception e) { } catch (Exception e) {
logger.log(Level.WARNING, "rpc call consumer error: " + v, e); logger.log(Level.WARNING, "rpc call consumer error: " + v, e);
publish(rpc.getBackTopic(), rpc.retError("服务调用失败！")); publish(rpc.getBackTopic(), rpc.retError("服务调用失败！"));
@ -736,6 +744,6 @@ public class ZHubClient extends AbstractConsumer implements IConsumer, IProducer
}; };
rpcTopics.add(topic); rpcTopics.add(topic);
subscribe(topic, consumer); subscribe(topic, typeToken, consumer);
} }
} }