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

[dev] idle ProcessorTracker auto destroy

Browse Source
This commit is contained in:
tjq 2020-06-21 18:57:45 +08:00
parent 10824c5c36
commit 51d7ba4743
7 changed files with 34 additions and 27 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
powerjob-server/src/main/resources/static/js/4.js
View File

File diff suppressed because one or more lines are too long

4
powerjob-server/src/main/resources/static/js/6.js
View File

File diff suppressed because one or more lines are too long

8
powerjob-server/src/main/resources/static/js/app.js
View File

File diff suppressed because one or more lines are too long

1
powerjob-worker/src/main/java/com/github/kfcfans/powerjob/worker/actors/TaskTrackerActor.java
View File

@ -127,6 +127,7 @@ public class TaskTrackerActor extends AbstractActor {
* ProcessorTracker 心跳处理器
*/
private void onReceiveProcessorTrackerStatusReportReq(ProcessorTrackerStatusReportReq req) {
TaskTracker taskTracker = TaskTrackerPool.getTaskTrackerPool(req.getInstanceId());
if (taskTracker == null) {
log.warn("[TaskTrackerActor] receive ProcessorTrackerStatusReportReq({}) but system can't find TaskTracker.", req);

2
powerjob-worker/src/main/java/com/github/kfcfans/powerjob/worker/background/ServerDiscoveryService.java
View File

@ -48,8 +48,10 @@ public class ServerDiscoveryService {
String ip = currentServer.split(":")[0];
// 直接请求当前Server的HTTP服务，可以少一次网络开销，减轻Server负担
String firstServerAddress = IP2ADDRESS.get(ip);
if (firstServerAddress != null) {
result = acquire(firstServerAddress);
}
}
for (String httpServerAddress : OhMyWorker.getConfig().getServerAddress()) {
if (StringUtils.isEmpty(result)) {

31
powerjob-worker/src/main/java/com/github/kfcfans/powerjob/worker/core/tracker/processor/ProcessorTracker.java
View File

@ -64,7 +64,7 @@ public class ProcessorTracker {
private ScheduledExecutorService timingPool;
private static final int THREAD_POOL_QUEUE_MAX_SIZE = 100;
// 最多，长时间空闲的 ProcessorTracker 会发起销毁请求
// 长时间空闲的 ProcessorTracker 会发起销毁请求
private static final long MAX_IDLE_TIME = 120000;
// 当 ProcessorTracker 出现根本性错误（比如 Processor 创建失败，所有的任务直接失败）
@ -237,18 +237,6 @@ public class ProcessorTracker {
}
}
// 上报状态之前，先重新发送失败的任务，只要有结果堆积，就不上报状态（让 PT 认为该 TT 失联然后重试相关任务）
while (!statusReportRetryQueue.isEmpty()) {
ProcessorReportTaskStatusReq req = statusReportRetryQueue.poll();
if (req != null) {
req.setReportTime(System.currentTimeMillis());
if (!AkkaUtils.reliableTransmit(taskTrackerActorRef, req)) {
statusReportRetryQueue.add(req);
return;
}
}
}
// 判断线程池活跃状态，长时间空闲则上报 TaskTracker 请求检查
if (threadPool.getActiveCount() > 0) {
lastIdleTime = -1;
@ -258,10 +246,23 @@ public class ProcessorTracker {
}else {
long idleTime = System.currentTimeMillis() - lastIdleTime;
if (idleTime > MAX_IDLE_TIME) {
lastIdleTime = System.currentTimeMillis();
log.warn("[ProcessorTracker-{}] ProcessorTracker have been idle for {}ms, it's time to tell TaskTracker.", instanceId, idleTime);
log.warn("[ProcessorTracker-{}] ProcessorTracker have been idle for {}ms, it's time to tell TaskTracker and then destroy self.", instanceId, idleTime);
// 不可靠通知，如果该请求失败，则整个任务处理集群缺失一个 ProcessorTracker，影响可接受
taskTrackerActorRef.tell(ProcessorTrackerStatusReportReq.buildIdleReport(instanceId), null);
destroy();
return;
}
}
}
// 上报状态之前，先重新发送失败的任务，只要有结果堆积，就不上报状态（让 PT 认为该 TT 失联然后重试相关任务）
while (!statusReportRetryQueue.isEmpty()) {
ProcessorReportTaskStatusReq req = statusReportRetryQueue.poll();
if (req != null) {
req.setReportTime(System.currentTimeMillis());
if (!AkkaUtils.reliableTransmit(taskTrackerActorRef, req)) {
statusReportRetryQueue.add(req);
return;
}
}

7
powerjob-worker/src/main/java/com/github/kfcfans/powerjob/worker/core/tracker/task/TaskTracker.java
View File

@ -272,9 +272,12 @@ public abstract class TaskTracker {
// 上报空闲，检查是否已经接收到全部该 ProcessorTracker 负责的任务
if (heartbeatReq.getType() == ProcessorTrackerStatusReportReq.IDLE) {
List<TaskDO> unfinishedTask = TaskPersistenceService.INSTANCE.getAllUnFinishedTaskByAddress(instanceId, heartbeatReq.getAddress());
String idlePtAddress = heartbeatReq.getAddress();
// 该 ProcessorTracker 已销毁，重置为初始状态
ptStatusHolder.getProcessorTrackerStatus(idlePtAddress).setDispatched(false);
List<TaskDO> unfinishedTask = TaskPersistenceService.INSTANCE.getAllUnFinishedTaskByAddress(instanceId, idlePtAddress);
if (!CollectionUtils.isEmpty(unfinishedTask)) {
log.warn("[TaskTracker-{}] ProcessorTracker is idle now but have unfinished tasks: {}", instanceId, unfinishedTask);
log.warn("[TaskTracker-{}] ProcessorTracker({}) is idle now but have unfinished tasks: {}", instanceId, idlePtAddress, unfinishedTask);
unfinishedTask.forEach(task -> updateTaskStatus(task.getTaskId(), TaskStatus.WORKER_PROCESS_FAILED.getValue(), System.currentTimeMillis(), "SYSTEM: unreceived process result"));
}
}