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

Compare commits

base: lxy:e2c2b626653c879e2b395eba3e0982c5d81fe788
Branches Tags
lxy:master
lxy:dev/redkale-2.2
lxy:dev/zhub-cli
lxy:dev/springboot-1.8
lxy:dev/redkale
lxy:dev/redkale-x
lxy:dev/springboot-17
lxy:dev/springboot
..
compare: lxy:f4771aadf28a000c3e0bcf1b12a885104f52079f
Branches Tags
lxy:dev/redkale-2.2
lxy:dev/zhub-cli
lxy:dev/springboot-1.8
lxy:dev/redkale
lxy:dev/redkale-x
lxy:dev/springboot-17
lxy:dev/springboot
lxy:master

No commits in common. "e2c2b626653c879e2b395eba3e0982c5d81fe788" and "f4771aadf28a000c3e0bcf1b12a885104f52079f" have entirely different histories.
e2c2b62665 ... f4771aadf2

3 changed files with 70 additions and 84 deletions
Show Stats Expand all files Collapse all files

12
pom.xml
View File

@ -5,8 +5,8 @@
<modelVersion>4.0.0</modelVersion> <modelVersion>4.0.0</modelVersion>
<groupId>net.tccn</groupId> <groupId>net.tccn</groupId>
<artifactId>zhub-client-spring</artifactId> <artifactId>zhub-cli</artifactId>
<version>0.1.1</version> <version>1.0</version>
<properties> <properties>
<maven.compiler.source>1.8</maven.compiler.source> <maven.compiler.source>1.8</maven.compiler.source>
@ -33,11 +33,11 @@
</dependency> </dependency>
</dependencies> </dependencies>
<repositories> <!--<repositories>
<repository> <repository>
<id>maven-nexus</id> <id>maven-nexus</id>
<name>maven-nexus</name> <name>maven-nexus</name>
<url>https://nexus.1216.top/repository/maven-public/</url> <url>http://47.106.237.198:8081/repository/maven-public/</url>
</repository> </repository>
</repositories> </repositories>
@ -46,8 +46,8 @@
<repository> <repository>
<id>mvn-release</id> <id>mvn-release</id>
<name>mvn-release</name> <name>mvn-release</name>
<url>https://nexus.1216.top/repository/maven-releases/</url> <url>http://47.106.237.198:8081/repository/maven-releases/</url>
</repository> </repository>
</distributionManagement> </distributionManagement>-->
</project> </project>

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

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

128
src/main/java/tccn/zhub/ZHubClient.java
View File

