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

.

Browse Source
This commit is contained in:
绝尘 2024-01-19 04:01:38 +08:00
parent 71ccfc919b
commit 11ed5d2018
2 changed files with 66 additions and 60 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

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

@ -25,6 +25,8 @@ import java.util.function.Function;
import java.util.logging.Level; import java.util.logging.Level;
import java.util.logging.Logger; import java.util.logging.Logger;
import static java.nio.charset.StandardCharsets.UTF_8;
@Local @Local
@AutoLoad(false) @AutoLoad(false)
@ResourceType(ZHubClient.class) @ResourceType(ZHubClient.class)
@ -221,110 +223,114 @@ public class ZHubClient extends AbstractConsumer implements IConsumer, IProducer
}).thenAcceptAsync(x -> { }).thenAcceptAsync(x -> {
// 定时调度事件，已加入耗时监控 // 定时调度事件，已加入耗时监控
new Thread(() -> { new Thread(() -> {
ExecutorService pool = Executors.newFixedThreadPool(1); ExecutorService executor = Executors.newSingleThreadExecutor();
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); executor.submit(timer.runnable).get(5, TimeUnit.SECONDS);
long end = System.currentTimeMillis(); long end = System.currentTimeMillis();
logger.finest(String.format("timer [%s] : elapsed time %s ms", timer.name, end - start)); logger.finest(String.format("timer [%s] : elapsed time %s ms", timer.name, end - start));
} catch (InterruptedException e) { } catch (InterruptedException | ExecutionException | TimeoutException e) {
if (e instanceof TimeoutException) {
executor = Executors.newSingleThreadExecutor();
logger.log(Level.WARNING, "TimeoutException [" + timer.name + "]", e);
} else {
e.printStackTrace(); e.printStackTrace();
} catch (TimeoutException e) { }
logger.log(Level.SEVERE, "timer [" + timer.name + "] time out: " + 5 + " S", e);
pool = Executors.newFixedThreadPool(1);
} catch (Exception e) {
logger.log(Level.WARNING, "timer [" + timer.name + "]", e);
} }
} }
}).start(); }).start();
// topic msg，已加入耗时监控 // topic msg，已加入耗时监控
new Thread(() -> { new Thread(() -> {
ExecutorService pool = Executors.newFixedThreadPool(1); ExecutorService executor = Executors.newSingleThreadExecutor();
while (true) { while (true) {
Event<String> event = null; Event<String> event = null;
try { try {
if ((event = topicQueue.take()) == null) { event = topicQueue.take();
continue; logger.log(Level.FINE, "topic[" + event.topic + "] :" + event.value);
}
String topic = event.topic; String topic = event.topic;
String value = event.value; String value = event.value;
pool.submit(() -> accept(topic, value)).get(5, TimeUnit.SECONDS); executor.submit(() -> accept(topic, value)).get(5, TimeUnit.SECONDS);
} catch (InterruptedException e) { } catch (InterruptedException | ExecutionException | TimeoutException e) {
e.printStackTrace(); if (e instanceof TimeoutException) {
} catch (TimeoutException e) { executor = Executors.newSingleThreadExecutor();
logger.log(Level.SEVERE, "topic[" + event.topic + "] event deal time out: " + 5 + " S, value: " + event.value, e); logger.log(Level.WARNING, "TimeoutException, topic[" + event.topic + "], value[" + event.value + "]", e);
pool = Executors.newFixedThreadPool(1); } else if (event != null) {
} 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 :" + event.value, e);
} }
} }
}).start(); }
}, "ZHub-topic-accept").start();
// rpc back ,仅做数据解析，暂无耗时监控 // rpc back ,仅做数据解析，暂无耗时监控
new Thread(() -> { new Thread(() -> {
ExecutorService executor = Executors.newSingleThreadExecutor();
while (true) { while (true) {
Event<String> event = null; Event<String> event = null;
try { try {
if ((event = rpcBackQueue.take()) == null) { event = rpcCallQueue.take();
continue; logger.info(String.format("rpc-call:[%s] %s", event.topic, event.value));
}
//if (event)
logger.finest(String.format("rpc-back:[%s]: %s", event.topic, event.value));
rpcAccept(event.value);
} catch (InterruptedException e) {
e.printStackTrace();
} catch (Exception e) {
logger.log(Level.WARNING, "rpc-back[" + event.topic + "] event accept error :" + event.value, e);
}
}
}).start();
// rpc call，已加入耗时监控
new Thread(() -> {
ExecutorService pool = Executors.newFixedThreadPool(1);
while (true) {
Event<String> event = null;
try {
if ((event = rpcCallQueue.take()) == null) {
continue;
}
logger.finest(String.format("rpc-call:[%s] %s", event.topic, event.value));
String topic = event.topic; String topic = event.topic;
String value = event.value; String value = event.value;
pool.submit(() -> accept(topic, value)).get(5, TimeUnit.SECONDS); executor.submit(() -> accept(topic, value)).get(5, TimeUnit.SECONDS);
} catch (InterruptedException e) { } catch (InterruptedException | ExecutionException | TimeoutException e) {
e.printStackTrace(); if (e instanceof TimeoutException) {
} catch (TimeoutException e) { executor = Executors.newSingleThreadExecutor();
logger.log(Level.SEVERE, "topic[" + event.topic + "] event deal time out: " + 5 + " S, value: " + event.value, e); logger.log(Level.WARNING, "rpc-call TimeoutException, topic[" + event.topic + "], value[" + event.value + "]", e);
pool = Executors.newFixedThreadPool(1); } else if (event != null) {
} 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 :" + event.value, e);
} }
} }
}).start(); }
}, "ZHub-rpc-call").start();
// rpc call，已加入耗时监控
new Thread(() -> {
ExecutorService executor = Executors.newSingleThreadExecutor();
while (true) {
Event<String> event = null;
try {
event = rpcCallQueue.take();
logger.info(String.format("rpc-call:[%s] %s", event.topic, event.value));
String topic = event.topic;
String value = event.value;
executor.submit(() -> accept(topic, value)).get(5, TimeUnit.SECONDS);
} catch (InterruptedException | ExecutionException | TimeoutException e) {
if (e instanceof TimeoutException) {
executor = Executors.newSingleThreadExecutor();
logger.log(Level.WARNING, "rpc-call TimeoutException, topic[" + event.topic + "], value[" + event.value + "]", e);
} else if (event != null) {
logger.log(Level.WARNING, "rpc-call[" + event.topic + "] event accept error :" + event.value, e);
}
}
}
}, "ZHub-rpc-call").start();
// send msg // send msg
new Thread(() -> { new Thread(() -> {
while (true) { while (true) {
String msg = null; String msg = null;
try { try {
if ((msg = sendMsgQueue.take()) == null) { msg = sendMsgQueue.take();
continue; writer.write(msg.getBytes(UTF_8));
}
// logger.log(Level.FINEST, "send-msg: [" + msg + "]");
writer.write(msg.getBytes());
writer.flush(); writer.flush();
} catch (InterruptedException e) { } catch (InterruptedException | IOException e) {
e.printStackTrace();
} catch (Exception e) {
logger.log(Level.WARNING, "send-msg[" + msg + "] event accept error :", e); logger.log(Level.WARNING, "send-msg[" + msg + "] event accept error :", e);
try {
Thread.sleep(5000);
assert msg != null;
writer.write(msg.getBytes(UTF_8));
writer.flush();
} catch (IOException | InterruptedException | NullPointerException ex) {
e.printStackTrace();
}
} }
} }
}).start(); }).start();
@ -423,7 +429,7 @@ public class ZHubClient extends AbstractConsumer implements IConsumer, IProducer
client.setKeepAlive(true); client.setKeepAlive(true);
writer = client.getOutputStream(); writer = client.getOutputStream();
reader = new BufferedReader(new InputStreamReader(client.getInputStream())); reader = new BufferedReader(new InputStreamReader(client.getInputStream(), UTF_8));
String groupid = getGroupid(); String groupid = getGroupid();
if (groupid == null || groupid.isEmpty()) { if (groupid == null || groupid.isEmpty()) {