This is an automated email from the ASF dual-hosted git repository.
jonyang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-eventmesh.git
The following commit(s) were added to refs/heads/master by this push:
new 0eda1e5e3 [ISSUE #3078]Refactor ThreadUtils (#3079)
0eda1e5e3 is described below
commit 0eda1e5e3014f9b32530eaab4a45aac3c9e81f9d
Author: mxsm <[email protected]>
AuthorDate: Sat Feb 11 20:26:17 2023 +0800
[ISSUE #3078]Refactor ThreadUtils (#3079)
* [ISSUE #3078]Refactor ThreadUtils
* polish code
---
.../apache/eventmesh/common/utils/ThreadUtils.java | 33 ++++++++++++++++++----
.../runtime/boot/AbstractRemotingServer.java | 5 +++-
.../client/group/ClientSessionGroupMapping.java | 2 +-
.../eventmesh/runtime/demo/AsyncPubClient.java | 2 +-
.../eventmesh/runtime/demo/BroadCastPubClient.java | 2 +-
5 files changed, 35 insertions(+), 9 deletions(-)
diff --git
a/eventmesh-common/src/main/java/org/apache/eventmesh/common/utils/ThreadUtils.java
b/eventmesh-common/src/main/java/org/apache/eventmesh/common/utils/ThreadUtils.java
index af69f0850..28a8121ca 100644
---
a/eventmesh-common/src/main/java/org/apache/eventmesh/common/utils/ThreadUtils.java
+++
b/eventmesh-common/src/main/java/org/apache/eventmesh/common/utils/ThreadUtils.java
@@ -20,20 +20,43 @@ package org.apache.eventmesh.common.utils;
import org.apache.logging.log4j.util.ProcessIdUtil;
import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.TimeUnit;
public class ThreadUtils {
private static volatile long currentPID = -1;
- public static void randomSleep(int min, int max) throws Exception {
+ public static void randomPause(long min, long max) {
+ randomPause(min, max, TimeUnit.MILLISECONDS);
+ }
+
+ public static void randomPause(long min, long max, TimeUnit timeUnit) {
// nextInt is normally exclusive of the top value, so add 1 to make it
inclusive
- int random = ThreadLocalRandom.current().nextInt(min, max + 1);
- Thread.sleep(random);
+ try {
+ long timeout = ThreadLocalRandom.current().nextLong(min, max + 1);
+ timeUnit.sleep(timeout);
+ } catch (InterruptedException ignore) {
+ //ignore
+ }
+ }
+ public static void randomPause(long max) {
+ randomPause(1, max);
}
- public static void randomSleep(int max) throws Exception {
- randomSleep(1, max);
+ public static void sleep(long timeout) {
+ sleep(timeout, TimeUnit.MILLISECONDS);
+ }
+
+ public static void sleep(long timeout, TimeUnit timeUnit) {
+ if (null == timeUnit) {
+ return;
+ }
+ try {
+ timeUnit.sleep(timeout);
+ } catch (InterruptedException ignore) {
+ //ignore
+ }
}
/**
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/AbstractRemotingServer.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/AbstractRemotingServer.java
index 603990025..18427fe6b 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/AbstractRemotingServer.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/AbstractRemotingServer.java
@@ -20,12 +20,15 @@ package org.apache.eventmesh.runtime.boot;
import org.apache.eventmesh.common.EventMeshThreadFactory;
import org.apache.eventmesh.common.utils.ThreadUtils;
+import java.util.concurrent.TimeUnit;
+
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
+
public abstract class AbstractRemotingServer {
private static final Logger LOGGER =
LoggerFactory.getLogger(AbstractRemotingServer.class);
@@ -102,7 +105,7 @@ public abstract class AbstractRemotingServer {
}
}
- ThreadUtils.randomSleep(DEFAULT_SLEEP_SECONDS);
+
ThreadUtils.randomPause(TimeUnit.SECONDS.toMillis(DEFAULT_SLEEP_SECONDS));
if (ioGroup != null) {
ioGroup.shutdownGracefully();
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/group/ClientSessionGroupMapping.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/group/ClientSessionGroupMapping.java
index 02fab45b8..2ebc2aa64 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/group/ClientSessionGroupMapping.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/group/ClientSessionGroupMapping.java
@@ -446,7 +446,7 @@ public class ClientSessionGroupMapping {
log.error("say goodbye to session error! {}", itr, e);
}
});
- ThreadUtils.randomSleep(50);
+ ThreadUtils.randomPause(50);
log.info("ClientSessionGroupMapping shutdown......");
}
diff --git
a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/demo/AsyncPubClient.java
b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/demo/AsyncPubClient.java
index afda4a105..b15586a08 100644
---
a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/demo/AsyncPubClient.java
+++
b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/demo/AsyncPubClient.java
@@ -49,7 +49,7 @@ public class AsyncPubClient {
});
for (int i = 0; i < 1; i++) {
- ThreadUtils.randomSleep(0, 500);
+ ThreadUtils.randomPause(0, 500);
pubClient.broadcast(MessageUtils.asyncMessage(ClientConstants.ASYNC_TOPIC, i),
5000);
}
}
diff --git
a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/demo/BroadCastPubClient.java
b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/demo/BroadCastPubClient.java
index 7c3da75e9..f9f7d34e2 100644
---
a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/demo/BroadCastPubClient.java
+++
b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/demo/BroadCastPubClient.java
@@ -30,7 +30,7 @@ public class BroadCastPubClient {
pubClient.init();
pubClient.heartbeat();
for (int i = 0; i < 10000; i++) {
- ThreadUtils.randomSleep(0, 500);
+ ThreadUtils.randomPause(0, 500);
pubClient.broadcast(MessageUtils.broadcastMessage(ClientConstants.BROADCAST_TOPIC,
i), 5000);
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
