com.github.kfcfans.powerjob.worker.core.tracker.task

类 TaskTracker

java.lang.Object

com.github.kfcfans.powerjob.worker.core.tracker.task.TaskTracker

直接已知子类:

CommonTaskTracker, FrequentTaskTracker

public abstract class TaskTracker
extends Object

负责管理 JobInstance 的运行，主要包括任务的派发（MR可能存在大量的任务）和状态的更新

从以下版本开始:

2020/4/8

作者:

tjq

嵌套类概要

嵌套类

限定符和类型	类和说明
protected class	TaskTracker.Dispatcher 定时扫描数据库中的task（出于内存占用量考虑，每次最多获取100个），并将需要执行的任务派发出去
protected static class	TaskTracker.InstanceStatisticsHolder 存储任务实例产生的各个Task状态，用于分析任务实例执行情况

字段概要

字段

限定符和类型	字段和说明
protected long	createTime
protected AtomicBoolean	finished
protected long	instanceId
protected InstanceInfo	instanceInfo
protected ProcessorTrackerStatusHolder	ptStatusHolder
protected ScheduledExecutorService	scheduledPool
protected TaskPersistenceService	taskPersistenceService

构造器概要

构造器

限定符	构造器和说明
protected	TaskTracker(ServerScheduleJobReq req)

方法概要

限定符和类型	方法和说明
void	broadcast(boolean preExecuteSuccess, long subInstanceId, String preTaskId, String result) 生成广播任务
static TaskTracker	create(ServerScheduleJobReq req) 静态方法创建 TaskTracker
void	destroy() 销毁自身，释放资源
protected void	dispatchTask(TaskDO task, String processorTrackerAddress) 派发任务到 ProcessorTracker
abstract InstanceDetail	fetchRunningStatus() 查询任务实例的详细运行状态
protected TaskTracker.InstanceStatisticsHolder	getInstanceStatisticsHolder(long subInstanceId) 获取任务实例产生的各个Task状态，用于分析任务实例执行情况
protected abstract void	initTaskTracker(ServerScheduleJobReq req) 初始化 TaskTracker
void	receiveProcessorTrackerHeartbeat(ProcessorTrackerStatusReportReq heartbeatReq) 处理 ProcessorTracker 的心跳信息
boolean	submitTask(List<TaskDO> newTaskList) 提交Task任务(MapReduce的Map，Broadcast的广播)，上层保证 batchSize，同时插入过多数据可能导致失败
void	updateTaskStatus(String taskId, int newStatus, long reportTime, String result) 更新Task状态 V1.0.0 -> V1.0.1（e405e283ad7f97b0b4e5d369c7de884c0caf9192） 锁方案变更，从 synchronized (taskId.intern()) 修改为分段锁，能大大减少内存占用，损失的只有理论并发度而已

从类继承的方法 java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

字段详细资料

createTime

protected long createTime

instanceId

protected long instanceId

instanceInfo

protected InstanceInfo instanceInfo

ptStatusHolder

protected ProcessorTrackerStatusHolder ptStatusHolder

taskPersistenceService

protected TaskPersistenceService taskPersistenceService

scheduledPool

protected ScheduledExecutorService scheduledPool

finished

protected AtomicBoolean finished

构造器详细资料

TaskTracker

protected TaskTracker(ServerScheduleJobReq req)

方法详细资料

create

public static TaskTracker create(ServerScheduleJobReq req)

静态方法创建 TaskTracker

参数:

req - 服务端调度任务请求

返回:

API/CRON -> CommonTaskTracker, FIX_RATE/FIX_DELAY -> FrequentTaskTracker

updateTaskStatus

public void updateTaskStatus(String taskId,
                             int newStatus,
                             long reportTime,
                             @Nullable
                             String result)

更新Task状态 V1.0.0 -> V1.0.1（e405e283ad7f97b0b4e5d369c7de884c0caf9192） 锁方案变更，从 synchronized (taskId.intern()) 修改为分段锁，能大大减少内存占用，损失的只有理论并发度而已

参数:

taskId - task的ID（task为任务实例的执行单位）

newStatus - task的新状态

reportTime - 上报时间

result - task的执行结果，未执行完成时为空

submitTask

public boolean submitTask(List<TaskDO> newTaskList)

提交Task任务(MapReduce的Map，Broadcast的广播)，上层保证 batchSize，同时插入过多数据可能导致失败

参数:

newTaskList - 新增的子任务列表

receiveProcessorTrackerHeartbeat

public void receiveProcessorTrackerHeartbeat(ProcessorTrackerStatusReportReq heartbeatReq)

处理 ProcessorTracker 的心跳信息

参数:

heartbeatReq - ProcessorTracker（任务的执行管理器）发来的心跳包，包含了其当前状态

broadcast

public void broadcast(boolean preExecuteSuccess,
                      long subInstanceId,
                      String preTaskId,
                      String result)

生成广播任务

参数:

preExecuteSuccess - 预执行广播任务运行状态

subInstanceId - 子实例ID

preTaskId - 预执行广播任务的taskId

result - 预执行广播任务的结果

destroy

public void destroy()

销毁自身，释放资源

dispatchTask

protected void dispatchTask(TaskDO task,
                            String processorTrackerAddress)

派发任务到 ProcessorTracker

参数:

task - 需要被执行的任务

processorTrackerAddress - ProcessorTracker的地址（IP:Port）

getInstanceStatisticsHolder

protected TaskTracker.InstanceStatisticsHolder getInstanceStatisticsHolder(long subInstanceId)

获取任务实例产生的各个Task状态，用于分析任务实例执行情况

参数:

subInstanceId - 子任务实例ID

返回:

InstanceStatisticsHolder

initTaskTracker

protected abstract void initTaskTracker(ServerScheduleJobReq req)

初始化 TaskTracker

参数:

req - 服务器调度任务实例运行请求

fetchRunningStatus

public abstract InstanceDetail fetchRunningStatus()

查询任务实例的详细运行状态

返回:

任务实例的详细运行状态