f = f(pje.getMessage());
+ f.setCode(pje.getCode());
+ return f;
+ }
+
+}
diff --git a/powerjob-common/src/main/java/tech/powerjob/common/response/ResultDTO.java b/powerjob-common/src/main/java/tech/powerjob/common/response/ResultDTO.java
index 03e1dc6be..5cd905193 100644
--- a/powerjob-common/src/main/java/tech/powerjob/common/response/ResultDTO.java
+++ b/powerjob-common/src/main/java/tech/powerjob/common/response/ResultDTO.java
@@ -8,6 +8,11 @@
/**
* The result object returned by the request
+ *
+ * 低版本由于 Jackson 序列化配置问题，导致无法在此对象上新增任何字段了，否则会报错 com.fasterxml.jackson.databind.exc.UnrecognizedPropertyException: Unrecognized field "code" (class tech.powerjob.common.response.ObjectResultDTO), not marked as ignorable (3 known properties: "data", "success", "message"])
+ * at [Source: (String)"{"success":true,"code":null,"data":2,"message":null}"; line: 1, column: 28] (through reference chain: tech.powerjob.common.response.ObjectResultDTO["code"])
+ *
+ * 短期内所有的新增字段需求，都通过新对象继承实现
*
* @author tjq
* @since 2020/3/30
@@ -17,9 +22,9 @@
@ToString
public class ResultDTO implements PowerSerializable {
- private boolean success;
- private T data;
- private String message;
+ protected boolean success;
+ protected T data;
+ protected String message;
public static ResultDTO success(T data) {
ResultDTO r = new ResultDTO<>();
diff --git a/powerjob-common/src/main/java/tech/powerjob/common/serialize/JsonUtils.java b/powerjob-common/src/main/java/tech/powerjob/common/serialize/JsonUtils.java
index 0480357a9..7f280bde6 100644
--- a/powerjob-common/src/main/java/tech/powerjob/common/serialize/JsonUtils.java
+++ b/powerjob-common/src/main/java/tech/powerjob/common/serialize/JsonUtils.java
@@ -1,8 +1,9 @@
package tech.powerjob.common.serialize;
+import com.fasterxml.jackson.annotation.JsonInclude;
import com.fasterxml.jackson.core.JsonParser;
-import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.DeserializationFeature;
import com.fasterxml.jackson.databind.MapperFeature;
import com.fasterxml.jackson.databind.json.JsonMapper;
import lombok.extern.slf4j.Slf4j;
@@ -28,8 +29,20 @@ public class JsonUtils {
.configure(MapperFeature.PROPAGATE_TRANSIENT_MARKER, true)
.configure(JsonParser.Feature.ALLOW_SINGLE_QUOTES, true)
.configure(JsonParser.Feature.IGNORE_UNDEFINED, true)
+ .configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false)
.build();
+ static {
+ JSON_MAPPER.setSerializationInclusion(JsonInclude.Include.NON_NULL);
+
+ // 非核心功能可降级，尽可能降低依赖冲突概率
+ try {
+ JSON_MAPPER.registerModule(new com.fasterxml.jackson.datatype.jsr310.JavaTimeModule());
+ } catch (Exception e) {
+ log.warn("[JsonUtils] registerJavaTimeModule failed, PowerJob can't process Java 8 date/time type now!", e);
+ }
+ }
+
private static final TypeReference> MAP_TYPE_REFERENCE = new TypeReference> () {};
private JsonUtils(){
@@ -37,6 +50,9 @@ private JsonUtils(){
}
public static String toJSONString(Object obj) {
+ if (obj == null) {
+ return null;
+ }
if (obj instanceof String) {
return (String) obj;
}
@@ -55,8 +71,9 @@ public static String toJSONStringUnsafe(Object obj) {
try {
return JSON_MAPPER.writeValueAsString(obj);
}catch (Exception e) {
- throw new PowerJobException(e);
+ ExceptionUtils.rethrow(e);
}
+ throw new ImpossibleException();
}
public static byte[] toBytes(Object obj) {
@@ -68,7 +85,7 @@ public static byte[] toBytes(Object obj) {
return null;
}
- public static T parseObject(String json, Class clz) throws JsonProcessingException {
+ public static T parseObject(String json, Class clz) throws Exception {
return JSON_MAPPER.readValue(json, clz);
}
@@ -110,6 +127,9 @@ public static T parseObjectIgnoreException(String json, Class clz) {
}
public static T parseObjectUnsafe(String json, Class clz) {
+ if (json == null) {
+ return null;
+ }
try {
return JSON_MAPPER.readValue(json, clz);
}catch (Exception e) {
@@ -117,4 +137,11 @@ public static T parseObjectUnsafe(String json, Class clz) {
}
throw new PowerJobException("impossible");
}
+
+ public static T toJavaObject(Object o, Class clz) {
+ if (o == null) {
+ return null;
+ }
+ return JSON_MAPPER.convertValue(o, clz);
+ }
}
diff --git a/powerjob-common/src/main/java/tech/powerjob/common/utils/CommonUtils.java b/powerjob-common/src/main/java/tech/powerjob/common/utils/CommonUtils.java
index 498d5b731..d787f3373 100644
--- a/powerjob-common/src/main/java/tech/powerjob/common/utils/CommonUtils.java
+++ b/powerjob-common/src/main/java/tech/powerjob/common/utils/CommonUtils.java
@@ -7,6 +7,8 @@
import org.apache.commons.lang3.time.DateFormatUtils;
import java.util.Collection;
+import java.util.Date;
+import java.util.Map;
import java.util.UUID;
import java.util.function.Supplier;
@@ -128,6 +130,16 @@ public static T requireNonNull(T obj, String msg) {
throw new PowerJobException(msg);
}
}
+ if (obj instanceof Collection) {
+ if (CollectionUtils.isEmpty((Collection) obj)) {
+ throw new PowerJobException(msg);
+ }
+ }
+ if (obj instanceof Map) {
+ if (MapUtils.isEmpty((Map) obj)) {
+ throw new PowerJobException(msg);
+ }
+ }
return obj;
}
@@ -147,6 +159,13 @@ public static String formatTime(Long ts) {
return OmsConstant.NONE;
}
+ public static String formatTime(Date date) {
+ if (date == null) {
+ return OmsConstant.NONE;
+ }
+ return formatTime(date.getTime());
+ }
+
/**
* 格式化字符串，如果是 null 或空则显示 N/A
* @param str 字符串
diff --git a/powerjob-common/src/main/java/tech/powerjob/common/utils/DigestUtils.java b/powerjob-common/src/main/java/tech/powerjob/common/utils/DigestUtils.java
new file mode 100644
index 000000000..4feeb90c0
--- /dev/null
+++ b/powerjob-common/src/main/java/tech/powerjob/common/utils/DigestUtils.java
@@ -0,0 +1,47 @@
+package tech.powerjob.common.utils;
+
+import lombok.SneakyThrows;
+import org.apache.commons.lang3.StringUtils;
+
+import java.math.BigInteger;
+import java.security.MessageDigest;
+
+/**
+ * 加密工具
+ *
+ * @author tjq
+ * @since 2023/3/25
+ */
+public class DigestUtils {
+
+ /**
+ * 32位小写 md5
+ * @param input 输入
+ * @return md5
+ */
+ @SneakyThrows
+ public static String md5(String input) {
+
+ if (StringUtils.isEmpty(input)) {
+ return null;
+ }
+
+ MessageDigest md5 = MessageDigest.getInstance("MD5");
+ md5.update(input.getBytes());
+ byte[] byteArray = md5.digest();
+
+ BigInteger bigInt = new BigInteger(1, byteArray);
+ // 参数16表示16进制
+ StringBuilder result = new StringBuilder(bigInt.toString(16));
+ // 不足32位高位补零
+ while(result.length() < 32) {
+ result.insert(0, "0");
+ }
+ return result.toString();
+ }
+
+ public static String rePassword(String password, String salt) {
+ String f1 = String.format("%s_%s_z", salt, password);
+ return String.format("%s_%s_b", salt, md5(f1));
+ }
+}
diff --git a/powerjob-common/src/main/java/tech/powerjob/common/utils/HttpUtils.java b/powerjob-common/src/main/java/tech/powerjob/common/utils/HttpUtils.java
index fa65ff834..ceb2af526 100644
--- a/powerjob-common/src/main/java/tech/powerjob/common/utils/HttpUtils.java
+++ b/powerjob-common/src/main/java/tech/powerjob/common/utils/HttpUtils.java
@@ -21,6 +21,7 @@ public class HttpUtils {
client = new OkHttpClient.Builder()
.connectTimeout(1, TimeUnit.SECONDS)
.readTimeout(5, TimeUnit.SECONDS)
+ .writeTimeout(10, TimeUnit.SECONDS)
.build();
}
diff --git a/powerjob-common/src/main/java/tech/powerjob/common/utils/MapUtils.java b/powerjob-common/src/main/java/tech/powerjob/common/utils/MapUtils.java
new file mode 100644
index 000000000..dd57bfcd9
--- /dev/null
+++ b/powerjob-common/src/main/java/tech/powerjob/common/utils/MapUtils.java
@@ -0,0 +1,86 @@
+package tech.powerjob.common.utils;
+
+import java.text.NumberFormat;
+import java.text.ParseException;
+import java.util.Map;
+
+/**
+ * MapUtils
+ *
+ * @author tjq
+ * @since 2024/2/24
+ */
+public class MapUtils {
+
+ public static String getString(Map map, K key) {
+ if (map != null) {
+ Object answer = map.get(key);
+ if (answer != null) {
+ return answer.toString();
+ }
+ }
+
+ return null;
+ }
+
+ public static String getString(Map map, K key, String defaultValue) {
+ String answer = getString(map, key);
+ if (answer == null) {
+ answer = defaultValue;
+ }
+
+ return answer;
+ }
+
+ public static Long getLong(Map map, K key, Long defaultValue) {
+ Long answer = getLong(map, key);
+ if (answer == null) {
+ answer = defaultValue;
+ }
+
+ return answer;
+ }
+
+ public static long getLongValue(Map map, K key) {
+ Long longObject = getLong(map, key);
+ return longObject == null ? 0L : longObject;
+ }
+
+ public static Long getLong(Map map, K key) {
+ Number answer = getNumber(map, key);
+ if (answer == null) {
+ return null;
+ } else {
+ return answer instanceof Long ? (Long)answer : answer.longValue();
+ }
+ }
+
+ public static Number getNumber(Map map, K key) {
+ if (map != null) {
+ Object answer = map.get(key);
+ if (answer != null) {
+ if (answer instanceof Number) {
+ return (Number)answer;
+ }
+
+ if (answer instanceof String) {
+ try {
+ String text = (String)answer;
+ return NumberFormat.getInstance().parse(text);
+ } catch (ParseException var4) {
+ }
+ }
+ }
+ }
+
+ return null;
+ }
+
+ public static boolean isEmpty(Map map) {
+ return map == null || map.isEmpty();
+ }
+
+ public static boolean isNotEmpty(Map map) {
+ return !isEmpty(map);
+ }
+}
diff --git a/powerjob-common/src/main/java/tech/powerjob/common/utils/NetUtils.java b/powerjob-common/src/main/java/tech/powerjob/common/utils/NetUtils.java
index 56fa7b3b7..5b7ebbff1 100644
--- a/powerjob-common/src/main/java/tech/powerjob/common/utils/NetUtils.java
+++ b/powerjob-common/src/main/java/tech/powerjob/common/utils/NetUtils.java
@@ -77,6 +77,10 @@ public static boolean checkIpPortAvailable(String ip, int port) {
* @return 本机 IP 地址
*/
public static String getLocalHost() {
+ return getLocalHostWithNetworkInterfaceChecker(null);
+ }
+
+ public static String getLocalHostWithNetworkInterfaceChecker(NetworkInterfaceChecker networkInterfaceChecker) {
if (HOST_ADDRESS != null) {
return HOST_ADDRESS;
}
@@ -87,31 +91,39 @@ public static String getLocalHost() {
return HOST_ADDRESS = addressFromJVM;
}
- InetAddress address = getLocalAddress();
+ InetAddress address = getLocalAddress(networkInterfaceChecker);
if (address != null) {
return HOST_ADDRESS = address.getHostAddress();
}
return LOCALHOST_VALUE;
}
+ /**
+ * 隔离调用 scope，核心场景才能直接调用 getLocalHost，方便查看使用点
+ * @return IP
+ */
+ public static String getLocalHost4Test() {
+ return getLocalHost();
+ }
+
/**
* Find first valid IP from local network card
*
* @return first valid local IP
*/
- public static InetAddress getLocalAddress() {
+ public static InetAddress getLocalAddress(NetworkInterfaceChecker networkInterfaceChecker) {
if (LOCAL_ADDRESS != null) {
return LOCAL_ADDRESS;
}
- InetAddress localAddress = getLocalAddress0();
+ InetAddress localAddress = getLocalAddress0(networkInterfaceChecker);
LOCAL_ADDRESS = localAddress;
return localAddress;
}
- private static InetAddress getLocalAddress0() {
+ private static InetAddress getLocalAddress0(NetworkInterfaceChecker networkInterfaceChecker) {
// @since 2.7.6, choose the {@link NetworkInterface} first
try {
- InetAddress addressOp = getFirstReachableInetAddress( findNetworkInterface());
+ InetAddress addressOp = getFirstReachableInetAddress( findNetworkInterface(networkInterfaceChecker));
if (addressOp != null) {
return addressOp;
}
@@ -161,7 +173,7 @@ private static InetAddress getFirstReachableInetAddress(NetworkInterface network
* @return If no {@link NetworkInterface} is available , return null
* @since 2.7.6
*/
- public static NetworkInterface findNetworkInterface() {
+ public static NetworkInterface findNetworkInterface(NetworkInterfaceChecker networkInterfaceChecker) {
List validNetworkInterfaces = emptyList();
try {
@@ -176,7 +188,11 @@ public static NetworkInterface findNetworkInterface() {
// Try to find the preferred one
for (NetworkInterface networkInterface : validNetworkInterfaces) {
if (isPreferredNetworkInterface(networkInterface)) {
- log.info("[Net] use preferred network interface: {}", networkInterface.getDisplayName());
+ log.info("[Net] use preferred network interface: {}", networkInterface);
+ return networkInterface;
+ }
+ if (isPassedCheckNetworkInterface(networkInterface, networkInterfaceChecker)) {
+ log.info("[Net] use PassedCheckNetworkInterface: {}", networkInterface);
return networkInterface;
}
}
@@ -191,6 +207,26 @@ public static NetworkInterface findNetworkInterface() {
return first(validNetworkInterfaces);
}
+ /**
+ * 通过用户方法判断是否为目标网卡
+ * @param networkInterface networkInterface
+ * @param networkInterfaceChecker 判断方法
+ * @return true or false
+ */
+ static boolean isPassedCheckNetworkInterface(NetworkInterface networkInterface, NetworkInterfaceChecker networkInterfaceChecker) {
+ if (networkInterfaceChecker == null) {
+ return false;
+ }
+ try {
+ boolean ok = networkInterfaceChecker.ok(networkInterface, getFirstReachableInetAddress(networkInterface));
+ log.info("[Net] try to choose NetworkInterface by NetworkInterfaceChecker, current NetworkInterface[{}], ok: {}", networkInterface, ok);
+ return ok;
+ } catch (Exception e) {
+ log.warn("[Net] isPassedCheckerNetworkInterface failed, current networkInterface: {}", networkInterface, e);
+ }
+ return false;
+ }
+
private static Optional toValidAddress(InetAddress address) {
if (address instanceof Inet6Address) {
Inet6Address v6Address = (Inet6Address) address;
@@ -344,4 +380,9 @@ static boolean ignoreInterfaceByConfig(String interfaceName) {
}
return false;
}
+
+ @FunctionalInterface
+ public interface NetworkInterfaceChecker {
+ boolean ok(NetworkInterface networkInterface, InetAddress inetAddress);
+ }
}
diff --git a/powerjob-common/src/main/java/tech/powerjob/common/utils/SysUtils.java b/powerjob-common/src/main/java/tech/powerjob/common/utils/SysUtils.java
new file mode 100644
index 000000000..019fe8edc
--- /dev/null
+++ b/powerjob-common/src/main/java/tech/powerjob/common/utils/SysUtils.java
@@ -0,0 +1,50 @@
+package tech.powerjob.common.utils;
+
+import org.apache.commons.lang3.StringUtils;
+import tech.powerjob.common.PowerJobDKey;
+import tech.powerjob.common.enums.SpEnv;
+
+/**
+ * 系统工具
+ *
+ * @author tjq
+ * @since 2025/8/17
+ */
+public class SysUtils {
+
+ /**
+ * 可用处理器核数
+ * @return 可用处理器核数
+ */
+ public static int availableProcessors() {
+ String property = System.getProperty(PowerJobDKey.SYS_AVAILABLE_PROCESSORS);
+ if (StringUtils.isEmpty(property)) {
+ return Runtime.getRuntime().availableProcessors();
+ }
+ return Integer.parseInt(property);
+ }
+
+ /**
+ * 判断是否为测试环境
+ * @return 测试环境
+ */
+ public static boolean isTestEnv() {
+ String property = System.getProperty(PowerJobDKey.SP_ENV);
+ if (StringUtils.isEmpty(property)) {
+ return false;
+ }
+ return SpEnv.TEST.getCode().equalsIgnoreCase(property);
+ }
+
+ /**
+ * 判断是否为试用环境
+ * @return 试用环境
+ */
+ public static boolean isTrialEnv() {
+ String property = System.getProperty(PowerJobDKey.SP_ENV);
+ if (StringUtils.isEmpty(property)) {
+ return false;
+ }
+ return SpEnv.TRIAL.getCode().equalsIgnoreCase(property);
+ }
+}
diff --git a/powerjob-common/src/main/java/tech/powerjob/common/utils/net/PingPongServer.java b/powerjob-common/src/main/java/tech/powerjob/common/utils/net/PingPongServer.java
new file mode 100644
index 000000000..53e4f0c3e
--- /dev/null
+++ b/powerjob-common/src/main/java/tech/powerjob/common/utils/net/PingPongServer.java
@@ -0,0 +1,14 @@
+package tech.powerjob.common.utils.net;
+
+import java.io.Closeable;
+
+/**
+ * socket 服务器，用于进行连通性测试
+ *
+ * @author tjq
+ * @since 2024/2/8
+ */
+public interface PingPongServer extends Closeable {
+
+ void initialize(int port) throws Exception;
+}
diff --git a/powerjob-common/src/main/java/tech/powerjob/common/utils/net/PingPongSocketServer.java b/powerjob-common/src/main/java/tech/powerjob/common/utils/net/PingPongSocketServer.java
new file mode 100644
index 000000000..1ec93d74b
--- /dev/null
+++ b/powerjob-common/src/main/java/tech/powerjob/common/utils/net/PingPongSocketServer.java
@@ -0,0 +1,63 @@
+package tech.powerjob.common.utils.net;
+
+import lombok.extern.slf4j.Slf4j;
+import tech.powerjob.common.utils.CommonUtils;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.net.ServerSocket;
+import java.net.Socket;
+import java.nio.charset.StandardCharsets;
+
+/**
+ * 简易服务器
+ *
+ * @author tjq
+ * @since 2024/2/8
+ */
+@Slf4j
+public class PingPongSocketServer implements PingPongServer {
+
+ private Thread thread;
+
+ private ServerSocket serverSocket;
+
+ private volatile boolean terminated = false;
+
+ @Override
+ public void initialize(int port) throws Exception{
+ serverSocket = new ServerSocket(port);
+
+ thread = new Thread(() -> {
+ while (true) {
+ if (terminated) {
+ return;
+ }
+ // 接收连接，如果没有连接，accept() 方法会阻塞
+ try (Socket socket = serverSocket.accept();OutputStream outputStream = socket.getOutputStream();) {
+
+ socket.setSoTimeout(2000);
+ socket.setKeepAlive(false);
+
+ outputStream.write(PingPongUtils.PONG.getBytes(StandardCharsets.UTF_8));
+ // BufferedReader.readLine() 会等待直到遇到换行符（\n）或回车符（\r\n），才会返回一行内容。如果服务器发送的数据没有这些换行符，readLine() 会一直阻塞，直到超时
+ outputStream.write(System.lineSeparator().getBytes(StandardCharsets.UTF_8));
+ outputStream.flush();
+ } catch (Exception e) {
+ if (!terminated) {
+ log.warn("[PingPongSocketServer] process accepted socket failed!", e);
+ }
+ }
+ }
+ }, "PingPongSocketServer-Thread");
+
+ thread.start();
+ }
+
+ @Override
+ public void close() throws IOException {
+ terminated = true;
+ CommonUtils.executeIgnoreException(() -> serverSocket.close());
+ thread.interrupt();
+ }
+}
diff --git a/powerjob-common/src/main/java/tech/powerjob/common/utils/net/PingPongUtils.java b/powerjob-common/src/main/java/tech/powerjob/common/utils/net/PingPongUtils.java
new file mode 100644
index 000000000..821f7352c
--- /dev/null
+++ b/powerjob-common/src/main/java/tech/powerjob/common/utils/net/PingPongUtils.java
@@ -0,0 +1,56 @@
+package tech.powerjob.common.utils.net;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.exception.ExceptionUtils;
+
+import java.io.*;
+import java.net.Socket;
+import java.net.UnknownHostException;
+import java.nio.charset.StandardCharsets;
+
+/**
+ * socket 连通性助手
+ *
+ * @author tjq
+ * @since 2024/2/8
+ */
+@Slf4j
+public class PingPongUtils {
+
+ static final String PING = "ping";
+ static final String PONG = "pong";
+
+ /**
+ * 验证目标 IP 和 端口的连通性
+ * @param targetIp 目标 IP
+ * @param targetPort 目标端口
+ * @return true or false
+ */
+ public static boolean checkConnectivity(String targetIp, int targetPort) {
+
+ try (Socket s = new Socket(targetIp, targetPort);InputStream is = s.getInputStream();OutputStream os = s.getOutputStream();BufferedReader br = new BufferedReader(new InputStreamReader(is))) {
+
+ s.setSoTimeout(2000);
+ s.setKeepAlive(false);
+
+ // 发送 PING 请求
+ os.write(PING.getBytes(StandardCharsets.UTF_8));
+ os.flush();
+
+ //读取服务器返回的消息
+ String content = br.readLine();
+
+ if (PONG.equalsIgnoreCase(content)) {
+ return true;
+ }
+ } catch (UnknownHostException e) {
+ log.warn("[SocketConnectivityUtils] unknown host: {}:{}", targetIp, targetPort);
+ } catch (IOException e) {
+ log.warn("[SocketConnectivityUtils] IOException: {}:{}, msg: {}", targetIp, targetPort, ExceptionUtils.getMessage(e));
+ } catch (Exception e) {
+ log.error("[SocketConnectivityUtils] unknown exception for check ip: {}:{}", targetIp, targetPort, e);
+ }
+
+ return false;
+ }
+}
diff --git a/powerjob-common/src/test/java/NetUtilsTest.java b/powerjob-common/src/test/java/NetUtilsTest.java
index 7d6c562a4..98bfeda01 100644
--- a/powerjob-common/src/test/java/NetUtilsTest.java
+++ b/powerjob-common/src/test/java/NetUtilsTest.java
@@ -12,19 +12,19 @@ public class NetUtilsTest {
@Test
public void testOrigin() {
- System.out.println(NetUtils.getLocalHost());
+ System.out.println(NetUtils.getLocalHost4Test());
}
@Test
public void testPreferredNetworkInterface() {
System.setProperty(PowerJobDKey.PREFERRED_NETWORK_INTERFACE, "en5");
- System.out.println(NetUtils.getLocalHost());
+ System.out.println(NetUtils.getLocalHost4Test());
}
@Test
public void testIgnoredNetworkInterface() {
System.setProperty(PowerJobDKey.IGNORED_NETWORK_INTERFACE_REGEX, "utun.|llw.");
- System.out.println(NetUtils.getLocalHost());
+ System.out.println(NetUtils.getLocalHost4Test());
}
}
diff --git a/powerjob-common/src/test/java/tech/powerjob/common/serialize/JsonUtilsTest.java b/powerjob-common/src/test/java/tech/powerjob/common/serialize/JsonUtilsTest.java
new file mode 100644
index 000000000..3e8b029e5
--- /dev/null
+++ b/powerjob-common/src/test/java/tech/powerjob/common/serialize/JsonUtilsTest.java
@@ -0,0 +1,81 @@
+package tech.powerjob.common.serialize;
+
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+import lombok.SneakyThrows;
+import lombok.experimental.Accessors;
+import lombok.extern.slf4j.Slf4j;
+import org.junit.jupiter.api.Test;
+
+import java.io.Serializable;
+import java.time.LocalDateTime;
+import java.util.Map;
+
+/**
+ * test json utils
+ *
+ * @author tjq
+ * @since 2024/3/16
+ */
+@Slf4j
+class JsonUtilsTest {
+
+ @Test
+ @SneakyThrows
+ void simpleTest() {
+ Person person = new Person().setName("mubao").setAge(18);
+ String jsonString = JsonUtils.toJSONString(person);
+ log.info("[JsonUtilsTest] person: {}, jsonString: {}", person, jsonString);
+ assert jsonString != null;
+ Person person2 = JsonUtils.parseObject(jsonString, Person.class);
+ assert person2.equals(person);
+ }
+
+ @Test
+ @SneakyThrows
+ void testAdvanceApi() {
+ PersonPlus personPlus = new PersonPlus();
+ personPlus.setName("gongbao").setAge(3);
+ personPlus.setBirthDay(LocalDateTime.now());
+
+ String jsonString = JsonUtils.toJSONString(personPlus);
+ PersonPlus personPlus2 = JsonUtils.parseObject(jsonString, PersonPlus.class);
+ assert personPlus2.equals(personPlus);
+ }
+
+ @Test
+ @SneakyThrows
+ void testMoreOrLessFields() {
+ PersonPlus personPlus = new PersonPlus().setBirthDay(LocalDateTime.now());
+ personPlus.setName("gongbao").setAge(3);
+
+ String originJsonStr = JsonUtils.toJSONString(personPlus);
+
+ Map personPlusMapMore = JsonUtils.parseMap(originJsonStr);
+ personPlusMapMore.put("extraKey", System.currentTimeMillis());
+
+ PersonPlus personPlusByMoreFieldsJsonStr = JsonUtils.parseObject(JsonUtils.toJSONString(personPlusMapMore), PersonPlus.class);
+ assert personPlusByMoreFieldsJsonStr.equals(personPlus);
+
+ Map personPlusMapLess = JsonUtils.parseMap(originJsonStr);
+ personPlusMapLess.remove("birthDay");
+
+ PersonPlus personPlusByLessFieldsJsonStr = JsonUtils.parseObject(JsonUtils.toJSONString(personPlusMapLess), PersonPlus.class);
+ assert personPlusByLessFieldsJsonStr.getName().equals(personPlus.getName());
+ assert personPlusByLessFieldsJsonStr.getAge().equals(personPlus.getAge());
+ }
+
+ @Data
+ @Accessors(chain = true)
+ static class Person implements Serializable {
+ private String name;
+ private Integer age;
+ }
+
+ @Data
+ @Accessors(chain = true)
+ @EqualsAndHashCode(callSuper = true)
+ static class PersonPlus extends Person {
+ private LocalDateTime birthDay;
+ }
+}
\ No newline at end of file
diff --git a/powerjob-common/src/test/java/tech/powerjob/common/utils/CommonUtilsTest.java b/powerjob-common/src/test/java/tech/powerjob/common/utils/CommonUtilsTest.java
new file mode 100644
index 000000000..26453dd4d
--- /dev/null
+++ b/powerjob-common/src/test/java/tech/powerjob/common/utils/CommonUtilsTest.java
@@ -0,0 +1,39 @@
+package tech.powerjob.common.utils;
+
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+import org.junit.jupiter.api.Test;
+import tech.powerjob.common.exception.PowerJobException;
+
+import java.util.Collections;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+/**
+ * CommonUtilsTest
+ *
+ * @author tjq
+ * @since 2024/8/11
+ */
+class CommonUtilsTest {
+
+ @Test
+ void testRequireNonNull() {
+
+ assertThrowsExactly(PowerJobException.class, () -> CommonUtils.requireNonNull(null, "NULL_OBJ"));
+ assertThrowsExactly(PowerJobException.class, () -> CommonUtils.requireNonNull("", "EMPTY_STR"));
+ assertThrowsExactly(PowerJobException.class, () -> CommonUtils.requireNonNull(Lists.newArrayList(), "EMPTY_COLLECTION"));
+ assertThrowsExactly(PowerJobException.class, () -> CommonUtils.requireNonNull(Collections.emptyMap(), "EMPTY_MAP"));
+
+ Map map = Maps.newHashMap();
+ map.put("1", 1);
+
+ CommonUtils.requireNonNull(1, "NORMAL");
+ CommonUtils.requireNonNull("1", "NORMAL");
+ CommonUtils.requireNonNull(Lists.newArrayList("1"), "NORMAL");
+ CommonUtils.requireNonNull(map, "NORMAL");
+
+ }
+
+}
\ No newline at end of file
diff --git a/powerjob-common/src/test/java/tech/powerjob/common/utils/net/PingPongSocketServerTest.java b/powerjob-common/src/test/java/tech/powerjob/common/utils/net/PingPongSocketServerTest.java
new file mode 100644
index 000000000..0d559111e
--- /dev/null
+++ b/powerjob-common/src/test/java/tech/powerjob/common/utils/net/PingPongSocketServerTest.java
@@ -0,0 +1,31 @@
+package tech.powerjob.common.utils.net;
+
+import org.junit.jupiter.api.Test;
+import tech.powerjob.common.utils.NetUtils;
+
+/**
+ * desc
+ *
+ * @author tjq
+ * @since 2024/2/8
+ */
+class PingPongSocketServerTest {
+
+ @Test
+ void test() throws Exception {
+
+ int port = 8877;
+
+ PingPongSocketServer pingPongSocketServer = new PingPongSocketServer();
+ pingPongSocketServer.initialize(port);
+
+ System.out.println("[PingPongSocketServerTest] finished initialize");
+
+ assert PingPongUtils.checkConnectivity(NetUtils.getLocalHost(), port);
+
+ assert !PingPongUtils.checkConnectivity(NetUtils.getLocalHost(), port + 1);
+
+ pingPongSocketServer.close();
+ System.out.println("[PingPongSocketServerTest] finished close");
+ }
+}
\ No newline at end of file
diff --git a/powerjob-official-processors/pom.xml b/powerjob-official-processors/pom.xml
index 7a4eb979f..56229b1f5 100644
--- a/powerjob-official-processors/pom.xml
+++ b/powerjob-official-processors/pom.xml
@@ -5,12 +5,13 @@
powerjob
tech.powerjob
- 4.3.6
+ 5.1.2
4.0.0
powerjob-official-processors
- 4.3.6
+ powerjob-official-processors
+ 5.1.2
jar
@@ -19,12 +20,11 @@
5.9.1
- 1.2.9
- 4.3.6
- 5.2.9.RELEASE
- 2.2.220
+ 1.2.13
+ 5.1.2
+ 2.2.224
8.0.28
- 5.3.23
+ 5.3.31
1.2.83
@@ -87,7 +87,7 @@
org.springframework
spring-jdbc
- ${spring.jdbc.version}
+ ${spring.version}
test
diff --git a/powerjob-official-processors/src/main/java/tech/powerjob/official/processors/impl/VerificationProcessor.java b/powerjob-official-processors/src/main/java/tech/powerjob/official/processors/impl/VerificationProcessor.java
index 4588c22e5..e8768be6b 100644
--- a/powerjob-official-processors/src/main/java/tech/powerjob/official/processors/impl/VerificationProcessor.java
+++ b/powerjob-official-processors/src/main/java/tech/powerjob/official/processors/impl/VerificationProcessor.java
@@ -4,6 +4,7 @@
import com.google.common.collect.Lists;
import lombok.*;
import org.apache.commons.lang3.RandomStringUtils;
+import org.apache.commons.lang3.StringUtils;
import tech.powerjob.common.exception.PowerJobException;
import tech.powerjob.common.utils.NetUtils;
import tech.powerjob.official.processors.CommonBasicProcessor;
@@ -35,10 +36,12 @@ protected ProcessResult process0(TaskContext taskContext) throws Exception {
final OmsLogger omsLogger = taskContext.getOmsLogger();
final String paramsStr = CommonUtils.parseParams(taskContext);
- final VerificationParam verificationParam = JSONObject.parseObject(paramsStr, VerificationParam.class);
+ final VerificationParam verificationParam = StringUtils.isEmpty(paramsStr) ? new VerificationParam() : JSONObject.parseObject(paramsStr, VerificationParam.class);
final Mode mode = Mode.of(verificationParam.getMode());
+ omsLogger.info("META: {}", taskContext.getInstanceMeta());
+
switch (mode) {
case ERROR:
return new ProcessResult(false, "EXECUTE_FAILED_FOR_TEST");
@@ -123,14 +126,14 @@ public ProcessResult reduce(TaskContext context, List taskResults) {
@Override
public ProcessResult preProcess(TaskContext context) throws Exception {
- context.getOmsLogger().info("start to preProcess, current worker IP is {}.", NetUtils.getLocalHost());
+ context.getOmsLogger().info("start to preProcess, current worker IP is {}.", NetUtils.getLocalHost4Test());
return new ProcessResult(true, "preProcess successfully!");
}
@Override
public ProcessResult postProcess(TaskContext context, List taskResults) throws Exception {
OmsLogger omsLogger = context.getOmsLogger();
- omsLogger.info("start to postProcess, current worker IP is {}.", NetUtils.getLocalHost());
+ omsLogger.info("start to postProcess, current worker IP is {}.", NetUtils.getLocalHost4Test());
omsLogger.info("====== All Node's Process Result ======");
taskResults.forEach(r -> omsLogger.info("taskId:{},success:{},result:{}", r.getTaskId(), r.isSuccess(), r.getResult()));
return new ProcessResult(true, "postProcess successfully!");
diff --git a/powerjob-official-processors/src/main/java/tech/powerjob/official/processors/impl/script/AbstractScriptProcessor.java b/powerjob-official-processors/src/main/java/tech/powerjob/official/processors/impl/script/AbstractScriptProcessor.java
index 40d4dde10..a6d3a09e2 100644
--- a/powerjob-official-processors/src/main/java/tech/powerjob/official/processors/impl/script/AbstractScriptProcessor.java
+++ b/powerjob-official-processors/src/main/java/tech/powerjob/official/processors/impl/script/AbstractScriptProcessor.java
@@ -1,5 +1,6 @@
package tech.powerjob.official.processors.impl.script;
+import tech.powerjob.common.utils.SysUtils;
import tech.powerjob.worker.common.utils.PowerFileUtils;
import tech.powerjob.worker.core.processor.ProcessResult;
import tech.powerjob.worker.core.processor.TaskContext;
@@ -31,7 +32,7 @@
@Slf4j
public abstract class AbstractScriptProcessor extends CommonBasicProcessor {
- private static final ForkJoinPool POOL = new ForkJoinPool(4 * Runtime.getRuntime().availableProcessors());
+ private static final ForkJoinPool POOL = new ForkJoinPool(4 * SysUtils.availableProcessors());
private static final Set DOWNLOAD_PROTOCOL = Sets.newHashSet("http", "https", "ftp");
protected static final String SH_SHELL = "/bin/sh";
protected static final String CMD_SHELL = "cmd.exe";
diff --git a/powerjob-remote/pom.xml b/powerjob-remote/pom.xml
index 511139c0a..dad931517 100644
--- a/powerjob-remote/pom.xml
+++ b/powerjob-remote/pom.xml
@@ -5,18 +5,19 @@
powerjob
tech.powerjob
- 4.3.6
+ 5.1.2
4.0.0
pom
powerjob-remote-framework
- powerjob-remote-benchmark
powerjob-remote-impl-http
powerjob-remote-impl-akka
+ powerjob-remote-impl-mu
powerjob-remote
+ powerjob-remote
8
@@ -24,7 +25,7 @@
UTF-8
5.9.0
- 1.2.9
+ 1.2.13
diff --git a/powerjob-remote/powerjob-remote-benchmark/pom.xml b/powerjob-remote/powerjob-remote-benchmark/pom.xml
deleted file mode 100644
index c9b80b946..000000000
--- a/powerjob-remote/powerjob-remote-benchmark/pom.xml
+++ /dev/null
@@ -1,113 +0,0 @@
-
-
-
- powerjob-remote
- tech.powerjob
- 4.3.6
-
- 4.0.0
-
- powerjob-remote-benchmark
-
-
- 8
- 8
- UTF-8
-
- 3.10.1
- 3.2.2
-
- 1.2.9
- 2.7.14
- 4.3.6
- 4.3.6
-
- 3.9.0
- 4.2.9
-
-
-
-
-
-
- io.gatling.highcharts
- gatling-charts-highcharts
- ${gatling.version}
- test
-
-
-
- ch.qos.logback
- logback-classic
- ${logback.version}
-
-
-
- org.springframework.boot
- spring-boot-starter-web
- ${springboot.version}
-
-
-
- tech.powerjob
- powerjob-remote-impl-http
- ${powerjob-remote-impl-http.version}
-
-
-
- tech.powerjob
- powerjob-remote-impl-akka
- ${powerjob-remote-impl-akka.version}
-
-
-
-
-
-
-
-
-
- org.springframework.boot
- spring-boot-maven-plugin
- ${springboot.version}
-
- tech.powerjob.remote.benchmark.BenchmarkApplication
-
-
-
-
- repackage
-
-
-
-
-
-
- maven-compiler-plugin
- ${maven-compiler-plugin.version}
-
-
- maven-jar-plugin
- ${maven-jar-plugin.version}
-
-
- io.gatling
- gatling-maven-plugin
- ${gatling-maven-plugin.version}
-
-
-
-
-
-
- org.apache.maven.plugins
- maven-deploy-plugin
-
- true
-
-
-
-
-
\ No newline at end of file
diff --git a/powerjob-remote/powerjob-remote-benchmark/src/main/java/tech/powerjob/remote/benchmark/BenchmarkApplication.java b/powerjob-remote/powerjob-remote-benchmark/src/main/java/tech/powerjob/remote/benchmark/BenchmarkApplication.java
deleted file mode 100644
index 64dcb25e1..000000000
--- a/powerjob-remote/powerjob-remote-benchmark/src/main/java/tech/powerjob/remote/benchmark/BenchmarkApplication.java
+++ /dev/null
@@ -1,19 +0,0 @@
-package tech.powerjob.remote.benchmark;
-
-import org.springframework.boot.SpringApplication;
-import org.springframework.boot.autoconfigure.SpringBootApplication;
-
-/**
- * 测试工程
- * 用于 remote 协议压测
- *
- * @author tjq
- * @since 2023/1/7
- */
-@SpringBootApplication
-public class BenchmarkApplication {
-
- public static void main(String[] args) {
- SpringApplication.run(BenchmarkApplication.class, args);
- }
-}
diff --git a/powerjob-remote/powerjob-remote-benchmark/src/main/java/tech/powerjob/remote/benchmark/EngineService.java b/powerjob-remote/powerjob-remote-benchmark/src/main/java/tech/powerjob/remote/benchmark/EngineService.java
deleted file mode 100644
index 41079878d..000000000
--- a/powerjob-remote/powerjob-remote-benchmark/src/main/java/tech/powerjob/remote/benchmark/EngineService.java
+++ /dev/null
@@ -1,68 +0,0 @@
-package tech.powerjob.remote.benchmark;
-
-import com.google.common.collect.Lists;
-import lombok.Getter;
-import org.springframework.stereotype.Service;
-import tech.powerjob.common.enums.Protocol;
-import tech.powerjob.remote.framework.BenchmarkActor;
-import tech.powerjob.remote.framework.base.Address;
-import tech.powerjob.remote.framework.base.ServerType;
-import tech.powerjob.remote.framework.engine.EngineConfig;
-import tech.powerjob.remote.framework.engine.impl.PowerJobRemoteEngine;
-import tech.powerjob.remote.framework.transporter.Transporter;
-
-import javax.annotation.PostConstruct;
-
-/**
- * EngineService
- *
- * @author tjq
- * @since 2023/1/7
- */
-@Service
-public class EngineService {
-
- public static final String HOST = "127.0.0.1";
-
- public static final int SERVER_AKKA_PORT = 10001;
- public static final int SERVER_HTTP_PORT = 10002;
-
- public static final int CLIENT_AKKA_PORT = 20001;
- public static final int CLIENT_HTTP_PORT = 20002;
-
- @Getter
- private Transporter akkaTransporter;
- @Getter
- private Transporter httpTransporter;
-
- @PostConstruct
- public void init() {
- // http server
- new PowerJobRemoteEngine().start(new EngineConfig()
- .setServerType(ServerType.SERVER)
- .setActorList(Lists.newArrayList(new BenchmarkActor()))
- .setType(Protocol.HTTP.name())
- .setBindAddress(new Address().setHost(HOST).setPort(SERVER_HTTP_PORT)));
-
- // akka server
- new PowerJobRemoteEngine().start(new EngineConfig()
- .setServerType(ServerType.SERVER)
- .setActorList(Lists.newArrayList(new BenchmarkActor()))
- .setType(Protocol.AKKA.name())
- .setBindAddress(new Address().setHost(HOST).setPort(SERVER_AKKA_PORT)));
-
- // http client
- httpTransporter = new PowerJobRemoteEngine().start(new EngineConfig()
- .setServerType(ServerType.WORKER)
- .setActorList(Lists.newArrayList(new BenchmarkActor()))
- .setType(Protocol.HTTP.name())
- .setBindAddress(new Address().setHost(HOST).setPort(CLIENT_HTTP_PORT))).getTransporter();
-
- // akka client
- akkaTransporter = new PowerJobRemoteEngine().start(new EngineConfig()
- .setServerType(ServerType.WORKER)
- .setActorList(Lists.newArrayList(new BenchmarkActor()))
- .setType(Protocol.AKKA.name())
- .setBindAddress(new Address().setHost(HOST).setPort(CLIENT_AKKA_PORT))).getTransporter();
- }
-}
diff --git a/powerjob-remote/powerjob-remote-benchmark/src/main/java/tech/powerjob/remote/benchmark/PressureTestController.java b/powerjob-remote/powerjob-remote-benchmark/src/main/java/tech/powerjob/remote/benchmark/PressureTestController.java
deleted file mode 100644
index c2f9fcc45..000000000
--- a/powerjob-remote/powerjob-remote-benchmark/src/main/java/tech/powerjob/remote/benchmark/PressureTestController.java
+++ /dev/null
@@ -1,82 +0,0 @@
-package tech.powerjob.remote.benchmark;
-
-import lombok.extern.slf4j.Slf4j;
-import org.apache.commons.lang3.BooleanUtils;
-import org.apache.commons.lang3.exception.ExceptionUtils;
-import org.springframework.web.bind.annotation.GetMapping;
-import org.springframework.web.bind.annotation.RequestMapping;
-import org.springframework.web.bind.annotation.RestController;
-import tech.powerjob.common.enums.Protocol;
-import tech.powerjob.remote.framework.BenchmarkActor;
-import tech.powerjob.remote.framework.base.Address;
-import tech.powerjob.remote.framework.base.HandlerLocation;
-import tech.powerjob.remote.framework.base.URL;
-
-import javax.annotation.Resource;
-
-import java.util.concurrent.CompletionStage;
-
-import static tech.powerjob.remote.benchmark.EngineService.*;
-
-/**
- * 压测测试入口
- *
- * @author tjq
- * @since 2023/1/7
- */
-@Slf4j
-@RestController
-@RequestMapping("/pressure")
-public class PressureTestController {
-
- private static final HandlerLocation HL = new HandlerLocation().setRootPath("benchmark").setMethodPath("standard");
-
- @Resource
- private EngineService engineService;
-
- @GetMapping("/tell")
- public void httpTell(String protocol, Integer blockMs, Integer responseSize, String content) {
- Address address = new Address().setHost(HOST);
- URL url = new URL().setLocation(HL).setAddress(address);
- final BenchmarkActor.BenchmarkRequest request = new BenchmarkActor.BenchmarkRequest().setContent(content).setBlockingMills(blockMs).setResponseSize(responseSize);
- try {
- if (Protocol.HTTP.name().equalsIgnoreCase(protocol)) {
- address.setPort(SERVER_HTTP_PORT);
- engineService.getHttpTransporter().tell(url, request);
- } else {
- address.setPort(SERVER_AKKA_PORT);
- engineService.getAkkaTransporter().tell(url, request);
- }
- } catch (Exception e) {
- log.error("[HttpTell] process failed!", e);
- ExceptionUtils.rethrow(e);
- }
- }
-
-
- @GetMapping("/ask")
- public void httpAsk(String protocol, Integer blockMs, Integer responseSize, String content, Boolean debug) {
- Address address = new Address().setHost(HOST);
- URL url = new URL().setLocation(HL).setAddress(address);
- final BenchmarkActor.BenchmarkRequest request = new BenchmarkActor.BenchmarkRequest().setContent(content).setBlockingMills(blockMs).setResponseSize(responseSize);
- try {
- CompletionStage responseOpt = null;
-
- if (Protocol.HTTP.name().equalsIgnoreCase(protocol)) {
- address.setPort(SERVER_HTTP_PORT);
- responseOpt = engineService.getHttpTransporter().ask(url, request, BenchmarkActor.BenchmarkResponse.class);
- } else {
- address.setPort(SERVER_AKKA_PORT);
- responseOpt = engineService.getAkkaTransporter().ask(url, request, BenchmarkActor.BenchmarkResponse.class);
- }
- final BenchmarkActor.BenchmarkResponse response = responseOpt.toCompletableFuture().get();
- if (BooleanUtils.isTrue(debug)) {
- log.info("[httpAsk] response: {}", response);
- }
- } catch (Exception e) {
- log.error("[httpAsk] process failed", e);
- ExceptionUtils.rethrow(e);
- }
- }
-
-}
diff --git a/powerjob-remote/powerjob-remote-benchmark/src/test/java/Engine.java b/powerjob-remote/powerjob-remote-benchmark/src/test/java/Engine.java
deleted file mode 100644
index fa70e9eb6..000000000
--- a/powerjob-remote/powerjob-remote-benchmark/src/test/java/Engine.java
+++ /dev/null
@@ -1,20 +0,0 @@
-import io.gatling.app.Gatling;
-import io.gatling.core.config.GatlingPropertiesBuilder;
-
-/**
- * 压测启动入口
- *
- * @author tjq
- * @since 2023/1/8
- */
-public class Engine {
-
- public static void main(String[] args) {
- GatlingPropertiesBuilder props = new GatlingPropertiesBuilder()
- .resourcesDirectory(IDEPathHelper.mavenResourcesDirectory.toString())
- .resultsDirectory(IDEPathHelper.resultsDirectory.toString())
- .binariesDirectory(IDEPathHelper.mavenBinariesDirectory.toString());
-
- Gatling.fromMap(props.build());
- }
-}
diff --git a/powerjob-remote/powerjob-remote-benchmark/src/test/java/IDEPathHelper.java b/powerjob-remote/powerjob-remote-benchmark/src/test/java/IDEPathHelper.java
deleted file mode 100644
index 5315cad1b..000000000
--- a/powerjob-remote/powerjob-remote-benchmark/src/test/java/IDEPathHelper.java
+++ /dev/null
@@ -1,33 +0,0 @@
-import java.net.URISyntaxException;
-import java.nio.file.Path;
-import java.nio.file.Paths;
-
-import static java.util.Objects.requireNonNull;
-
-/**
- * @author gatling-maven-plugin-demo-java
- */
-public class IDEPathHelper {
-
- static final Path mavenSourcesDirectory;
- static final Path mavenResourcesDirectory;
- static final Path mavenBinariesDirectory;
- static final Path resultsDirectory;
- static final Path recorderConfigFile;
-
- static {
- try {
- Path projectRootDir = Paths.get(requireNonNull(IDEPathHelper.class.getResource("gatling.conf"), "Couldn't locate gatling.conf").toURI()).getParent().getParent().getParent();
- Path mavenTargetDirectory = projectRootDir.resolve("target");
- Path mavenSrcTestDirectory = projectRootDir.resolve("src").resolve("test");
-
- mavenSourcesDirectory = mavenSrcTestDirectory.resolve("java");
- mavenResourcesDirectory = mavenSrcTestDirectory.resolve("resources");
- mavenBinariesDirectory = mavenTargetDirectory.resolve("test-classes");
- resultsDirectory = mavenTargetDirectory.resolve("gatling");
- recorderConfigFile = mavenResourcesDirectory.resolve("recorder.conf");
- } catch (URISyntaxException e) {
- throw new ExceptionInInitializerError(e);
- }
- }
-}
diff --git a/powerjob-remote/powerjob-remote-benchmark/src/test/java/tech/powerjob/remote/benchmark/Constant.java b/powerjob-remote/powerjob-remote-benchmark/src/test/java/tech/powerjob/remote/benchmark/Constant.java
deleted file mode 100644
index 71820edb3..000000000
--- a/powerjob-remote/powerjob-remote-benchmark/src/test/java/tech/powerjob/remote/benchmark/Constant.java
+++ /dev/null
@@ -1,14 +0,0 @@
-package tech.powerjob.remote.benchmark;
-
-/**
- * Constant
- * 压测时需要修改的常量
- *
- * @author tjq
- * @since 2023/1/8
- */
-public class Constant {
-
- public static final String SERVER_HOST = "127.0.0.1";
-
-}
diff --git a/powerjob-remote/powerjob-remote-benchmark/src/test/java/tech/powerjob/remote/benchmark/HttpSimulation.java b/powerjob-remote/powerjob-remote-benchmark/src/test/java/tech/powerjob/remote/benchmark/HttpSimulation.java
deleted file mode 100644
index 00b1a8b48..000000000
--- a/powerjob-remote/powerjob-remote-benchmark/src/test/java/tech/powerjob/remote/benchmark/HttpSimulation.java
+++ /dev/null
@@ -1,62 +0,0 @@
-package tech.powerjob.remote.benchmark;
-
-import static io.gatling.javaapi.core.CoreDsl.*;
-import static io.gatling.javaapi.http.HttpDsl.*;
-
-import io.gatling.javaapi.core.*;
-import io.gatling.javaapi.http.*;
-/**
- * 以 HTTP 为入口压测
- *
- * @author tjq
- * @since 2023/1/8
- */
-public class HttpSimulation extends Simulation {
-
- String baseUrl = String.format("http://%s:8080", Constant.SERVER_HOST);
- HttpProtocolBuilder httpProtocol = http // 4
- .baseUrl(baseUrl) // 5
- .acceptHeader("application/json") // 6
- .doNotTrackHeader("1")
- .acceptLanguageHeader("en-US,en;q=0.5")
- .acceptEncodingHeader("gzip, deflate")
- .userAgentHeader("Mozilla/5.0 (Windows NT 5.1; rv:31.0) Gecko/20100101 Firefox/31.0");
-
- ScenarioBuilder warmup = scenario("WarmupSimulation")
- .exec(http("PowerJob-Warmup-HTTP")
- .get("/pressure/ask?protocol=HTTP&debug=false&responseSize=1024"))
- .exec(http("PowerJob-Warmup-AKKA")
- .get("/pressure/ask?protocol=AKKA&debug=false&responseSize=1024"))
- ;
-
- ScenarioBuilder httpAsk = scenario("HttpSimulation") // 7
- .exec(http("PowerJob-Remote-Http") // 请求名称，用于压测报表展示
- .get("/pressure/ask?protocol=HTTP&debug=false&responseSize=1024")) // 9
- ;
-
- ScenarioBuilder akkaAsk = scenario("AkkaSimulation") // 7
- .exec(http("PowerJob-Remote-AKKA") // 请求名称，用于压测报表展示
- .get("/pressure/ask?protocol=AKKA&debug=false&responseSize=1024")) // 9
- ;
-
-
- /*
- atOnceUsers(10) 一次模拟的用户数量(10)
- nothingFor(4 seconds) 在指定的时间段(4 seconds)内什么都不干
- constantUsersPerSec(10) during(20 seconds) 以固定的速度模拟用户，指定每秒模拟的用户数(10)，指定模拟测试时间长度(20 seconds)
- rampUsersPerSec(10) to (20) during(20 seconds) 在指定的时间(20 seconds)内，使每秒模拟的用户从数量1(10)逐渐增加到数量2(20)，速度匀速
- heavisideUsers(100) over(10 seconds) 在指定的时间(10 seconds)内使用类似单位阶跃函数的方法逐渐增加模拟并发的用户，直到总数达到指定的数量(100).简单说就是每秒并发用户数递增
- */
-
- {
- setUp( // 11
- warmup.injectOpen(constantUsersPerSec(50).during(10))
- .andThen(
- httpAsk.injectOpen(incrementUsersPerSec(100).times(10).eachLevelLasting(10))
- )
- .andThen(
- akkaAsk.injectOpen(incrementUsersPerSec(100).times(10).eachLevelLasting(10))
- )
- ).protocols(httpProtocol); // 13
- }
-}
diff --git a/powerjob-remote/powerjob-remote-benchmark/src/test/resources/gatling.conf b/powerjob-remote/powerjob-remote-benchmark/src/test/resources/gatling.conf
deleted file mode 100644
index c6db76269..000000000
--- a/powerjob-remote/powerjob-remote-benchmark/src/test/resources/gatling.conf
+++ /dev/null
@@ -1,127 +0,0 @@
-#########################
-# Gatling Configuration #
-#########################
-
-# This file contains all the settings configurable for Gatling with their default values
-
-gatling {
- core {
- #outputDirectoryBaseName = "" # The prefix for each simulation result folder (then suffixed by the report generation timestamp)
- #runDescription = "" # The description for this simulation run, displayed in each report
- #encoding = "utf-8" # Encoding to use throughout Gatling for file and string manipulation
- #simulationClass = "" # The FQCN of the simulation to run (when used in conjunction with noReports, the simulation for which assertions will be validated)
- #elFileBodiesCacheMaxCapacity = 200 # Cache size for request body EL templates, set to 0 to disable
- #rawFileBodiesCacheMaxCapacity = 200 # Cache size for request body Raw templates, set to 0 to disable
- #rawFileBodiesInMemoryMaxSize = 1000 # Below this limit, raw file bodies will be cached in memory
- #pebbleFileBodiesCacheMaxCapacity = 200 # Cache size for request body Peeble templates, set to 0 to disable
- #feederAdaptiveLoadModeThreshold = 100 # File size threshold (in MB). Below load eagerly in memory, above use batch mode with default buffer size
- #shutdownTimeout = 10000 # Milliseconds to wait for the actor system to shutdown
- extract {
- regex {
- #cacheMaxCapacity = 200 # Cache size for the compiled regexes, set to 0 to disable caching
- }
- xpath {
- #cacheMaxCapacity = 200 # Cache size for the compiled XPath queries, set to 0 to disable caching
- }
- jsonPath {
- #cacheMaxCapacity = 200 # Cache size for the compiled jsonPath queries, set to 0 to disable caching
- }
- css {
- #cacheMaxCapacity = 200 # Cache size for the compiled CSS selectors queries, set to 0 to disable caching
- }
- }
- directory {
- #simulations = user-files/simulations # Directory where simulation classes are located (for bundle packaging only)
- #resources = user-files/resources # Directory where resources, such as feeder files and request bodies are located (for bundle packaging only)
- #reportsOnly = "" # If set, name of report folder to look for in order to generate its report
- #binaries = "" # If set, name of the folder where compiles classes are located: Defaults to GATLING_HOME/target.
- #results = results # Name of the folder where all reports folder are located
- }
- }
- socket {
- #connectTimeout = 10000 # Timeout in millis for establishing a TCP socket
- #tcpNoDelay = true
- #soKeepAlive = false # if TCP keepalive configured at OS level should be used
- #soReuseAddress = false
- }
- netty {
- #useNativeTransport = true # if Netty native transport should be used instead of Java NIO
- #allocator = "pooled" # switch to unpooled for unpooled ByteBufAllocator
- #maxThreadLocalCharBufferSize = 200000 # Netty's default is 16k
- }
- ssl {
- #useOpenSsl = true # if OpenSSL should be used instead of JSSE (only the latter can be debugged with -Djava.net.debug=ssl)
- #useOpenSslFinalizers = false # if OpenSSL contexts should be freed with Finalizer or if using RefCounted is fine
- #handshakeTimeout = 10000 # TLS handshake timeout in millis
- #useInsecureTrustManager = true # Use an insecure TrustManager that trusts all server certificates
- #enabledProtocols = [] # Array of enabled protocols for HTTPS, if empty use Netty's defaults
- #enabledCipherSuites = [] # Array of enabled cipher suites for HTTPS, if empty enable all available ciphers
- #sessionCacheSize = 0 # SSLSession cache size, set to 0 to use JDK's default
- #sessionTimeout = 0 # SSLSession timeout in seconds, set to 0 to use JDK's default (24h)
- #enableSni = true # When set to true, enable Server Name indication (SNI)
- keyStore {
- #type = "" # Type of SSLContext's KeyManagers store
- #file = "" # Location of SSLContext's KeyManagers store
- #password = "" # Password for SSLContext's KeyManagers store
- #algorithm = "" # Algorithm used SSLContext's KeyManagers store
- }
- trustStore {
- #type = "" # Type of SSLContext's TrustManagers store
- #file = "" # Location of SSLContext's TrustManagers store
- #password = "" # Password for SSLContext's TrustManagers store
- #algorithm = "" # Algorithm used by SSLContext's TrustManagers store
- }
- }
- charting {
- #noReports = false # When set to true, don't generate HTML reports
- #maxPlotPerSeries = 1000 # Number of points per graph in Gatling reports
- #useGroupDurationMetric = false # Switch group timings from cumulated response time to group duration.
- indicators {
- #lowerBound = 800 # Lower bound for the requests' response time to track in the reports and the console summary
- #higherBound = 1200 # Higher bound for the requests' response time to track in the reports and the console summary
- #percentile1 = 50 # Value for the 1st percentile to track in the reports, the console summary and Graphite
- #percentile2 = 75 # Value for the 2nd percentile to track in the reports, the console summary and Graphite
- #percentile3 = 95 # Value for the 3rd percentile to track in the reports, the console summary and Graphite
- #percentile4 = 99 # Value for the 4th percentile to track in the reports, the console summary and Graphite
- }
- }
- http {
- #fetchedCssCacheMaxCapacity = 200 # Cache size for CSS parsed content, set to 0 to disable
- #fetchedHtmlCacheMaxCapacity = 200 # Cache size for HTML parsed content, set to 0 to disable
- #perUserCacheMaxCapacity = 200 # Per virtual user cache size, set to 0 to disable
- #warmUpUrl = "https://gatling.io" # The URL to use to warm-up the HTTP stack (blank means disabled)
- #enableGA = true # Very light Google Analytics (Gatling and Java version), please support
- #pooledConnectionIdleTimeout = 60000 # Timeout in millis for a connection to stay idle in the pool
- #requestTimeout = 60000 # Timeout in millis for performing an HTTP request
- #enableHostnameVerification = false # When set to true, enable hostname verification: SSLEngine.setHttpsEndpointIdentificationAlgorithm("HTTPS")
- dns {
- #queryTimeout = 5000 # Timeout in millis of each DNS query in millis
- #maxQueriesPerResolve = 6 # Maximum allowed number of DNS queries for a given name resolution
- }
- }
- jms {
- #replyTimeoutScanPeriod = 1000 # scan period for timedout reply messages
- }
- data {
- #writers = [console, file] # The list of DataWriters to which Gatling write simulation data (currently supported : console, file, graphite)
- console {
- #light = false # When set to true, displays a light version without detailed request stats
- #writePeriod = 5 # Write interval, in seconds
- }
- file {
- #bufferSize = 8192 # FileDataWriter's internal data buffer size, in bytes
- }
- leak {
- #noActivityTimeout = 30 # Period, in seconds, for which Gatling may have no activity before considering a leak may be happening
- }
- graphite {
- #light = false # only send the all* stats
- #host = "localhost" # The host where the Carbon server is located
- #port = 2003 # The port to which the Carbon server listens to (2003 is default for plaintext, 2004 is default for pickle)
- #protocol = "tcp" # The protocol used to send data to Carbon (currently supported : "tcp", "udp")
- #rootPathPrefix = "gatling" # The common prefix of all metrics sent to Graphite
- #bufferSize = 8192 # Internal data buffer size, in bytes
- #writePeriod = 1 # Write period, in seconds
- }
- }
-}
diff --git a/powerjob-remote/powerjob-remote-framework/pom.xml b/powerjob-remote/powerjob-remote-framework/pom.xml
index e63f358a0..8d6c7966f 100644
--- a/powerjob-remote/powerjob-remote-framework/pom.xml
+++ b/powerjob-remote/powerjob-remote-framework/pom.xml
@@ -5,19 +5,20 @@
powerjob-remote
tech.powerjob
- 4.3.6
+ 5.1.2
4.0.0
- 4.3.6
+ 5.1.2
powerjob-remote-framework
+ powerjob-remote-framework
8
8
UTF-8
- 4.3.6
+ 5.1.2
0.10.2
diff --git a/powerjob-remote/powerjob-remote-framework/src/main/java/tech/powerjob/remote/framework/actor/TestActor.java b/powerjob-remote/powerjob-remote-framework/src/main/java/tech/powerjob/remote/framework/actor/TestActor.java
new file mode 100644
index 000000000..c597a5f09
--- /dev/null
+++ b/powerjob-remote/powerjob-remote-framework/src/main/java/tech/powerjob/remote/framework/actor/TestActor.java
@@ -0,0 +1,27 @@
+package tech.powerjob.remote.framework.actor;
+
+import lombok.extern.slf4j.Slf4j;
+
+/**
+ * 内置一个用来通用测试的 TestActor
+ *
+ * @author tjq
+ * @since 2022/12/31
+ */
+@Slf4j
+@Actor(path = "test")
+public class TestActor {
+
+ public static void simpleStaticMethod() {
+ }
+
+ public void simpleMethod() {
+ }
+
+ @Handler(path = "method1")
+ public String handlerMethod1() {
+ log.info("[TestActor] handlerMethod1");
+ return "1";
+ }
+
+}
diff --git a/powerjob-remote/powerjob-remote-framework/src/main/java/tech/powerjob/remote/framework/base/RemotingException.java b/powerjob-remote/powerjob-remote-framework/src/main/java/tech/powerjob/remote/framework/base/RemotingException.java
index fd3185db6..bb881eb27 100644
--- a/powerjob-remote/powerjob-remote-framework/src/main/java/tech/powerjob/remote/framework/base/RemotingException.java
+++ b/powerjob-remote/powerjob-remote-framework/src/main/java/tech/powerjob/remote/framework/base/RemotingException.java
@@ -1,7 +1,5 @@
package tech.powerjob.remote.framework.base;
-import java.io.IOException;
-
/**
* RemotingException
*
@@ -13,4 +11,8 @@ public class RemotingException extends RuntimeException {
public RemotingException(String message) {
super(message);
}
+
+ public RemotingException(String message, Throwable throwable) {
+ super(message, throwable);
+ }
}
diff --git a/powerjob-remote/powerjob-remote-framework/src/main/java/tech/powerjob/remote/framework/cs/CSInitializerConfig.java b/powerjob-remote/powerjob-remote-framework/src/main/java/tech/powerjob/remote/framework/cs/CSInitializerConfig.java
index a64051fd7..0546e2636 100644
--- a/powerjob-remote/powerjob-remote-framework/src/main/java/tech/powerjob/remote/framework/cs/CSInitializerConfig.java
+++ b/powerjob-remote/powerjob-remote-framework/src/main/java/tech/powerjob/remote/framework/cs/CSInitializerConfig.java
@@ -19,7 +19,14 @@
@Accessors(chain = true)
public class CSInitializerConfig implements Serializable {
+ /**
+ * 需要绑定的地址（本地）
+ */
private Address bindAddress;
+ /**
+ * 外部地址（需要 NAT 等情况存在）
+ */
+ private Address externalAddress;
private ServerType serverType;
}
diff --git a/powerjob-remote/powerjob-remote-framework/src/main/java/tech/powerjob/remote/framework/engine/EngineConfig.java b/powerjob-remote/powerjob-remote-framework/src/main/java/tech/powerjob/remote/framework/engine/EngineConfig.java
index a9b82961d..d038114c7 100644
--- a/powerjob-remote/powerjob-remote-framework/src/main/java/tech/powerjob/remote/framework/engine/EngineConfig.java
+++ b/powerjob-remote/powerjob-remote-framework/src/main/java/tech/powerjob/remote/framework/engine/EngineConfig.java
@@ -30,6 +30,10 @@ public class EngineConfig implements Serializable {
* 绑定的本地地址
*/
private Address bindAddress;
+ /**
+ * 外部地址（需要 NAT 等情况存在）
+ */
+ private Address externalAddress;
/**
* actor实例，交由使用侧自己实例化以便自行注入各种 bean
*/
diff --git a/powerjob-remote/powerjob-remote-framework/src/main/java/tech/powerjob/remote/framework/engine/impl/CSInitializerFactory.java b/powerjob-remote/powerjob-remote-framework/src/main/java/tech/powerjob/remote/framework/engine/impl/CSInitializerFactory.java
index 8b02f3d43..7590da18a 100644
--- a/powerjob-remote/powerjob-remote-framework/src/main/java/tech/powerjob/remote/framework/engine/impl/CSInitializerFactory.java
+++ b/powerjob-remote/powerjob-remote-framework/src/main/java/tech/powerjob/remote/framework/engine/impl/CSInitializerFactory.java
@@ -4,9 +4,11 @@
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.reflections.Reflections;
import tech.powerjob.common.OmsConstant;
+import tech.powerjob.common.enums.Protocol;
import tech.powerjob.common.exception.PowerJobException;
import tech.powerjob.remote.framework.cs.CSInitializer;
+import java.util.Optional;
import java.util.Set;
/**
@@ -18,8 +20,26 @@
@Slf4j
class CSInitializerFactory {
+ private static final String OFFICIAL_HTTP_CS_INITIALIZER = "tech.powerjob.remote.http.HttpVertxCSInitializer";
+ /**
+ * 未来底层框架摆脱 vertx 时可能会用这个 classname，or 开发者自己实现的 http 协议也可以用这个 classname，总之预留战未来
+ */
+ private static final String OFFICIAL_HTTP_CS_INITIALIZER2 = "tech.powerjob.remote.http.HttpCSInitializer";
+ private static final String OFFICIAL_AKKA_CS_INITIALIZER = "tech.powerjob.remote.akka.AkkaCSInitializer";
+ private static final String OFFICIAL_MU_CS_INITIALIZER = "tech.powerjob.remote.mu.MuCSInitializer";
+
+ private static final String EXTEND_CS_INITIALIZER_PATTERN = "tech.powerjob.remote.%s.CSInitializer";
+
static CSInitializer build(String targetType) {
+ CSInitializer officialCSInitializer = tryLoadCSInitializerByClassName(targetType);
+ if (officialCSInitializer != null) {
+ return officialCSInitializer;
+ }
+
+ log.info("[CSInitializerFactory] try load CSInitializerFactory by name failed, start to use Reflections!");
+
+ // JAVA SPI 机制太笨了，短期内继续保留 Reflections 官网下高版本兼容性
Reflections reflections = new Reflections(OmsConstant.PACKAGE);
Set> cSInitializerClzSet = reflections.getSubTypesOf(CSInitializer.class);
@@ -41,4 +61,59 @@ static CSInitializer build(String targetType) {
throw new PowerJobException(String.format("can't load CSInitializer[%s], ensure your package name start with 'tech.powerjob' and import the dependencies!", targetType));
}
+
+ /**
+ * 官方组件直接使用固定类名尝试加载，确保 reflections 不兼容情况下，至少能使用官方通讯协议
+ * @param targetType 协议类型
+ * @return CSInitializer
+ */
+ private static CSInitializer tryLoadCSInitializerByClassName(String targetType) {
+
+ if (Protocol.HTTP.name().equalsIgnoreCase(targetType)) {
+ Optional httpCsIOpt = tryLoadCSInitializerByClzName(OFFICIAL_HTTP_CS_INITIALIZER);
+ if (httpCsIOpt.isPresent()) {
+ return httpCsIOpt.get();
+ }
+ Optional httpCsIOpt2 = tryLoadCSInitializerByClzName(OFFICIAL_HTTP_CS_INITIALIZER2);
+ if (httpCsIOpt2.isPresent()) {
+ return httpCsIOpt2.get();
+ }
+ }
+
+ if (Protocol.AKKA.name().equalsIgnoreCase(targetType)) {
+ Optional akkaCSIOpt = tryLoadCSInitializerByClzName(OFFICIAL_AKKA_CS_INITIALIZER);
+ if (akkaCSIOpt.isPresent()) {
+ return akkaCSIOpt.get();
+ }
+ }
+
+ if (Protocol.MU.name().equalsIgnoreCase(targetType)) {
+ Optional muCSIOpt = tryLoadCSInitializerByClzName(OFFICIAL_MU_CS_INITIALIZER);
+ if (muCSIOpt.isPresent()) {
+ return muCSIOpt.get();
+ }
+ }
+
+ // 尝试加载按规范命名的处理器，比如使用方自定义了 http2 协议，将其类名定为 tech.powerjob.remote.http2.CSInitializer 依然可确保在 Reflections 不可用的情况下完成加载
+ String clz = String.format(EXTEND_CS_INITIALIZER_PATTERN, targetType);
+ Optional extOpt = tryLoadCSInitializerByClzName(clz);
+ return extOpt.orElse(null);
+
+ }
+
+
+ private static Optional tryLoadCSInitializerByClzName(String clzName) {
+ try {
+ log.info("[CSInitializerFactory] try to load CSInitializer by classname: {}", clzName);
+ Class clz = Class.forName(clzName);
+ CSInitializer o = (CSInitializer) clz.getDeclaredConstructor().newInstance();
+ log.info("[CSInitializerFactory] load CSInitializer[{}] successfully, obj: {}", clzName, o);
+ return Optional.of(o);
+ } catch (ClassNotFoundException ce) {
+ log.warn("[CSInitializerFactory] load CSInitializer by classname[{}] failed due to ClassNotFound: {}", clzName, ExceptionUtils.getMessage(ce));
+ } catch (Exception e) {
+ log.warn("[CSInitializerFactory] load CSInitializer by classname[{}] failed.", clzName, e);
+ }
+ return Optional.empty();
+ }
}
diff --git a/powerjob-remote/powerjob-remote-framework/src/main/java/tech/powerjob/remote/framework/engine/impl/PowerJobRemoteEngine.java b/powerjob-remote/powerjob-remote-framework/src/main/java/tech/powerjob/remote/framework/engine/impl/PowerJobRemoteEngine.java
index 2f9cbc6b8..7ace6a4e4 100644
--- a/powerjob-remote/powerjob-remote-framework/src/main/java/tech/powerjob/remote/framework/engine/impl/PowerJobRemoteEngine.java
+++ b/powerjob-remote/powerjob-remote-framework/src/main/java/tech/powerjob/remote/framework/engine/impl/PowerJobRemoteEngine.java
@@ -2,7 +2,9 @@
import com.google.common.base.Stopwatch;
import lombok.extern.slf4j.Slf4j;
+import tech.powerjob.common.utils.SysUtils;
import tech.powerjob.remote.framework.actor.ActorInfo;
+import tech.powerjob.remote.framework.actor.TestActor;
import tech.powerjob.remote.framework.cs.CSInitializer;
import tech.powerjob.remote.framework.cs.CSInitializerConfig;
import tech.powerjob.remote.framework.engine.EngineConfig;
@@ -27,6 +29,8 @@ public class PowerJobRemoteEngine implements RemoteEngine {
@Override
public EngineOutput start(EngineConfig engineConfig) {
+ reConfig(engineConfig);
+
final String engineType = engineConfig.getType();
EngineOutput engineOutput = new EngineOutput();
log.info("[PowerJobRemoteEngine] [{}] start remote engine with config: {}", engineType, engineConfig);
@@ -41,6 +45,7 @@ public EngineOutput start(EngineConfig engineConfig) {
csInitializer.init(new CSInitializerConfig()
.setBindAddress(engineConfig.getBindAddress())
+ .setExternalAddress(engineConfig.getExternalAddress())
.setServerType(engineConfig.getServerType())
);
@@ -63,4 +68,12 @@ public EngineOutput start(EngineConfig engineConfig) {
public void close() throws IOException {
csInitializer.close();
}
+
+ private static void reConfig(EngineConfig engineConfig) {
+ boolean testEnv = SysUtils.isTestEnv();
+ if (testEnv) {
+ log.info("[PowerJobRemoteEngine] add 'TestActor' due to current is test env");
+ engineConfig.getActorList().add(new TestActor());
+ }
+ }
}
diff --git a/powerjob-remote/powerjob-remote-framework/src/test/java/tech/powerjob/remote/framework/engine/impl/ActorFactoryTest.java b/powerjob-remote/powerjob-remote-framework/src/test/java/tech/powerjob/remote/framework/engine/impl/ActorFactoryTest.java
index 7e80d2aef..7e604b4d8 100644
--- a/powerjob-remote/powerjob-remote-framework/src/test/java/tech/powerjob/remote/framework/engine/impl/ActorFactoryTest.java
+++ b/powerjob-remote/powerjob-remote-framework/src/test/java/tech/powerjob/remote/framework/engine/impl/ActorFactoryTest.java
@@ -3,12 +3,7 @@
import com.google.common.collect.Lists;
import lombok.extern.slf4j.Slf4j;
import org.junit.jupiter.api.Test;
-import tech.powerjob.remote.framework.actor.ActorInfo;
-import tech.powerjob.remote.framework.test.TestActor;
-
-import java.util.List;
-import java.util.Set;
-import java.util.stream.Collectors;
+import tech.powerjob.remote.framework.actor.TestActor;
/**
* HandlerFactoryTest
diff --git a/powerjob-remote/powerjob-remote-framework/src/test/java/tech/powerjob/remote/framework/test/TestActor.java b/powerjob-remote/powerjob-remote-framework/src/test/java/tech/powerjob/remote/framework/test/TestActor.java
deleted file mode 100644
index a1d3351ad..000000000
--- a/powerjob-remote/powerjob-remote-framework/src/test/java/tech/powerjob/remote/framework/test/TestActor.java
+++ /dev/null
@@ -1,42 +0,0 @@
-package tech.powerjob.remote.framework.test;
-
-import lombok.extern.slf4j.Slf4j;
-import tech.powerjob.remote.framework.actor.Actor;
-import tech.powerjob.remote.framework.actor.Handler;
-
-import java.util.Map;
-
-/**
- * TestActor
- *
- * @author tjq
- * @since 2022/12/31
- */
-@Slf4j
-@Actor(path = "/test")
-public class TestActor {
-
- public static void simpleStaticMethod() {
- }
-
- public void simpleMethod() {
- }
-
- @Handler(path = "/method1")
- public String handlerMethod1() {
- log.info("[TestActor] handlerMethod1");
- return "1";
- }
-
- @Handler(path = "/method2")
- public String handlerMethod2(String name) {
- log.info("[TestActor] handlerMethod2 req: {}", name);
- return name;
- }
-
- @Handler(path = "/returnEmpty")
- public void handlerEmpty(Map req) {
- log.info("[TestActor] handlerEmpty req: {}", req);
- }
-
-}
diff --git a/powerjob-remote/powerjob-remote-impl-akka/pom.xml b/powerjob-remote/powerjob-remote-impl-akka/pom.xml
index 1ee9b4f41..f622342b4 100644
--- a/powerjob-remote/powerjob-remote-impl-akka/pom.xml
+++ b/powerjob-remote/powerjob-remote-impl-akka/pom.xml
@@ -5,19 +5,20 @@
powerjob-remote
tech.powerjob
- 4.3.6
+ 5.1.2
4.0.0
powerjob-remote-impl-akka
- 4.3.6
+ powerjob-remote-impl-akka
+ 5.1.2
8
8
UTF-8
- 4.3.6
+ 5.1.2
2.6.13
diff --git a/powerjob-remote/powerjob-remote-impl-akka/src/main/java/tech/powerjob/remote/akka/AkkaCSInitializer.java b/powerjob-remote/powerjob-remote-impl-akka/src/main/java/tech/powerjob/remote/akka/AkkaCSInitializer.java
index cad05a8f9..42faeca3c 100644
--- a/powerjob-remote/powerjob-remote-impl-akka/src/main/java/tech/powerjob/remote/akka/AkkaCSInitializer.java
+++ b/powerjob-remote/powerjob-remote-impl-akka/src/main/java/tech/powerjob/remote/akka/AkkaCSInitializer.java
@@ -9,7 +9,9 @@
import com.typesafe.config.Config;
import com.typesafe.config.ConfigFactory;
import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
import tech.powerjob.common.serialize.JsonUtils;
+import tech.powerjob.common.utils.SysUtils;
import tech.powerjob.remote.framework.actor.ActorInfo;
import tech.powerjob.remote.framework.base.Address;
import tech.powerjob.remote.framework.cs.CSInitializer;
@@ -47,8 +49,22 @@ public void init(CSInitializerConfig config) {
// 初始化 ActorSystem（macOS上 new ServerSocket 检测端口占用的方法并不生效，可能是AKKA是Scala写的缘故？没办法...只能靠异常重试了）
Map overrideConfig = Maps.newHashMap();
- overrideConfig.put("akka.remote.artery.canonical.hostname", bindAddress.getHost());
- overrideConfig.put("akka.remote.artery.canonical.port", bindAddress.getPort());
+
+ Address externalAddress = config.getExternalAddress();
+
+ if (externalAddress == null || StringUtils.equalsIgnoreCase(externalAddress.toFullAddress(), bindAddress.toFullAddress())) {
+ overrideConfig.put("akka.remote.artery.canonical.hostname", bindAddress.getHost());
+ overrideConfig.put("akka.remote.artery.canonical.port", bindAddress.getPort());
+ log.info("[PowerJob-AKKA] not exist externalIp, overrideConfig: {}", overrideConfig);
+ } else {
+ overrideConfig.put("akka.remote.artery.canonical.hostname", externalAddress.getHost());
+ overrideConfig.put("akka.remote.artery.canonical.port", externalAddress.getPort());
+
+ overrideConfig.put("akka.remote.artery.bind.hostname", "0.0.0.0");
+ overrideConfig.put("akka.remote.artery.bind.port", bindAddress.getPort());
+
+ log.info("[PowerJob-AKKA] exist externalAddress[{}], final overrideConfig: {}", externalAddress, overrideConfig);
+ }
Config akkaBasicConfig = ConfigFactory.load(AkkaConstant.AKKA_CONFIG);
Config akkaFinalConfig = ConfigFactory.parseMap(overrideConfig).withFallback(akkaBasicConfig);
@@ -73,7 +89,7 @@ public Transporter buildTransporter() {
@Override
public void bindHandlers(List actorInfos) {
- int cores = Runtime.getRuntime().availableProcessors();
+ int cores = SysUtils.availableProcessors();
actorInfos.forEach(actorInfo -> {
String rootPath = actorInfo.getAnno().path();
AkkaMappingService.ActorConfig actorConfig = AkkaMappingService.parseActorName(rootPath);
diff --git a/powerjob-remote/powerjob-remote-impl-http/pom.xml b/powerjob-remote/powerjob-remote-impl-http/pom.xml
index 84577fadf..d82c52758 100644
--- a/powerjob-remote/powerjob-remote-impl-http/pom.xml
+++ b/powerjob-remote/powerjob-remote-impl-http/pom.xml
@@ -5,12 +5,13 @@
powerjob-remote
tech.powerjob
- 4.3.6
+ 5.1.2
4.0.0
powerjob-remote-impl-http
- 4.3.6
+ powerjob-remote-impl-http
+ 5.1.2
8
@@ -18,7 +19,7 @@
UTF-8
4.3.7
- 4.3.6
+ 5.1.2
diff --git a/powerjob-remote/powerjob-remote-impl-http/src/main/java/tech/powerjob/remote/http/HttpVertxCSInitializer.java b/powerjob-remote/powerjob-remote-impl-http/src/main/java/tech/powerjob/remote/http/HttpVertxCSInitializer.java
index 9d37a6e8e..14fcfcdac 100644
--- a/powerjob-remote/powerjob-remote-impl-http/src/main/java/tech/powerjob/remote/http/HttpVertxCSInitializer.java
+++ b/powerjob-remote/powerjob-remote-impl-http/src/main/java/tech/powerjob/remote/http/HttpVertxCSInitializer.java
@@ -1,5 +1,8 @@
package tech.powerjob.remote.http;
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.core.JsonParser;
+import com.fasterxml.jackson.databind.DeserializationFeature;
import io.netty.handler.codec.http.HttpResponseStatus;
import io.vertx.core.Handler;
import io.vertx.core.Vertx;
@@ -13,6 +16,7 @@
import io.vertx.ext.web.handler.BodyHandler;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.ArrayUtils;
import tech.powerjob.common.exception.PowerJobException;
import tech.powerjob.remote.framework.actor.ActorInfo;
import tech.powerjob.remote.framework.actor.HandlerInfo;
@@ -38,6 +42,8 @@
* - vertx 唯一的缺点是其作为相对上层的框架，可能存在较为严重的包冲突问题，尤其是对于那些本身跑在 vertx-framework 上的用户
* - 不过该问题可以通过更换协议解决，预计后续提供一个基于 netty 和自定义协议的实现
*
+ * 20240316 note：注意类名被强依赖，后续若有改动需要同步更改
+ *
* @author tjq
* @since 2022/12/31
*/
@@ -58,6 +64,19 @@ public String type() {
@Override
public void init(CSInitializerConfig config) {
this.config = config;
+
+ // 【Vertx 版本升级时必须注意】临时解决 vertx 自带的 jackson 序列化无法支持字段升级问题（默认特性居然是不支持增删字段的序列化方式，外国框架也是一坨...）
+ try {
+ io.vertx.core.json.jackson.DatabindCodec.mapper()
+ .configure(com.fasterxml.jackson.databind.MapperFeature.PROPAGATE_TRANSIENT_MARKER, true)
+ .configure(JsonParser.Feature.ALLOW_SINGLE_QUOTES, true)
+ .configure(JsonParser.Feature.IGNORE_UNDEFINED, true)
+ .configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false)
+ .setSerializationInclusion(JsonInclude.Include.NON_NULL);
+ } catch (Throwable t) {
+ log.warn("[HttpVertxCSInitializer] hack jackson failed!", t);
+ }
+
vertx = VertxInitializer.buildVertx();
httpServer = VertxInitializer.buildHttpServer(vertx);
httpClient = VertxInitializer.buildHttpClient(vertx);
@@ -108,8 +127,10 @@ private Handler buildRequestHandler(ActorInfo actorInfo, Handler
Optional> powerSerializeClz = RemoteUtils.findPowerSerialize(method.getParameterTypes());
// 内部框架，严格模式，绑定失败直接报错
- if (!powerSerializeClz.isPresent()) {
- throw new PowerJobException("can't find any 'PowerSerialize' object in handler args: " + handlerInfo.getLocation());
+ if (ArrayUtils.isNotEmpty(method.getParameterTypes())) {
+ if (!powerSerializeClz.isPresent()) {
+ throw new PowerJobException("can't find any 'PowerSerialize' object in handler args: " + handlerInfo.getLocation());
+ }
}
return ctx -> {
diff --git a/powerjob-remote/powerjob-remote-impl-http/src/main/java/tech/powerjob/remote/http/vertx/VertxInitializer.java b/powerjob-remote/powerjob-remote-impl-http/src/main/java/tech/powerjob/remote/http/vertx/VertxInitializer.java
index 8d0eda3dc..1dfb34dc8 100644
--- a/powerjob-remote/powerjob-remote-impl-http/src/main/java/tech/powerjob/remote/http/vertx/VertxInitializer.java
+++ b/powerjob-remote/powerjob-remote-impl-http/src/main/java/tech/powerjob/remote/http/vertx/VertxInitializer.java
@@ -10,6 +10,7 @@
import org.apache.commons.lang3.StringUtils;
import tech.powerjob.common.OmsConstant;
import tech.powerjob.common.PowerJobDKey;
+import tech.powerjob.common.utils.SysUtils;
/**
* VertxInitializer
@@ -31,7 +32,7 @@ public class VertxInitializer {
private static final int SERVER_IDLE_TIMEOUT_S = 300;
public static Vertx buildVertx() {
- final int cpuCores = Runtime.getRuntime().availableProcessors();
+ final int cpuCores = SysUtils.availableProcessors();
VertxOptions options = new VertxOptions()
.setWorkerPoolSize(Math.max(16, 2 * cpuCores))
.setInternalBlockingPoolSize(Math.max(32, 4 * cpuCores));
@@ -63,7 +64,7 @@ public static HttpClient buildHttpClient(Vertx vertx) {
HttpClientOptions httpClientOptions = new HttpClientOptions()
.setMetricsName(OmsConstant.PACKAGE)
.setConnectTimeout(CONNECTION_TIMEOUT_MS)
- .setMaxPoolSize(Math.max(8, Runtime.getRuntime().availableProcessors()) * 2);
+ .setMaxPoolSize(Math.max(8, SysUtils.availableProcessors()) * 2);
// 长连接
String keepaliveTimeout = System.getProperty(PowerJobDKey.TRANSPORTER_KEEP_ALIVE_TIMEOUT, String.valueOf(DEFAULT_KEEP_ALIVE_TIMEOUT));
diff --git a/powerjob-remote/powerjob-remote-impl-mu/DESIGN.md b/powerjob-remote/powerjob-remote-impl-mu/DESIGN.md
new file mode 100644
index 000000000..479c48827
--- /dev/null
+++ b/powerjob-remote/powerjob-remote-impl-mu/DESIGN.md
@@ -0,0 +1,563 @@
+# PowerJob Mu 协议技术设计方案
+
+## 1. 项目背景
+
+### 1.1 业务场景
+PowerJob 是一个分布式任务调度框架，采用 Server-Worker 架构。在实际部署中，经常遇到以下网络环境限制：
+- Worker 节点部署在内网环境，只能单向访问公网的 Server
+- Server 无法直接访问 Worker 的内网 IP 地址
+- 业务需要支持 Server 向 Worker 的主动通讯（任务下发、状态查询等）
+
+### 1.2 现有协议局限性
+- **HTTP 协议**：基于请求-响应模式，Server 向 Worker 推送需要 Worker 轮询，实时性差
+- **AKKA 协议**：功能强大但配置复杂，在 NAT 环境下需要复杂的网络配置
+
+### 1.3 设计目标
+1. 支持 Worker 仅可出站访问的网络环境
+2. 实现 Server 与 Worker 的双向通讯
+3. 支持完整的节点间通讯矩阵（Worker↔Server, Worker↔Worker, Server↔Server）
+4. 保持高性能和低延迟
+5. 简化网络配置和部署复杂度
+
+## 2. 总体设计
+
+### 2.1 核心设计原则
+
+#### 连接复用 (Connection Reuse)
+- Worker 主动建立到 Server 的长连接
+- Server 通过注册机制维护 Worker 连接映射
+- Server 复用 Worker 建立的连接进行反向通讯
+
+#### 延迟连接 (Lazy Connection)
+- 节点启动时不立即建立所有连接
+- 首次通讯时才建立目标连接
+- 避免启动时的网络依赖和连接失败
+
+#### 统一架构 (Unified Architecture)
+- 所有节点都具备 Server 和 Client 双重能力
+- 统一的消息格式和处理流程
+- 支持任意节点间的直接通讯
+
+### 2.2 架构图
+
+```
+ PowerJob Mu Protocol Architecture
+
+ ┌──────────────────────────────────────────────────────────────────┐
+ │ Application Layer │
+ ├──────────────────────────────────────────────────────────────────┤
+ │ Actor System (Handler Registration & Message Routing) │
+ ├──────────────────────────────────────────────────────────────────┤
+ │ Transport Layer │
+ │ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │
+ │ │MuTransporter│ │ChannelMgr │ │ConnectionMgr│ │
+ │ │ │ │ │ │ │ │
+ │ │- tell() │ │- Worker Reg │ │- Lazy Conn │ │
+ │ │- ask() │ │- Ask/Resp │ │- Conn Pool │ │
+ │ └─────────────┘ └─────────────┘ └─────────────┘ │
+ ├──────────────────────────────────────────────────────────────────┤
+ │ Protocol Layer │
+ │ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │
+ │ │ MuMessage │ │MuMessageCodec│ │Message Types│ │
+ │ │ │ │ │ │ │ │
+ │ │- Type │ │- Encode │ │- TELL │ │
+ │ │- RequestId │ │- Decode │ │- ASK │ │
+ │ │- Path │ │- Length │ │- RESPONSE │ │
+ │ │- Payload │ │- JSON │ │- HEARTBEAT │ │
+ │ └─────────────┘ └─────────────┘ └─────────────┘ │
+ ├──────────────────────────────────────────────────────────────────┤
+ │ Network Layer │
+ │ Netty Framework │
+ │ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │
+ │ │EventLoopGroup│ │ Bootstrap │ │ChannelPipe │ │
+ │ │ │ │ │ │ │ │
+ │ │- Boss │ │- Server │ │- Codec │ │
+ │ │- Worker │ │- Client │ │- Handler │ │
+ │ └─────────────┘ └─────────────┘ └─────────────┘ │
+ └──────────────────────────────────────────────────────────────────┘
+```
+
+## 3. 详细设计
+
+### 3.1 通讯流程设计
+
+#### 3.1.1 Worker 注册流程
+```mermaid
+sequenceDiagram
+ participant W as Worker
+ participant S as Server
+
+ W->>S: 1. 建立TCP连接
+ W->>S: 2. 发送HEARTBEAT消息 (包含Worker地址)
+ S->>S: 3. 注册Worker连接到ChannelManager
+ S->>S: 4. 建立 workerAddress -> Channel 映射
+
+ Note over W,S: 连接建立完成，可以双向通讯
+
+ W->>S: 5. 定期发送HEARTBEAT (保活)
+ S->>W: 6. 可以通过已注册连接发送消息
+```
+
+#### 3.1.2 延迟连接流程
+```mermaid
+sequenceDiagram
+ participant W1 as Worker1
+ participant W2 as Worker2
+
+ W1->>W1: 1. 需要向Worker2发送消息
+ W1->>W1: 2. 检查连接池，无现有连接
+ W1->>W2: 3. 建立新的TCP连接
+ W1->>W2: 4. 发送消息
+ W1->>W1: 5. 连接加入连接池，供后续复用
+```
+
+### 3.2 核心组件设计
+
+#### 3.2.1 MuCSInitializer
+```java
+public class MuCSInitializer implements CSInitializer {
+
+ // 根据节点类型初始化不同服务
+ public void init(CSInitializerConfig config) {
+ if (config.getServerType() == ServerType.SERVER) {
+ initServer(); // 启动Netty服务端 + 连接管理器
+ } else {
+ initWorker(); // 启动Netty服务端 + 连接管理器
+ }
+ }
+
+ private void initServer() {
+ // 1. 创建ServerHandler
+ // 2. 启动Netty服务端监听
+ // 3. 初始化连接管理器
+ }
+
+ private void initWorker() {
+ // 1. 创建WorkerHandler
+ // 2. 启动Netty服务端监听 (支持Worker间通讯)
+ // 3. 初始化连接管理器
+ }
+}
+```
+
+#### 3.2.2 MuTransporter
+```java
+public class MuTransporter implements Transporter {
+
+ public void tell(URL url, PowerSerializable request) {
+ if (当前节点是Worker) {
+ // 使用连接管理器建立到目标的连接
+ connectionManager.getOrCreateConnection(url.getAddress())
+ .thenAccept(channel -> channel.writeAndFlush(message));
+ } else {
+ // Server端需要区分目标类型
+ if (目标是Worker) {
+ // 使用已注册的Worker连接
+ Channel channel = channelManager.getWorkerChannel(url.getAddress());
+ channel.writeAndFlush(message);
+ } else {
+ // 使用连接管理器连接到目标Server
+ connectionManager.getOrCreateConnection(url.getAddress())
+ .thenAccept(channel -> channel.writeAndFlush(message));
+ }
+ }
+ }
+
+ public CompletionStage ask(URL url, PowerSerializable request, Class clz) {
+ // 1. 生成唯一的requestId
+ // 2. 注册Future到ChannelManager
+ // 3. 按照tell的逻辑发送ASK消息
+ // 4. 返回Future等待响应
+ }
+}
+```
+
+#### 3.2.3 ChannelManager
+```java
+public class ChannelManager {
+
+ // Worker地址到连接的映射 (Server端用)
+ private final ConcurrentMap workerChannels;
+
+ // 请求ID到Future的映射 (Ask模式用)
+ private final ConcurrentMap> pendingRequests;
+
+ // 请求ID到响应类型的映射 (类型转换用)
+ private final ConcurrentMap> requestResponseTypes;
+
+ public void registerWorkerChannel(Address workerAddress, Channel channel) {
+ String key = workerAddress.getHost() + ":" + workerAddress.getPort();
+ workerChannels.put(key, channel);
+ // 监听连接关闭，自动清理映射
+ }
+
+ public void completePendingRequest(String requestId, Object response) {
+ CompletableFuture