@ -21,14 +21,15 @@ import java.util.HashMap;
import java.util.HashSet; import java.util.HashSet;
import java.util.Map; import java.util.Map;
import java.util.UUID; import java.util.UUID;
import java.util.concurrent.*; import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.function.BiConsumer;
import java.util.function.Consumer; import java.util.function.Consumer;
import java.util.function.Function; 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;
@Component @Component
public class ZHubClient extends AbstractConsumer implements IConsumer, IProducer { public class ZHubClient extends AbstractConsumer implements IConsumer, IProducer {
@ -59,6 +60,12 @@ 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 final BiConsumer<Runnable, Integer> threadBuilder = (r, n) -> {
for (int i = 0; i < n; i++) {
new Thread(() -> r.run()).start();
}
};
/*private static boolean isFirst = true; /*private static boolean isFirst = true;
private boolean isMain = false;*/ private boolean isMain = false;*/
private static final 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
@ -205,119 +212,98 @@ public class ZHubClient extends AbstractConsumer implements IConsumer, IProducer
}).start(); }).start();
// 定时调度事件 // 定时调度事件
new Thread(() -> { threadBuilder.accept(() -> {
ExecutorService executor = Executors.newSingleThreadExecutor();
while (true) { while (true) {
Timer timer = null; Timer timer = null;
try { try {
timer = timerQueue.take(); if ((timer = timerQueue.take()) == null) {
return;
}
long start = System.currentTimeMillis(); long start = System.currentTimeMillis();
executor.submit(timer.runnable).get(5, TimeUnit.SECONDS); timer.runnable.run();
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 | ExecutionException | TimeoutException e) { } catch (InterruptedException e) {
if (e instanceof TimeoutException) {
executor = Executors.newSingleThreadExecutor();
logger.log(Level.WARNING, "TimeoutException [" + timer.name + "]", e);
} else {
e.printStackTrace(); e.printStackTrace();
} catch (Exception e) {
logger.log(Level.WARNING, "timer [" + timer.name + "]", e);
} }
} }
} }, 1);
}).start();
// topic msg // topic msg
new Thread(() -> { threadBuilder.accept(() -> {
ExecutorService executor = Executors.newSingleThreadExecutor();
while (true) { while (true) {
Event<String> event = null; Event<String> event = null;
try { try {
event = topicQueue.take(); if ((event = topicQueue.take()) == null) {
continue;
}
logger.log(Level.FINE, "topic[" + event.topic + "] :" + event.value); logger.log(Level.FINE, "topic[" + event.topic + "] :" + event.value);
accept(event.topic, event.value);
String topic = event.topic; } catch (InterruptedException e) {
String value = event.value; e.printStackTrace();
executor.submit(() -> accept(topic, value)).get(5, TimeUnit.SECONDS); } catch (Exception e) {
} catch (InterruptedException | ExecutionException | TimeoutException e) {
if (e instanceof TimeoutException) {
executor = Executors.newSingleThreadExecutor();
logger.log(Level.WARNING, "TimeoutException, topic[" + event.topic + "], value[" + event.value + "]", e);
} else if (event != null) {
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);
} }
} }
} }, 1);
}, "ZHub-topic-accept").start();
// rpc back // rpc back
new Thread(() -> { threadBuilder.accept(() -> {
ExecutorService executor = Executors.newSingleThreadExecutor();
while (true) { while (true) {
Event<String> event = null; Event<String> event = null;
try { try {
event = rpcBackQueue.take(); if ((event = rpcBackQueue.take()) == null) {
continue;
}
//if (event)
logger.info(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); rpcAccept(event.value);
} catch (InterruptedException e) {
String value = event.value; e.printStackTrace();
executor.submit(() -> rpcAccept(value)).get(5, TimeUnit.SECONDS); } catch (Exception e) {
} catch (InterruptedException | ExecutionException | TimeoutException e) {
if (e instanceof TimeoutException) {
executor = Executors.newSingleThreadExecutor();
logger.log(Level.WARNING, "rpc-back TimeoutException, topic[" + event.topic + "], value[" + event.value + "]", e);
} else if (event != null) {
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 :" + event.value, e);
} }
throw new RuntimeException(e);
} }
} }, 1);
}).start();
// rpc call // rpc call
new Thread(() -> { threadBuilder.accept(() -> {
ExecutorService executor = Executors.newSingleThreadExecutor();
while (true) { while (true) {
Event<String> event = null; Event<String> event = null;
try { try {
event = rpcCallQueue.take(); if ((event = rpcCallQueue.take()) == null) {
continue;
}
logger.info(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);
String topic = event.topic; } catch (InterruptedException e) {
String value = event.value; e.printStackTrace();
executor.submit(() -> accept(topic, value)).get(5, TimeUnit.SECONDS); } catch (Exception e) {
} 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); logger.log(Level.WARNING, "rpc-call[" + event.topic + "] event accept error :" + event.value, e);
} }
} }
} }, 1);
}, "ZHub-rpc-call").start();
// send msg // send msg
new Thread(() -> { threadBuilder.accept(() -> {
while (true) { while (true) {
String msg = null; String msg = null;
try { try {
msg = sendMsgQueue.take(); if ((msg = sendMsgQueue.take()) == null) {
writer.write(msg.getBytes(UTF_8)); continue;
}
// logger.log(Level.FINEST, "send-msg: [" + msg + "]");
writer.write(msg.getBytes());
writer.flush(); writer.flush();
} catch (InterruptedException | IOException e) { } catch (InterruptedException e) {
logger.log(Level.WARNING, "send-msg[" + msg + "] event accept error :", e);
try {
Thread.sleep(3000);
assert msg != null;
writer.write(msg.getBytes(UTF_8));
writer.flush();
} catch (IOException | InterruptedException | NullPointerException ex) {
e.printStackTrace(); e.printStackTrace();
} catch (Exception e) {
logger.log(Level.WARNING, "send-msg[" + msg + "] event accept error :", e);
} }
} }
} }, 1);
}).start();
} }