diff --git a/.DS_Store b/.DS_Store new file mode 100644 index 00000000..0e8ca631 Binary files /dev/null and b/.DS_Store differ diff --git a/oh-my-scheduler-client/pom.xml b/oh-my-scheduler-client/pom.xml index 13608555..36051002 100644 --- a/oh-my-scheduler-client/pom.xml +++ b/oh-my-scheduler-client/pom.xml @@ -16,7 +16,6 @@ 1.0.0-SNAPSHOT 5.6.1 - 1.2.68 @@ -26,12 +25,6 @@ oh-my-scheduler-common ${oms.common.version} - - - com.alibaba - fastjson - ${fastjson.version} - org.junit.jupiter diff --git a/oh-my-scheduler-client/src/main/java/com/github/kfcfans/oms/client/OhMyClient.java b/oh-my-scheduler-client/src/main/java/com/github/kfcfans/oms/client/OhMyClient.java index 15a07b55..c63fa41e 100644 --- a/oh-my-scheduler-client/src/main/java/com/github/kfcfans/oms/client/OhMyClient.java +++ b/oh-my-scheduler-client/src/main/java/com/github/kfcfans/oms/client/OhMyClient.java @@ -1,9 +1,9 @@ package com.github.kfcfans.oms.client; -import com.alibaba.fastjson.JSONObject; import com.github.kfcfans.common.OpenAPIConstant; import com.github.kfcfans.common.response.ResultDTO; import com.github.kfcfans.common.utils.HttpUtils; +import com.github.kfcfans.common.utils.JsonUtils; import lombok.extern.slf4j.Slf4j; import okhttp3.FormBody; import okhttp3.RequestBody; @@ -43,7 +43,7 @@ public class OhMyClient { String url = getUrl(OpenAPIConstant.ASSERT) + "?appName=" + appName; String result = HttpUtils.get(url); if (StringUtils.isNotEmpty(result)) { - ResultDTO resultDTO = JSONObject.parseObject(result, ResultDTO.class); + ResultDTO resultDTO = JsonUtils.parseObject(result, ResultDTO.class); if (resultDTO.isSuccess()) { appId = Long.parseLong(resultDTO.getData().toString()); }else { @@ -72,7 +72,7 @@ public class OhMyClient { .add("appId", appId.toString()) .build(); String post = HttpUtils.post(url, body); - return JSONObject.parseObject(post, ResultDTO.class); + return JsonUtils.parseObject(post, ResultDTO.class); } /** @@ -88,7 +88,7 @@ public class OhMyClient { .add("appId", appId.toString()) .build(); String post = HttpUtils.post(url, body); - return JSONObject.parseObject(post, ResultDTO.class); + return JsonUtils.parseObject(post, ResultDTO.class); } /** @@ -108,7 +108,7 @@ public class OhMyClient { builder.add("instanceParams", instanceParams); } String post = HttpUtils.post(url, builder.build()); - return JSONObject.parseObject(post, ResultDTO.class); + return JsonUtils.parseObject(post, ResultDTO.class); } public ResultDTO runJob(Long jobId) throws Exception { return runJob(jobId, null); @@ -128,7 +128,7 @@ public class OhMyClient { .add("appId", appId.toString()) .build(); String post = HttpUtils.post(url, body); - return JSONObject.parseObject(post, ResultDTO.class); + return JsonUtils.parseObject(post, ResultDTO.class); } /** @@ -143,6 +143,6 @@ public class OhMyClient { .add("instanceId", instanceId.toString()) .build(); String post = HttpUtils.post(url, body); - return JSONObject.parseObject(post, ResultDTO.class); + return JsonUtils.parseObject(post, ResultDTO.class); } } diff --git a/oh-my-scheduler-common/pom.xml b/oh-my-scheduler-common/pom.xml index b80f7089..7a165d10 100644 --- a/oh-my-scheduler-common/pom.xml +++ b/oh-my-scheduler-common/pom.xml @@ -18,6 +18,7 @@ 3.10 28.2-jre 4.4.1 + 2.6.4 @@ -48,6 +49,18 @@ okhttp ${okhttp.version} + + + + com.typesafe.akka + akka-remote_2.13 + ${akka.version} + + + com.typesafe.akka + akka-serialization-jackson_2.13 + ${akka.version} + \ No newline at end of file diff --git a/oh-my-scheduler-common/src/main/java/com/github/kfcfans/common/OmsSerializable.java b/oh-my-scheduler-common/src/main/java/com/github/kfcfans/common/OmsSerializable.java new file mode 100644 index 00000000..26a8b427 --- /dev/null +++ b/oh-my-scheduler-common/src/main/java/com/github/kfcfans/common/OmsSerializable.java @@ -0,0 +1,12 @@ +package com.github.kfcfans.common; + +import java.io.Serializable; + +/** + * OMS 序列化接口 + * + * @author tjq + * @since 2020/4/16 + */ +public interface OmsSerializable extends Serializable { +} diff --git a/oh-my-scheduler-common/src/main/java/com/github/kfcfans/common/request/ServerQueryInstanceStatusReq.java b/oh-my-scheduler-common/src/main/java/com/github/kfcfans/common/request/ServerQueryInstanceStatusReq.java index 3c900192..06022a9b 100644 --- a/oh-my-scheduler-common/src/main/java/com/github/kfcfans/common/request/ServerQueryInstanceStatusReq.java +++ b/oh-my-scheduler-common/src/main/java/com/github/kfcfans/common/request/ServerQueryInstanceStatusReq.java @@ -1,11 +1,10 @@ package com.github.kfcfans.common.request; +import com.github.kfcfans.common.OmsSerializable; import lombok.AllArgsConstructor; import lombok.Data; import lombok.NoArgsConstructor; -import java.io.Serializable; - /** * 服务器查询实例运行状态，需要返回详细的运行数据 * @@ -15,6 +14,6 @@ import java.io.Serializable; @Data @NoArgsConstructor @AllArgsConstructor -public class ServerQueryInstanceStatusReq implements Serializable { +public class ServerQueryInstanceStatusReq implements OmsSerializable { private Long instanceId; } diff --git a/oh-my-scheduler-common/src/main/java/com/github/kfcfans/common/request/ServerScheduleJobReq.java b/oh-my-scheduler-common/src/main/java/com/github/kfcfans/common/request/ServerScheduleJobReq.java index 8eb92fdb..831d4201 100644 --- a/oh-my-scheduler-common/src/main/java/com/github/kfcfans/common/request/ServerScheduleJobReq.java +++ b/oh-my-scheduler-common/src/main/java/com/github/kfcfans/common/request/ServerScheduleJobReq.java @@ -1,8 +1,8 @@ package com.github.kfcfans.common.request; +import com.github.kfcfans.common.OmsSerializable; import lombok.Data; -import java.io.Serializable; import java.util.List; /** @@ -12,7 +12,7 @@ import java.util.List; * @since 2020/3/17 */ @Data -public class ServerScheduleJobReq implements Serializable { +public class ServerScheduleJobReq implements OmsSerializable { // 可用处理器地址，可能多值，逗号分隔 private List allWorkerAddress; diff --git a/oh-my-scheduler-common/src/main/java/com/github/kfcfans/common/request/ServerStopInstanceReq.java b/oh-my-scheduler-common/src/main/java/com/github/kfcfans/common/request/ServerStopInstanceReq.java index cc27ac19..161ddcbf 100644 --- a/oh-my-scheduler-common/src/main/java/com/github/kfcfans/common/request/ServerStopInstanceReq.java +++ b/oh-my-scheduler-common/src/main/java/com/github/kfcfans/common/request/ServerStopInstanceReq.java @@ -1,10 +1,10 @@ package com.github.kfcfans.common.request; +import com.github.kfcfans.common.OmsSerializable; import lombok.AllArgsConstructor; import lombok.Data; import lombok.NoArgsConstructor; -import java.io.Serializable; /** * 服务器要求任务实例停止执行请求 @@ -15,6 +15,6 @@ import java.io.Serializable; @Data @NoArgsConstructor @AllArgsConstructor -public class ServerStopInstanceReq implements Serializable { +public class ServerStopInstanceReq implements OmsSerializable { private Long instanceId; } diff --git a/oh-my-scheduler-common/src/main/java/com/github/kfcfans/common/request/TaskTrackerReportInstanceStatusReq.java b/oh-my-scheduler-common/src/main/java/com/github/kfcfans/common/request/TaskTrackerReportInstanceStatusReq.java index 6babdbe3..8e07c676 100644 --- a/oh-my-scheduler-common/src/main/java/com/github/kfcfans/common/request/TaskTrackerReportInstanceStatusReq.java +++ b/oh-my-scheduler-common/src/main/java/com/github/kfcfans/common/request/TaskTrackerReportInstanceStatusReq.java @@ -1,8 +1,8 @@ package com.github.kfcfans.common.request; +import com.github.kfcfans.common.OmsSerializable; import lombok.Data; -import java.io.Serializable; /** * TaskTracker 将状态上报给服务器 @@ -11,7 +11,7 @@ import java.io.Serializable; * @since 2020/3/17 */ @Data -public class TaskTrackerReportInstanceStatusReq implements Serializable { +public class TaskTrackerReportInstanceStatusReq implements OmsSerializable { private Long jobId; private Long instanceId; diff --git a/oh-my-scheduler-common/src/main/java/com/github/kfcfans/common/request/WorkerHeartbeat.java b/oh-my-scheduler-common/src/main/java/com/github/kfcfans/common/request/WorkerHeartbeat.java index 9647bc8e..a4a0d37e 100644 --- a/oh-my-scheduler-common/src/main/java/com/github/kfcfans/common/request/WorkerHeartbeat.java +++ b/oh-my-scheduler-common/src/main/java/com/github/kfcfans/common/request/WorkerHeartbeat.java @@ -1,9 +1,9 @@ package com.github.kfcfans.common.request; +import com.github.kfcfans.common.OmsSerializable; import com.github.kfcfans.common.model.SystemMetrics; import lombok.Data; -import java.io.Serializable; /** * Worker 上报健康信息（worker定时发送的heartbeat） @@ -12,7 +12,7 @@ import java.io.Serializable; * @since 2020/3/25 */ @Data -public class WorkerHeartbeat implements Serializable { +public class WorkerHeartbeat implements OmsSerializable { // 本机地址 -> IP:port private String workerAddress; diff --git a/oh-my-scheduler-common/src/main/java/com/github/kfcfans/common/response/AskResponse.java b/oh-my-scheduler-common/src/main/java/com/github/kfcfans/common/response/AskResponse.java index 786d6361..a08667fc 100644 --- a/oh-my-scheduler-common/src/main/java/com/github/kfcfans/common/response/AskResponse.java +++ b/oh-my-scheduler-common/src/main/java/com/github/kfcfans/common/response/AskResponse.java @@ -1,10 +1,10 @@ package com.github.kfcfans.common.response; +import com.github.kfcfans.common.OmsSerializable; import lombok.AllArgsConstructor; import lombok.Data; import lombok.NoArgsConstructor; -import java.io.Serializable; /** * Pattens.ask 的响应 @@ -15,7 +15,7 @@ import java.io.Serializable; @Data @NoArgsConstructor @AllArgsConstructor -public class AskResponse implements Serializable { +public class AskResponse implements OmsSerializable { private boolean success; private Object extra; diff --git a/oh-my-scheduler-common/src/main/java/com/github/kfcfans/common/response/ResultDTO.java b/oh-my-scheduler-common/src/main/java/com/github/kfcfans/common/response/ResultDTO.java index e6b5c200..27ea0688 100644 --- a/oh-my-scheduler-common/src/main/java/com/github/kfcfans/common/response/ResultDTO.java +++ b/oh-my-scheduler-common/src/main/java/com/github/kfcfans/common/response/ResultDTO.java @@ -1,5 +1,6 @@ package com.github.kfcfans.common.response; +import com.github.kfcfans.common.OmsSerializable; import lombok.Getter; import lombok.Setter; import lombok.ToString; @@ -14,7 +15,7 @@ import org.apache.commons.lang3.exception.ExceptionUtils; @Getter @Setter @ToString -public class ResultDTO { +public class ResultDTO implements OmsSerializable { private boolean success; // 数据（success为 true 时存在） diff --git a/oh-my-scheduler-common/src/main/java/com/github/kfcfans/common/utils/JsonUtils.java b/oh-my-scheduler-common/src/main/java/com/github/kfcfans/common/utils/JsonUtils.java new file mode 100644 index 00000000..ff7ca3b2 --- /dev/null +++ b/oh-my-scheduler-common/src/main/java/com/github/kfcfans/common/utils/JsonUtils.java @@ -0,0 +1,27 @@ +package com.github.kfcfans.common.utils; + +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.ObjectMapper; + +/** + * JSON工具类 + * + * @author tjq + * @since 2020/4/16 + */ +public class JsonUtils { + + private static final ObjectMapper objectMapper = new ObjectMapper(); + + public static String toJSONString(Object obj) { + try { + return objectMapper.writeValueAsString(obj); + }catch (Exception ignore) { + } + return null; + } + + public static T parseObject(String json, Class clz) throws JsonProcessingException { + return objectMapper.readValue(json, clz); + } +} diff --git a/oh-my-scheduler-server/pom.xml b/oh-my-scheduler-server/pom.xml index ccf5e0ad..31d3354e 100644 --- a/oh-my-scheduler-server/pom.xml +++ b/oh-my-scheduler-server/pom.xml @@ -15,7 +15,6 @@ 2.9.2 - 2.6.4 2.2.6.RELEASE 1.0.0-SNAPSHOT 3.4.2 @@ -46,13 +45,6 @@ ${mysql.version} - - - com.typesafe.akka - akka-remote_2.13 - ${akka.version} - - org.springframework.boot @@ -84,7 +76,6 @@ ${swagger.version} - diff --git a/oh-my-scheduler-server/src/main/java/com/github/kfcfans/oms/server/akka/requests/FriendQueryWorkerClusterStatusReq.java b/oh-my-scheduler-server/src/main/java/com/github/kfcfans/oms/server/akka/requests/FriendQueryWorkerClusterStatusReq.java index 2ac25c5a..335a8042 100644 --- a/oh-my-scheduler-server/src/main/java/com/github/kfcfans/oms/server/akka/requests/FriendQueryWorkerClusterStatusReq.java +++ b/oh-my-scheduler-server/src/main/java/com/github/kfcfans/oms/server/akka/requests/FriendQueryWorkerClusterStatusReq.java @@ -1,10 +1,10 @@ package com.github.kfcfans.oms.server.akka.requests; +import com.github.kfcfans.common.OmsSerializable; import lombok.AllArgsConstructor; import lombok.Data; import lombok.NoArgsConstructor; -import java.io.Serializable; /** * 查询 Worker 集群状态 @@ -15,6 +15,6 @@ import java.io.Serializable; @Data @NoArgsConstructor @AllArgsConstructor -public class FriendQueryWorkerClusterStatusReq implements Serializable { +public class FriendQueryWorkerClusterStatusReq implements OmsSerializable { private Long appId; } diff --git a/oh-my-scheduler-server/src/main/java/com/github/kfcfans/oms/server/akka/requests/Ping.java b/oh-my-scheduler-server/src/main/java/com/github/kfcfans/oms/server/akka/requests/Ping.java index caad7d57..01048cf4 100644 --- a/oh-my-scheduler-server/src/main/java/com/github/kfcfans/oms/server/akka/requests/Ping.java +++ b/oh-my-scheduler-server/src/main/java/com/github/kfcfans/oms/server/akka/requests/Ping.java @@ -1,8 +1,8 @@ package com.github.kfcfans.oms.server.akka.requests; +import com.github.kfcfans.common.OmsSerializable; import lombok.Data; -import java.io.Serializable; /** * 检测目标机器是否存活 @@ -11,6 +11,6 @@ import java.io.Serializable; * @since 2020/4/5 */ @Data -public class Ping implements Serializable { +public class Ping implements OmsSerializable { private long currentTime; } diff --git a/oh-my-scheduler-server/src/main/java/com/github/kfcfans/oms/server/persistence/repository/InstanceLogRepository.java b/oh-my-scheduler-server/src/main/java/com/github/kfcfans/oms/server/persistence/repository/InstanceLogRepository.java index 1e08fb8c..ecee1bbc 100644 --- a/oh-my-scheduler-server/src/main/java/com/github/kfcfans/oms/server/persistence/repository/InstanceLogRepository.java +++ b/oh-my-scheduler-server/src/main/java/com/github/kfcfans/oms/server/persistence/repository/InstanceLogRepository.java @@ -46,9 +46,9 @@ public interface InstanceLogRepository extends JpaRepository findByJobIdInAndStatusAndExpectedTriggerTimeLessThan(List jobIds, int status, long time); - List findByJobIdInAndStatusAndActualTriggerTimeLessThan(List jobIds, int status, long time); - List findByJobIdInAndStatusAndGmtModifiedBefore(List jobIds, int status, Date time); + List findByAppIdInAndStatusAndExpectedTriggerTimeLessThan(List jobIds, int status, long time); + List findByAppIdInAndStatusAndActualTriggerTimeLessThan(List jobIds, int status, long time); + List findByAppIdInAndStatusAndGmtModifiedBefore(List jobIds, int status, Date time); InstanceLogDO findByInstanceId(long instanceId); diff --git a/oh-my-scheduler-server/src/main/java/com/github/kfcfans/oms/server/service/timing/InstanceStatusCheckService.java b/oh-my-scheduler-server/src/main/java/com/github/kfcfans/oms/server/service/timing/InstanceStatusCheckService.java index c93ddf8b..7c0c8b0b 100644 --- a/oh-my-scheduler-server/src/main/java/com/github/kfcfans/oms/server/service/timing/InstanceStatusCheckService.java +++ b/oh-my-scheduler-server/src/main/java/com/github/kfcfans/oms/server/service/timing/InstanceStatusCheckService.java @@ -73,7 +73,7 @@ public class InstanceStatusCheckService { // 1. 检查等待 WAITING_DISPATCH 状态的任务 long threshold = System.currentTimeMillis() - DISPATCH_TIMEOUT_MS; - List waitingDispatchInstances = instanceLogRepository.findByJobIdInAndStatusAndExpectedTriggerTimeLessThan(partAppIds, InstanceStatus.WAITING_DISPATCH.getV(), threshold); + List waitingDispatchInstances = instanceLogRepository.findByAppIdInAndStatusAndExpectedTriggerTimeLessThan(partAppIds, InstanceStatus.WAITING_DISPATCH.getV(), threshold); if (!CollectionUtils.isEmpty(waitingDispatchInstances)) { log.warn("[InstanceStatusCheckService] instances({}) is not triggered as expected.", waitingDispatchInstances); waitingDispatchInstances.forEach(instance -> { @@ -85,7 +85,7 @@ public class InstanceStatusCheckService { // 2. 检查 WAITING_WORKER_RECEIVE 状态的任务 threshold = System.currentTimeMillis() - RECEIVE_TIMEOUT_MS; - List waitingWorkerReceiveInstances = instanceLogRepository.findByJobIdInAndStatusAndActualTriggerTimeLessThan(partAppIds, InstanceStatus.WAITING_WORKER_RECEIVE.getV(), threshold); + List waitingWorkerReceiveInstances = instanceLogRepository.findByAppIdInAndStatusAndActualTriggerTimeLessThan(partAppIds, InstanceStatus.WAITING_WORKER_RECEIVE.getV(), threshold); if (!CollectionUtils.isEmpty(waitingWorkerReceiveInstances)) { log.warn("[InstanceStatusCheckService] instances({}) did n’t receive any reply from worker.", waitingWorkerReceiveInstances); waitingWorkerReceiveInstances.forEach(instance -> { @@ -97,7 +97,7 @@ public class InstanceStatusCheckService { // 3. 检查 RUNNING 状态的任务（一定时间没收到 TaskTracker 的状态报告，视为失败） threshold = System.currentTimeMillis() - RUNNING_TIMEOUT_MS; - List failedInstances = instanceLogRepository.findByJobIdInAndStatusAndGmtModifiedBefore(partAppIds, InstanceStatus.RUNNING.getV(), new Date(threshold)); + List failedInstances = instanceLogRepository.findByAppIdInAndStatusAndGmtModifiedBefore(partAppIds, InstanceStatus.RUNNING.getV(), new Date(threshold)); if (!CollectionUtils.isEmpty(failedInstances)) { log.warn("[InstanceStatusCheckService] instances({}) has not received status report for a long time.", failedInstances); failedInstances.forEach(instance -> { diff --git a/oh-my-scheduler-server/src/main/resources/application.properties b/oh-my-scheduler-server/src/main/resources/application.properties index b31a1a54..c9445500 100644 --- a/oh-my-scheduler-server/src/main/resources/application.properties +++ b/oh-my-scheduler-server/src/main/resources/application.properties @@ -3,7 +3,8 @@ server.port=7700 # db config spring.datasource.driver-class-name=com.mysql.cj.jdbc.Driver -spring.datasource.url=jdbc:mysql://remotehost:3391/oms?charset=utf8mb4&useSSL=false +# JDBC配置不支持utf8mb4，需要更改my.conf +spring.datasource.url=jdbc:mysql://remotehost:3391/oms?useUnicode=true&characterEncoding=UTF-8 spring.datasource.username=root spring.datasource.password=No1Bug2Please3! # Hikari 数据源专用配置 diff --git a/oh-my-scheduler-server/src/main/resources/oms-server.akka.conf b/oh-my-scheduler-server/src/main/resources/oms-server.akka.conf index c6657860..d0d2da66 100644 --- a/oh-my-scheduler-server/src/main/resources/oms-server.akka.conf +++ b/oh-my-scheduler-server/src/main/resources/oms-server.akka.conf @@ -3,7 +3,11 @@ akka { # cluster is better(recommend by official document), but I prefer remote provider = remote # TODO : 临时使用 Java 序列化，开发完成后切换到 protocol-buffers - allow-java-serialization = on + allow-java-serialization = off + + serialization-bindings { + "com.github.kfcfans.common.OmsSerializable" = jackson-cbor + } } remote { artery { diff --git a/oh-my-scheduler-worker/pom.xml b/oh-my-scheduler-worker/pom.xml index e5387236..0b612350 100644 --- a/oh-my-scheduler-worker/pom.xml +++ b/oh-my-scheduler-worker/pom.xml @@ -15,14 +15,11 @@ 5.2.4.RELEASE - 2.6.4 1.0.0-SNAPSHOT 1.4.200 3.4.2 5.6.1 5.0.0-RC5 - 1.2.68 - 4.4.1 2.6 @@ -36,13 +33,6 @@ - - - com.typesafe.akka - akka-remote_2.13 - ${akka.version} - - com.github.kfcfans @@ -78,20 +68,6 @@ test - - - com.squareup.okhttp3 - okhttp - ${okhttp.version} - - - - - com.alibaba - fastjson - ${fastjson.version} - - commons-io @@ -99,8 +75,6 @@ ${commons.io.version} - - ch.qos.logback diff --git a/oh-my-scheduler-worker/src/main/java/com/github/kfcfans/oms/worker/OhMyWorker.java b/oh-my-scheduler-worker/src/main/java/com/github/kfcfans/oms/worker/OhMyWorker.java index 0c3464dd..ccafb82f 100644 --- a/oh-my-scheduler-worker/src/main/java/com/github/kfcfans/oms/worker/OhMyWorker.java +++ b/oh-my-scheduler-worker/src/main/java/com/github/kfcfans/oms/worker/OhMyWorker.java @@ -2,9 +2,9 @@ package com.github.kfcfans.oms.worker; import akka.actor.ActorSystem; import akka.actor.Props; -import com.alibaba.fastjson.JSONObject; import com.github.kfcfans.common.response.ResultDTO; import com.github.kfcfans.common.utils.CommonUtils; +import com.github.kfcfans.common.utils.JsonUtils; import com.github.kfcfans.oms.worker.actors.ProcessorTrackerActor; import com.github.kfcfans.oms.worker.actors.TaskTrackerActor; import com.github.kfcfans.oms.worker.background.ServerDiscoveryService; @@ -131,7 +131,7 @@ public class OhMyWorker implements ApplicationContextAware, InitializingBean { String realUrl = String.format(url, server, appName); try { String resultDTOStr = CommonUtils.executeWithRetry0(() -> HttpUtils.get(realUrl)); - ResultDTO resultDTO = JSONObject.parseObject(resultDTOStr, ResultDTO.class); + ResultDTO resultDTO = JsonUtils.parseObject(resultDTOStr, ResultDTO.class); if (resultDTO.isSuccess()) { Long appId = Long.valueOf(resultDTO.getData().toString()); log.info("[OhMyWorker] assert appName({}) succeed, the appId for this application is {}.", appName, appId); diff --git a/oh-my-scheduler-worker/src/main/java/com/github/kfcfans/oms/worker/background/ServerDiscoveryService.java b/oh-my-scheduler-worker/src/main/java/com/github/kfcfans/oms/worker/background/ServerDiscoveryService.java index 5bbc1a8c..bf721768 100644 --- a/oh-my-scheduler-worker/src/main/java/com/github/kfcfans/oms/worker/background/ServerDiscoveryService.java +++ b/oh-my-scheduler-worker/src/main/java/com/github/kfcfans/oms/worker/background/ServerDiscoveryService.java @@ -1,8 +1,8 @@ package com.github.kfcfans.oms.worker.background; -import com.alibaba.fastjson.JSONObject; import com.github.kfcfans.common.response.ResultDTO; import com.github.kfcfans.common.utils.CommonUtils; +import com.github.kfcfans.common.utils.JsonUtils; import com.github.kfcfans.oms.worker.OhMyWorker; import com.github.kfcfans.common.utils.HttpUtils; import com.github.kfcfans.oms.worker.core.tracker.task.TaskTracker; @@ -95,9 +95,12 @@ public class ServerDiscoveryService { }catch (Exception ignore) { } if (!StringUtils.isEmpty(result)) { - ResultDTO resultDTO = JSONObject.parseObject(result, ResultDTO.class); - if (resultDTO.isSuccess()) { - return resultDTO.getData().toString(); + try { + ResultDTO resultDTO = JsonUtils.parseObject(result, ResultDTO.class); + if (resultDTO.isSuccess()) { + return resultDTO.getData().toString(); + } + }catch (Exception ignore) { } } return null; diff --git a/oh-my-scheduler-worker/src/main/java/com/github/kfcfans/oms/worker/persistence/TaskPersistenceService.java b/oh-my-scheduler-worker/src/main/java/com/github/kfcfans/oms/worker/persistence/TaskPersistenceService.java index c55130f8..09b56ad9 100644 --- a/oh-my-scheduler-worker/src/main/java/com/github/kfcfans/oms/worker/persistence/TaskPersistenceService.java +++ b/oh-my-scheduler-worker/src/main/java/com/github/kfcfans/oms/worker/persistence/TaskPersistenceService.java @@ -27,7 +27,7 @@ public class TaskPersistenceService { // 默认重试参数 private static final int RETRY_TIMES = 3; - private static final long RETRY_INTERVAL_MS = 100; + private static final long RETRY_INTERVAL_MS = 200; private static volatile boolean initialized = false; public static TaskPersistenceService INSTANCE = new TaskPersistenceService(); diff --git a/oh-my-scheduler-worker/src/main/java/com/github/kfcfans/oms/worker/pojo/request/BroadcastTaskPreExecuteFinishedReq.java b/oh-my-scheduler-worker/src/main/java/com/github/kfcfans/oms/worker/pojo/request/BroadcastTaskPreExecuteFinishedReq.java index 8c4d8b21..0be1ce92 100644 --- a/oh-my-scheduler-worker/src/main/java/com/github/kfcfans/oms/worker/pojo/request/BroadcastTaskPreExecuteFinishedReq.java +++ b/oh-my-scheduler-worker/src/main/java/com/github/kfcfans/oms/worker/pojo/request/BroadcastTaskPreExecuteFinishedReq.java @@ -1,8 +1,8 @@ package com.github.kfcfans.oms.worker.pojo.request; +import com.github.kfcfans.common.OmsSerializable; import lombok.Data; -import java.io.Serializable; /** * 广播任务 preExecute 结束信息 @@ -11,7 +11,7 @@ import java.io.Serializable; * @since 2020/3/23 */ @Data -public class BroadcastTaskPreExecuteFinishedReq implements Serializable { +public class BroadcastTaskPreExecuteFinishedReq implements OmsSerializable { private Long instanceId; private Long subInstanceId; diff --git a/oh-my-scheduler-worker/src/main/java/com/github/kfcfans/oms/worker/pojo/request/ProcessorMapTaskRequest.java b/oh-my-scheduler-worker/src/main/java/com/github/kfcfans/oms/worker/pojo/request/ProcessorMapTaskRequest.java index 317eda23..47556b15 100644 --- a/oh-my-scheduler-worker/src/main/java/com/github/kfcfans/oms/worker/pojo/request/ProcessorMapTaskRequest.java +++ b/oh-my-scheduler-worker/src/main/java/com/github/kfcfans/oms/worker/pojo/request/ProcessorMapTaskRequest.java @@ -1,5 +1,6 @@ package com.github.kfcfans.oms.worker.pojo.request; +import com.github.kfcfans.common.OmsSerializable; import com.github.kfcfans.oms.worker.common.ThreadLocalStore; import com.github.kfcfans.oms.worker.common.utils.SerializerUtils; import com.github.kfcfans.oms.worker.persistence.TaskDO; @@ -8,7 +9,6 @@ import lombok.AllArgsConstructor; import lombok.Getter; import lombok.NoArgsConstructor; -import java.io.Serializable; import java.util.List; /** @@ -19,7 +19,7 @@ import java.util.List; */ @Getter @NoArgsConstructor -public class ProcessorMapTaskRequest implements Serializable { +public class ProcessorMapTaskRequest implements OmsSerializable { private Long instanceId; private Long subInstanceId; diff --git a/oh-my-scheduler-worker/src/main/java/com/github/kfcfans/oms/worker/pojo/request/ProcessorReportTaskStatusReq.java b/oh-my-scheduler-worker/src/main/java/com/github/kfcfans/oms/worker/pojo/request/ProcessorReportTaskStatusReq.java index 5f000447..71bc3f2e 100644 --- a/oh-my-scheduler-worker/src/main/java/com/github/kfcfans/oms/worker/pojo/request/ProcessorReportTaskStatusReq.java +++ b/oh-my-scheduler-worker/src/main/java/com/github/kfcfans/oms/worker/pojo/request/ProcessorReportTaskStatusReq.java @@ -1,10 +1,10 @@ package com.github.kfcfans.oms.worker.pojo.request; +import com.github.kfcfans.common.OmsSerializable; import lombok.AllArgsConstructor; import lombok.Data; import lombok.NoArgsConstructor; -import java.io.Serializable; /** * worker 上报 task 执行情况 @@ -15,7 +15,7 @@ import java.io.Serializable; @Data @NoArgsConstructor @AllArgsConstructor -public class ProcessorReportTaskStatusReq implements Serializable { +public class ProcessorReportTaskStatusReq implements OmsSerializable { private Long instanceId; private String taskId; diff --git a/oh-my-scheduler-worker/src/main/java/com/github/kfcfans/oms/worker/pojo/request/ProcessorTrackerStatusReportReq.java b/oh-my-scheduler-worker/src/main/java/com/github/kfcfans/oms/worker/pojo/request/ProcessorTrackerStatusReportReq.java index f6041332..06b7f5fc 100644 --- a/oh-my-scheduler-worker/src/main/java/com/github/kfcfans/oms/worker/pojo/request/ProcessorTrackerStatusReportReq.java +++ b/oh-my-scheduler-worker/src/main/java/com/github/kfcfans/oms/worker/pojo/request/ProcessorTrackerStatusReportReq.java @@ -1,11 +1,10 @@ package com.github.kfcfans.oms.worker.pojo.request; -import com.github.kfcfans.common.utils.NetUtils; +import com.github.kfcfans.common.OmsSerializable; import com.github.kfcfans.oms.worker.OhMyWorker; import lombok.Data; import lombok.NoArgsConstructor; -import java.io.Serializable; /** * ProcessorTracker 定时向 TaskTracker 上报健康状态 @@ -15,7 +14,7 @@ import java.io.Serializable; */ @Data @NoArgsConstructor -public class ProcessorTrackerStatusReportReq implements Serializable { +public class ProcessorTrackerStatusReportReq implements OmsSerializable { private Long instanceId; diff --git a/oh-my-scheduler-worker/src/main/java/com/github/kfcfans/oms/worker/pojo/request/TaskTrackerStartTaskReq.java b/oh-my-scheduler-worker/src/main/java/com/github/kfcfans/oms/worker/pojo/request/TaskTrackerStartTaskReq.java index e9ba5f63..33330211 100644 --- a/oh-my-scheduler-worker/src/main/java/com/github/kfcfans/oms/worker/pojo/request/TaskTrackerStartTaskReq.java +++ b/oh-my-scheduler-worker/src/main/java/com/github/kfcfans/oms/worker/pojo/request/TaskTrackerStartTaskReq.java @@ -1,5 +1,6 @@ package com.github.kfcfans.oms.worker.pojo.request; +import com.github.kfcfans.common.OmsSerializable; import com.github.kfcfans.oms.worker.OhMyWorker; import com.github.kfcfans.oms.worker.persistence.TaskDO; import com.github.kfcfans.oms.worker.pojo.model.InstanceInfo; @@ -7,7 +8,6 @@ import lombok.Getter; import lombok.NoArgsConstructor; import lombok.Setter; -import java.io.Serializable; /** * TaskTracker 派发 task 进行执行 @@ -18,7 +18,7 @@ import java.io.Serializable; @Getter @Setter @NoArgsConstructor -public class TaskTrackerStartTaskReq implements Serializable { +public class TaskTrackerStartTaskReq implements OmsSerializable { // TaskTracker 地址 private String taskTrackerAddress; diff --git a/oh-my-scheduler-worker/src/main/java/com/github/kfcfans/oms/worker/pojo/request/TaskTrackerStopInstanceReq.java b/oh-my-scheduler-worker/src/main/java/com/github/kfcfans/oms/worker/pojo/request/TaskTrackerStopInstanceReq.java index f60e1c5c..71da8327 100644 --- a/oh-my-scheduler-worker/src/main/java/com/github/kfcfans/oms/worker/pojo/request/TaskTrackerStopInstanceReq.java +++ b/oh-my-scheduler-worker/src/main/java/com/github/kfcfans/oms/worker/pojo/request/TaskTrackerStopInstanceReq.java @@ -1,8 +1,8 @@ package com.github.kfcfans.oms.worker.pojo.request; +import com.github.kfcfans.common.OmsSerializable; import lombok.Data; -import java.io.Serializable; /** * TaskTracker 停止 ProcessorTracker，释放相关资源 @@ -12,7 +12,7 @@ import java.io.Serializable; * @since 2020/3/25 */ @Data -public class TaskTrackerStopInstanceReq implements Serializable { +public class TaskTrackerStopInstanceReq implements OmsSerializable { private Long instanceId; // 保留字段，暂时没用 diff --git a/oh-my-scheduler-worker/src/main/resources/oms-worker.akka.conf b/oh-my-scheduler-worker/src/main/resources/oms-worker.akka.conf index d77550a5..918b06f7 100644 --- a/oh-my-scheduler-worker/src/main/resources/oms-worker.akka.conf +++ b/oh-my-scheduler-worker/src/main/resources/oms-worker.akka.conf @@ -2,8 +2,11 @@ akka { actor { # cluster is better(recommend by official document), but I prefer remote provider = remote - # TODO : 临时使用 Java 序列化，开发完成后切换到 protocol-buffers - allow-java-serialization = on + allow-java-serialization = off + + serialization-bindings { + "com.github.kfcfans.common.OmsSerializable" = jackson-cbor + } } remote { artery { diff --git a/oh-my-scheduler-worker/src/test/java/com/github/kfcfans/oms/processors/TestBasicProcessor.java b/oh-my-scheduler-worker/src/test/java/com/github/kfcfans/oms/processors/TestBasicProcessor.java index b1996a04..c68a2b4b 100644 --- a/oh-my-scheduler-worker/src/test/java/com/github/kfcfans/oms/processors/TestBasicProcessor.java +++ b/oh-my-scheduler-worker/src/test/java/com/github/kfcfans/oms/processors/TestBasicProcessor.java @@ -1,6 +1,6 @@ package com.github.kfcfans.oms.processors; -import com.alibaba.fastjson.JSONObject; +import com.github.kfcfans.common.utils.JsonUtils; import com.github.kfcfans.oms.worker.core.processor.ProcessResult; import com.github.kfcfans.oms.worker.core.processor.TaskContext; import com.github.kfcfans.oms.worker.core.processor.sdk.BasicProcessor; @@ -16,7 +16,7 @@ public class TestBasicProcessor implements BasicProcessor { @Override public ProcessResult process(TaskContext context) throws Exception { System.out.println("======== BasicProcessor#process ========"); - System.out.println("TaskContext: " + JSONObject.toJSONString(context) + ";time = " + System.currentTimeMillis()); + System.out.println("TaskContext: " + JsonUtils.toJSONString(context) + ";time = " + System.currentTimeMillis()); return new ProcessResult(true, System.currentTimeMillis() + "success"); } diff --git a/oh-my-scheduler-worker/src/test/java/com/github/kfcfans/oms/processors/TestBroadcastProcessor.java b/oh-my-scheduler-worker/src/test/java/com/github/kfcfans/oms/processors/TestBroadcastProcessor.java index 35427c7e..e74597f6 100644 --- a/oh-my-scheduler-worker/src/test/java/com/github/kfcfans/oms/processors/TestBroadcastProcessor.java +++ b/oh-my-scheduler-worker/src/test/java/com/github/kfcfans/oms/processors/TestBroadcastProcessor.java @@ -1,6 +1,6 @@ package com.github.kfcfans.oms.processors; -import com.alibaba.fastjson.JSONObject; +import com.github.kfcfans.common.utils.JsonUtils; import com.github.kfcfans.oms.worker.core.processor.ProcessResult; import com.github.kfcfans.oms.worker.core.processor.TaskContext; import com.github.kfcfans.oms.worker.core.processor.sdk.BroadcastProcessor; @@ -17,14 +17,14 @@ public class TestBroadcastProcessor implements BroadcastProcessor { @Override public ProcessResult preProcess(TaskContext taskContext) throws Exception { System.out.println("=============== TestBroadcastProcessor#preProcess ==============="); - System.out.println("taskContext:" + JSONObject.toJSONString(taskContext)); + System.out.println("taskContext:" + JsonUtils.toJSONString(taskContext)); return new ProcessResult(true, "preProcess success"); } @Override public ProcessResult postProcess(TaskContext taskContext, Map taskId2Result) throws Exception { System.out.println("=============== TestBroadcastProcessor#postProcess ==============="); - System.out.println("taskContext:" + JSONObject.toJSONString(taskContext)); + System.out.println("taskContext:" + JsonUtils.toJSONString(taskContext)); System.out.println("taskId2Result:" + taskId2Result); return new ProcessResult(true, "postProcess success"); } @@ -32,7 +32,7 @@ public class TestBroadcastProcessor implements BroadcastProcessor { @Override public ProcessResult process(TaskContext context) throws Exception { System.out.println("=============== TestBroadcastProcessor#process ==============="); - System.out.println("taskContext:" + JSONObject.toJSONString(context)); + System.out.println("taskContext:" + JsonUtils.toJSONString(context)); return new ProcessResult(true, "processSuccess"); } } diff --git a/oh-my-scheduler-worker/src/test/java/com/github/kfcfans/oms/processors/TestMapReduceProcessor.java b/oh-my-scheduler-worker/src/test/java/com/github/kfcfans/oms/processors/TestMapReduceProcessor.java index 2649e130..48e66450 100644 --- a/oh-my-scheduler-worker/src/test/java/com/github/kfcfans/oms/processors/TestMapReduceProcessor.java +++ b/oh-my-scheduler-worker/src/test/java/com/github/kfcfans/oms/processors/TestMapReduceProcessor.java @@ -1,6 +1,6 @@ package com.github.kfcfans.oms.processors; -import com.alibaba.fastjson.JSONObject; +import com.github.kfcfans.common.utils.JsonUtils; import com.github.kfcfans.oms.worker.core.processor.ProcessResult; import com.github.kfcfans.oms.worker.core.processor.TaskContext; import com.github.kfcfans.oms.worker.core.processor.sdk.MapReduceProcessor; @@ -24,7 +24,7 @@ public class TestMapReduceProcessor extends MapReduceProcessor { @Override public ProcessResult reduce(TaskContext taskContext, Map taskId2Result) { System.out.println("============== TestMapReduceProcessor#reduce =============="); - System.out.println("taskContext:" + JSONObject.toJSONString(taskContext)); + System.out.println("taskContext:" + JsonUtils.toJSONString(taskContext)); System.out.println("taskId2Result:" + taskId2Result); return new ProcessResult(true, "REDUCE_SUCCESS"); } @@ -33,7 +33,7 @@ public class TestMapReduceProcessor extends MapReduceProcessor { public ProcessResult process(TaskContext context) throws Exception { System.out.println("============== TestMapReduceProcessor#process =============="); System.out.println("isRootTask:" + isRootTask()); - System.out.println("taskContext:" + JSONObject.toJSONString(context)); + System.out.println("taskContext:" + JsonUtils.toJSONString(context)); if (isRootTask()) { System.out.println("==== MAP ===="); @@ -50,7 +50,7 @@ public class TestMapReduceProcessor extends MapReduceProcessor { return new ProcessResult(true, "MAP_SUCCESS"); }else { System.out.println("==== NORMAL_PROCESS ===="); - System.out.println("subTask: " + JSONObject.toJSONString(context.getSubTask())); + System.out.println("subTask: " + JsonUtils.toJSONString(context.getSubTask())); Thread.sleep(1000); if (context.getCurrentRetryTimes() == 0) { return new ProcessResult(false, "FIRST_FAILED"); diff --git a/oh-my-scheduler-worker/~/.DS_Store b/oh-my-scheduler-worker/~/.DS_Store new file mode 100644 index 00000000..5a217eb2 Binary files /dev/null and b/oh-my-scheduler-worker/~/.DS_Store differ diff --git a/oh-my-scheduler-worker/~/.oms/script/python/2.py b/oh-my-scheduler-worker/~/.oms/script/python/2.py new file mode 100755 index 00000000..29e965f8 --- /dev/null +++ b/oh-my-scheduler-worker/~/.oms/script/python/2.py @@ -0,0 +1 @@ +print 'Hello World!' \ No newline at end of file diff --git a/oh-my-scheduler-worker/~/.oms/script/shell/1.sh b/oh-my-scheduler-worker/~/.oms/script/shell/1.sh new file mode 100755 index 00000000..c6d4c4d9 --- /dev/null +++ b/oh-my-scheduler-worker/~/.oms/script/shell/1.sh @@ -0,0 +1 @@ +ls -a \ No newline at end of file diff --git a/oh-my-scheduler-worker/~/.oms/script/shell/18.sh b/oh-my-scheduler-worker/~/.oms/script/shell/18.sh new file mode 100755 index 00000000..30ad8db1 --- /dev/null +++ b/oh-my-scheduler-worker/~/.oms/script/shell/18.sh @@ -0,0 +1,17 @@ + + + + + + + + oms-console + + + +
+ + + diff --git a/oh-my-scheduler-worker/~/.oms/script/shell/277.sh b/oh-my-scheduler-worker/~/.oms/script/shell/277.sh new file mode 100755 index 00000000..c6d4c4d9 --- /dev/null +++ b/oh-my-scheduler-worker/~/.oms/script/shell/277.sh @@ -0,0 +1 @@ +ls -a \ No newline at end of file diff --git a/oh-my-scheduler-worker/~/.oms/script/shell/2777.sh b/oh-my-scheduler-worker/~/.oms/script/shell/2777.sh new file mode 100755 index 00000000..013c184c --- /dev/null +++ b/oh-my-scheduler-worker/~/.oms/script/shell/2777.sh @@ -0,0 +1 @@ +pwd \ No newline at end of file diff --git a/oh-my-scheduler-worker/~/oms/.DS_Store b/oh-my-scheduler-worker/~/oms/.DS_Store new file mode 100644 index 00000000..43f8d602 Binary files /dev/null and b/oh-my-scheduler-worker/~/oms/.DS_Store differ diff --git a/oh-my-scheduler-worker/~/oms/script/.DS_Store b/oh-my-scheduler-worker/~/oms/script/.DS_Store new file mode 100644 index 00000000..908022ca Binary files /dev/null and b/oh-my-scheduler-worker/~/oms/script/.DS_Store differ diff --git a/oh-my-scheduler-worker/~/oms/script/shell/.DS_Store b/oh-my-scheduler-worker/~/oms/script/shell/.DS_Store new file mode 100644 index 00000000..fe8f87b4 Binary files /dev/null and b/oh-my-scheduler-worker/~/oms/script/shell/.DS_Store differ diff --git a/oh-my-scheduler-worker/~/oms/script/shell/18.sh b/oh-my-scheduler-worker/~/oms/script/shell/18.sh new file mode 100644 index 00000000..e69de29b diff --git a/others/logs/TestRecord.md b/others/logs/TestRecord.md index 3b2d800e..9d5e3050 100644 --- a/others/logs/TestRecord.md +++ b/others/logs/TestRecord.md @@ -48,4 +48,46 @@ java.lang.RuntimeException: create root task failed. #### SystemMetric算分问题 问题：java.lang.management.OperatingSystemMXBean#getSystemLoadAverage 不一定能获取CPU当前负载，可能返回负数代表不可用... -解决方案：印度Windows上getSystemLoadAverage()固定返回-1...太坑了...先做个保护性判断继续测试吧... \ No newline at end of file +解决方案：印度Windows上getSystemLoadAverage()固定返回-1...太坑了...先做个保护性判断继续测试吧... + +#### 未知的数组越界问题（可能是数据库性能问题） +秒级Broadcast任务在第四次执行时，当Processor完成执行上报状态时，TaskTracker报错，错误的本质原因是无法从数据库中找到这个task对应的记录... +时间表达式：FIX_DELAY，对应的TaskTracker为FrequentTaskTracker + +异常堆栈 +```text +2020-04-16 18:05:09 ERROR - [TaskPersistenceService] getTaskStatus failed, instanceId=1586857062542,taskId=4. +java.lang.IndexOutOfBoundsException: Index: 0, Size: 0 + at java.util.LinkedList.checkElementIndex(LinkedList.java:555) + at java.util.LinkedList.get(LinkedList.java:476) + at com.github.kfcfans.oms.worker.persistence.TaskPersistenceService.lambda$getTaskStatus$10(TaskPersistenceService.java:214) + at com.github.kfcfans.common.utils.CommonUtils.executeWithRetry(CommonUtils.java:37) + at com.github.kfcfans.oms.worker.persistence.TaskPersistenceService.execute(TaskPersistenceService.java:310) + at com.github.kfcfans.oms.worker.persistence.TaskPersistenceService.getTaskStatus(TaskPersistenceService.java:212) + at com.github.kfcfans.oms.worker.core.tracker.task.TaskTracker.updateTaskStatus(TaskTracker.java:107) + at com.github.kfcfans.oms.worker.core.tracker.task.TaskTracker.broadcast(TaskTracker.java:214) + at com.github.kfcfans.oms.worker.actors.TaskTrackerActor.onReceiveBroadcastTaskPreExecuteFinishedReq(TaskTrackerActor.java:106) + at akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:24) + at akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:20) + at scala.PartialFunction.applyOrElse(PartialFunction.scala:187) + at scala.PartialFunction.applyOrElse$(PartialFunction.scala:186) + at akka.japi.pf.UnitCaseStatement.applyOrElse(CaseStatements.scala:20) + at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:241) + at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:242) + at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:242) + at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:242) + at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:242) + at akka.actor.Actor.aroundReceive(Actor.scala:534) + at akka.actor.Actor.aroundReceive$(Actor.scala:532) + at akka.actor.AbstractActor.aroundReceive(AbstractActor.scala:220) + at akka.actor.ActorCell.receiveMessage(ActorCell.scala:573) + at akka.actor.ActorCell.invoke(ActorCell.scala:543) + at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:269) + at akka.dispatch.Mailbox.run(Mailbox.scala:230) + at akka.dispatch.Mailbox.exec(Mailbox.scala:242) + at java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:289) + at java.util.concurrent.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1056) + at java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1692) + at java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:157) +2020-04-16 18:05:09 WARN - [TaskTracker-1586857062542] query TaskStatus from DB failed when try to update new TaskStatus(taskId=4,newStatus=6). +``` \ No newline at end of file