third_codes/PowerJob
1
0
Fork 1
mirror of https://github.com/PowerJob/PowerJob.git synced 2025-07-17 00:00:04 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix FrequentTaskTracker's StandaloneJob always success bug

Browse Source
This commit is contained in:
tjq 2020-04-17 20:06:00 +08:00
parent 593ee714b8
commit 072d94f283
7 changed files with 45 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
oh-my-scheduler-server/src/main/java/com/github/kfcfans/oms/server/akka/actors/ServerActor.java
View File

@ -1,6 +1,7 @@
package com.github.kfcfans.oms.server.akka.actors;
import akka.actor.AbstractActor;
import com.github.kfcfans.common.InstanceStatus;
import com.github.kfcfans.common.request.TaskTrackerReportInstanceStatusReq;
import com.github.kfcfans.common.request.WorkerHeartbeat;
import com.github.kfcfans.common.response.AskResponse;
@ -45,8 +46,10 @@ public class ServerActor extends AbstractActor {
try {
InstanceManager.updateStatus(req);
// 回复接收成功
getSender().tell(AskResponse.succeed(null), getSelf());
// 结束状态（成功/失败）需要回复消息
if (!InstanceStatus.generalizedRunningStatus.contains(req.getInstanceStatus())) {
getSender().tell(AskResponse.succeed(null), getSelf());
}
}catch (Exception e) {
log.error("[ServerActor] update instance status failed for request: {}.", req, e);
}

2
oh-my-scheduler-worker-samples/src/main/java/com/github/kfcfans/oms/server/processors/BroadcastProcessorDemo.java
View File

@ -6,6 +6,7 @@ import com.github.kfcfans.oms.worker.core.processor.TaskContext;
import com.github.kfcfans.oms.worker.core.processor.TaskResult;
import com.github.kfcfans.oms.worker.core.processor.sdk.BroadcastProcessor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import java.util.List;
import java.util.concurrent.ThreadLocalRandom;
@ -18,6 +19,7 @@ import java.util.concurrent.ThreadLocalRandom;
* @since 2020/4/17
*/
@Slf4j
@Component
public class BroadcastProcessorDemo extends BroadcastProcessor {
@Override

2
oh-my-scheduler-worker-samples/src/main/java/com/github/kfcfans/oms/server/processors/MapReduceProcessorDemo.java
View File

@ -12,6 +12,7 @@ import lombok.Getter;
import lombok.NoArgsConstructor;
import lombok.ToString;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import java.util.List;
import java.util.concurrent.ThreadLocalRandom;
@ -24,6 +25,7 @@ import java.util.concurrent.ThreadLocalRandom;
* @since 2020/4/17
*/
@Slf4j
@Component
public class MapReduceProcessorDemo extends MapReduceProcessor {
// 每一批发送任务大小

6
oh-my-scheduler-worker-samples/src/main/java/com/github/kfcfans/oms/server/processors/StandaloneProcessorDemo.java
View File

@ -5,6 +5,7 @@ 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;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import java.util.concurrent.ThreadLocalRandom;
@ -16,15 +17,16 @@ import java.util.concurrent.ThreadLocalRandom;
* @since 2020/4/17
*/
@Slf4j
@Component
public class StandaloneProcessorDemo implements BasicProcessor {
@Override
public ProcessResult process(TaskContext context) throws Exception {
System.out.println("================ StandaloneProcessorDemo#process ================");
System.out.println("TaskContext: " + JSONObject.toJSONString(context));
boolean success = ThreadLocalRandom.current().nextBoolean();
System.out.println("TaskContext: " + JSONObject.toJSONString(context));
System.out.println("ProcessSuccess: " + success);
return new ProcessResult(success, context + ": " + success);
}
}

27
oh-my-scheduler-worker-samples/src/main/resources/logback.xml Normal file
View File

@ -0,0 +1,27 @@
<?xml version="1.0" encoding="UTF-8"?>
<configuration>
<!-- %m输出的信息,%p日志级别,%t线程名,%d日期,%c类的全名,%i索引【从数字0开始递增】,,, -->
<!-- appender是configuration的子节点，是负责写日志的组件。 -->
<!-- ConsoleAppender：把日志输出到控制台 -->
<appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
<encoder>
<!-- <pattern>%red(%d{yyyy-MM-dd HH:mm:ss}) %highlight(%-5level) %green([%thread]) - %cyan(%msg%n)</pattern>-->
<pattern>%red(%d{yyyy-MM-dd HH:mm:ss}) %highlight(%-5level) - %cyan(%msg%n)</pattern>
<!-- 控制台也要使用UTF-8，不要使用GBK，否则会中文乱码 -->
<charset>UTF-8</charset>
</encoder>
</appender>
<logger name="com.zaxxer.hikari" level="INFO">
<appender-ref ref="STDOUT"/>
</logger>
<logger name="com.github.kfcfans.oms" level="DEBUG" additivity="false">
<appender-ref ref="STDOUT"/>
</logger>
<!-- 控制台输出日志级别 -->
<root level="INFO">
<appender-ref ref="STDOUT"/>
</root>
</configuration>

8
oh-my-scheduler-worker/src/main/java/com/github/kfcfans/oms/worker/core/tracker/task/FrequentTaskTracker.java
View File

@ -257,17 +257,17 @@ public class FrequentTaskTracker extends TaskTracker {
continue;
}
String result;
switch (executeType) {
// STANDALONE 代表任务确实已经执行完毕了
case STANDALONE:
// 查询数据库获取结果（STANDALONE每个SubInstance只会有一条Task记录）
result = taskPersistenceService.getAllTask(instanceId, subInstanceId).get(0).getResult();
onFinished(subInstanceId, true, result, iterator);
TaskDO resultTask = taskPersistenceService.getAllTask(instanceId, subInstanceId).get(0);
boolean success = resultTask.getStatus() == TaskStatus.WORKER_PROCESS_SUCCESS.getValue();
onFinished(subInstanceId, success, resultTask.getResult(), iterator);
continue;
// MAP 不关心结果，最简单
case MAP:
result = String.format("total:%d,succeed:%d,failed:%d", holder.getTotalTaskNum(), holder.succeedNum, holder.failedNum);
String result = String.format("total:%d,succeed:%d,failed:%d", holder.getTotalTaskNum(), holder.succeedNum, holder.failedNum);
onFinished(subInstanceId, holder.failedNum == 0, result, iterator);
continue;
// MapReduce 和 BroadCast 需要根据是否有 LAST_TASK 来判断结束与否

2
oh-my-scheduler-worker/src/main/java/com/github/kfcfans/oms/worker/core/tracker/task/TaskTracker.java
View File

@ -146,7 +146,7 @@ public abstract class TaskTracker {
// 处理失败的情况
int configTaskRetryNum = instanceInfo.getTaskRetryNum();
if (nTaskStatus == TaskStatus.WORKER_PROCESS_FAILED && configTaskRetryNum > 1) {
if (nTaskStatus == TaskStatus.WORKER_PROCESS_FAILED && configTaskRetryNum >= 1) {
// 失败不是主要的情况，多查一次数据库也问题不大（况且前面有缓存顶着，大部分情况之前不会去查DB）
Optional<TaskDO> taskOpt = taskPersistenceService.getTask(instanceId, taskId);