聊聊PowerJob的LightTaskTracker

序

本文主要研究一下PowerJob的LightTaskTracker

TaskTracker

tech/powerjob/worker/core/tracker/task/TaskTracker.java

 
@Slf4j
public abstract class TaskTracker {
 
    /**
     * TaskTracker创建时间
     */
    protected final long createTime;
    /**
     * 任务实例ID，使用频率过高，从 InstanceInfo 提取出来单独保存一份
     */
    protected final long instanceId;
    /**
     * 任务实例信息
     */
    protected final InstanceInfo instanceInfo;
    /**
     * 追加的工作流上下文数据
     *
     * @since 2021/02/05
     */
    protected final Map<String, String> appendedWfContext;
    /**
     * worker 运行时元数据
     */
    protected final WorkerRuntime workerRuntime;
    /**
     * 是否结束
     */
    protected final AtomicBoolean finished;
    /**
     * 连续上报多次失败后放弃上报，视为结果不可达，TaskTracker down
     */
    protected int reportFailedCnt = 0;
 
    protected static final int MAX_REPORT_FAILED_THRESHOLD = 5;
 
    protected TaskTracker(ServerScheduleJobReq req, WorkerRuntime workerRuntime) {
        this.createTime = System.currentTimeMillis();
        this.workerRuntime = workerRuntime;
        this.instanceId = req.getInstanceId();
 
        this.instanceInfo = new InstanceInfo();
 
        // PowerJob 值拷贝场景不多，引入三方值拷贝类库可能引入类冲突等问题，综合评估手写 ROI 最高
        instanceInfo.setJobId(req.getJobId());
        instanceInfo.setInstanceId(req.getInstanceId());
        instanceInfo.setWfInstanceId(req.getWfInstanceId());
        instanceInfo.setExecuteType(req.getExecuteType());
        instanceInfo.setProcessorType(req.getProcessorType());
        instanceInfo.setProcessorInfo(req.getProcessorInfo());
        instanceInfo.setJobParams(req.getJobParams());
        instanceInfo.setInstanceParams(req.getInstanceParams());
        instanceInfo.setThreadConcurrency(req.getThreadConcurrency());
        instanceInfo.setTaskRetryNum(req.getTaskRetryNum());
        instanceInfo.setLogConfig(req.getLogConfig());
 
        // 特殊处理超时时间
        if (instanceInfo.getInstanceTimeoutMS() <= 0) {
            instanceInfo.setInstanceTimeoutMS(Integer.MAX_VALUE);
        }
        // 只有工作流中的任务允许向工作流中追加上下文数据
        this.appendedWfContext = req.getWfInstanceId() == null ? Collections.emptyMap() : Maps.newConcurrentMap();
        this.finished = new AtomicBoolean(false);
    }
 
    /**
     * 销毁
     */
    public abstract void destroy();
 
    /**
     * 停止任务
     */
    public abstract void stopTask();
 
 
    /**
     * 查询任务实例的详细运行状态
     *
     * @return 任务实例的详细运行状态
     */
    public abstract InstanceDetail fetchRunningStatus();
 
    //......
}

TaskTracker是个抽象类，其构造器接收ServerScheduleJobReq、WorkerRuntime，然后根据ServerScheduleJobReq构建InstanceInfo；它定义了destroy、stopTask、fetchRunningStatus抽象方法

LightTaskTracker

tech/powerjob/worker/core/tracker/task/light/LightTaskTracker.java

 
@Slf4j
public class LightTaskTracker extends TaskTracker {
    /**
     * statusReportScheduledFuture
     */
    private final ScheduledFuture<?> statusReportScheduledFuture;
    /**
     * timeoutCheckScheduledFuture
     */
    private final ScheduledFuture<?> timeoutCheckScheduledFuture;
    /**
     * processFuture
     */
    private final Future<ProcessResult> processFuture;
    /**
     * 执行线程
     */
    private final AtomicReference<Thread> executeThread;
    /**
     * 处理器信息
     */
    private final ProcessorBean processorBean;
    /**
     * 上下文
     */
    private final TaskContext taskContext;
    /**
     * 任务状态
     */
    private TaskStatus status;
    /**
     * 任务开始执行的时间
     */
    private Long taskStartTime;
    /**
     * 任务执行结束的时间 或者 任务被 kill 掉的时间
     */
    private Long taskEndTime;
    /**
     * 任务处理结果
     */
    private ProcessResult result;
 
    private final AtomicBoolean timeoutFlag = new AtomicBoolean(false);
 
    protected final AtomicBoolean stopFlag = new AtomicBoolean(false);
 
    protected final AtomicBoolean destroyFlag = new AtomicBoolean(false);
 
 
    public LightTaskTracker(ServerScheduleJobReq req, WorkerRuntime workerRuntime) {
        super(req, workerRuntime);
        try {
            taskContext = constructTaskContext(req, workerRuntime);
            // 等待处理
            status = TaskStatus.WORKER_RECEIVED;
            // 加载 Processor
            processorBean = workerRuntime.getProcessorLoader().load(new ProcessorDefinition().setProcessorType(req.getProcessorType()).setProcessorInfo(req.getProcessorInfo()));
            executeThread = new AtomicReference<>();
            long delay = Integer.parseInt(System.getProperty(PowerJobDKey.WORKER_STATUS_CHECK_PERIOD, "15")) * 1000L;
            // 初始延迟加入随机值，避免在高并发场景下所有请求集中在一个时间段
            long initDelay = RandomUtils.nextInt(5000, 10000);
            // 上报任务状态
            statusReportScheduledFuture = workerRuntime.getExecutorManager().getLightweightTaskStatusCheckExecutor().scheduleWithFixedDelay(this::checkAndReportStatus, initDelay, delay, TimeUnit.MILLISECONDS);
            // 超时控制
            if (instanceInfo.getInstanceTimeoutMS() != Integer.MAX_VALUE) {
                if (instanceInfo.getInstanceTimeoutMS() < 1000L) {
                    timeoutCheckScheduledFuture = workerRuntime.getExecutorManager().getLightweightTaskStatusCheckExecutor().scheduleAtFixedRate(this::timeoutCheck, instanceInfo.getInstanceTimeoutMS(), instanceInfo.getInstanceTimeoutMS() / 10, TimeUnit.MILLISECONDS);
                } else {
                    // 执行时间超过 1 s 的任务，超时检测最小颗粒度为 1 s
                    timeoutCheckScheduledFuture = workerRuntime.getExecutorManager().getLightweightTaskStatusCheckExecutor().scheduleAtFixedRate(this::timeoutCheck, instanceInfo.getInstanceTimeoutMS(), 1000L, TimeUnit.MILLISECONDS);
                }
            } else {
                timeoutCheckScheduledFuture = null;
            }
            // 提交任务到线程池
            processFuture = workerRuntime.getExecutorManager().getLightweightTaskExecutorService().submit(this::processTask);
        } catch (Exception e) {
            log.error("[TaskTracker-{}] fail to create TaskTracker for req:{} ", instanceId, req);
            destroy();
            throw e;
        }
 
    }
 
    //......
}

LightTaskTracker继承了TaskTracker，其构造器根据ServerScheduleJobReq创建ProcessorDefinition，然后使用workerRuntime.getProcessorLoader().load方法进行加载，之后通过workerRuntime.getExecutorManager().getLightweightTaskStatusCheckExecutor().scheduleWithFixedDelay调度checkAndReportStatus；若设置了instanceTimeout则调度timeoutCheck；最后通过workerRuntime.getExecutorManager().getLightweightTaskExecutorService().submit来执行processTask

checkAndReportStatus

 
    private synchronized void checkAndReportStatus() {
        if (destroyFlag.get()) {
            // 已经被销毁，不需要上报状态
            log.info("[TaskTracker-{}] has been destroyed,final status is {},needn't to report status!", instanceId, status);
            return;
        }
        TaskTrackerReportInstanceStatusReq reportInstanceStatusReq = new TaskTrackerReportInstanceStatusReq();
        reportInstanceStatusReq.setAppId(workerRuntime.getAppId());
        reportInstanceStatusReq.setJobId(instanceInfo.getJobId());
        reportInstanceStatusReq.setInstanceId(instanceId);
        reportInstanceStatusReq.setWfInstanceId(instanceInfo.getWfInstanceId());
        reportInstanceStatusReq.setTotalTaskNum(1);
        reportInstanceStatusReq.setReportTime(System.currentTimeMillis());
        reportInstanceStatusReq.setStartTime(createTime);
        reportInstanceStatusReq.setSourceAddress(workerRuntime.getWorkerAddress());
        reportInstanceStatusReq.setSucceedTaskNum(0);
        reportInstanceStatusReq.setFailedTaskNum(0);
 
        if (stopFlag.get()) {
            if (finished.get()) {
                // 已经被成功打断
                destroy();
                return;
            }
            final Thread workerThread = executeThread.get();
            if (!finished.get() && workerThread != null) {
                // 未能成功打断任务，强制停止
                try {
                    if (tryForceStopThread(workerThread)) {
                        finished.set(true);
                        taskEndTime = System.currentTimeMillis();
                        result = new ProcessResult(false, SystemInstanceResult.USER_STOP_INSTANCE_FORCE_STOP);
                        log.warn("[TaskTracker-{}] task need stop, force stop thread {} success!", instanceId, workerThread.getName());
                        // 被终止的任务不需要上报状态
                        destroy();
                        return;
                    }
                } catch (Exception e) {
                    log.warn("[TaskTracker-{}] task need stop,fail to stop thread {}", instanceId, workerThread.getName(), e);
                }
            }
        }
        if (finished.get()) {
            if (result.isSuccess()) {
                reportInstanceStatusReq.setSucceedTaskNum(1);
                reportInstanceStatusReq.setInstanceStatus(InstanceStatus.SUCCEED.getV());
            } else {
                reportInstanceStatusReq.setFailedTaskNum(1);
                reportInstanceStatusReq.setInstanceStatus(InstanceStatus.FAILED.getV());
            }
            // 处理工作流上下文
            if (taskContext.getWorkflowContext().getWfInstanceId() != null) {
                reportInstanceStatusReq.setAppendedWfContext(taskContext.getWorkflowContext().getAppendedContextData());
            }
            reportInstanceStatusReq.setResult(suit(result.getMsg()));
            reportInstanceStatusReq.setEndTime(taskEndTime);
            // 微操一下，上报最终状态时重新设置下时间，并且增加一小段偏移，保证在并发上报运行中状态以及最终状态时，最终状态的上报时间晚于运行中的状态
            reportInstanceStatusReq.setReportTime(System.currentTimeMillis() + 1);
            reportFinalStatusThenDestroy(workerRuntime, reportInstanceStatusReq);
            return;
        }
        // 未完成的任务，只需要上报状态
        reportInstanceStatusReq.setInstanceStatus(InstanceStatus.RUNNING.getV());
        log.info("[TaskTracker-{}] report status({}) success,real status is {}", instanceId, reportInstanceStatusReq, status);
        TransportUtils.ttReportInstanceStatus(reportInstanceStatusReq, workerRuntime.getServerDiscoveryService().getCurrentServerAddress(), workerRuntime.getTransporter());
    }

checkAndReportStatus方法构建TaskTrackerReportInstanceStatusReq，然后根据stopFlag和finished进行对应处理，针对未完成的任务执行TransportUtils.ttReportInstanceStatus进行上报

timeoutCheck

 
    private void timeoutCheck() {
        if (taskStartTime == null || System.currentTimeMillis() - taskStartTime < instanceInfo.getInstanceTimeoutMS()) {
            return;
        }
        if (finished.get() && result != null) {
            timeoutCheckScheduledFuture.cancel(true);
            return;
        }
        // 首次判断超时
        if (timeoutFlag.compareAndSet(false, true)) {
            // 超时，仅尝试打断任务
            log.warn("[TaskTracker-{}] task timeout,taskStarTime:{},currentTime:{},runningTimeLimit:{}, try to interrupt it.", instanceId, taskStartTime, System.currentTimeMillis(), instanceInfo.getInstanceTimeoutMS());
            processFuture.cancel(true);
            return;
        }
        if (finished.get()) {
            // 已经成功被打断
            log.warn("[TaskTracker-{}] task timeout,taskStarTime:{},endTime:{}, interrupt success.", instanceId, taskStartTime, taskEndTime);
            return;
        }
        Thread workerThread = executeThread.get();
        if (workerThread == null) {
            return;
        }
        // 未能成功打断任务，强制终止
        try {
            if (tryForceStopThread(workerThread)) {
                finished.set(true);
                taskEndTime = System.currentTimeMillis();
                result = new ProcessResult(false, SystemInstanceResult.INSTANCE_EXECUTE_TIMEOUT_FORCE_STOP);
                log.warn("[TaskTracker-{}] task timeout, force stop thread {} success!", instanceId, workerThread.getName());
            }
        } catch (Exception e) {
            log.warn("[TaskTracker-{}] task timeout,fail to stop thread {}", instanceId, workerThread.getName(), e);
        }
    }
 
    private boolean tryForceStopThread(Thread thread) {
 
        String threadName = thread.getName();
 
        String allowStopThread = System.getProperty(PowerJobDKey.WORKER_ALLOWED_FORCE_STOP_THREAD);
        if (!StringUtils.equalsIgnoreCase(allowStopThread, Boolean.TRUE.toString())) {
            log.warn("[TaskTracker-{}] PowerJob not allowed to force stop a thread by config", instanceId);
            return false;
        }
 
        log.warn("[TaskTracker-{}] fail to interrupt the thread[{}], try to force stop.", instanceId, threadName);
        try {
            thread.stop();
            return true;
        } catch (Throwable t) {
            log.warn("[TaskTracker-{}] stop thread[{}] failed, msg: {}", instanceId, threadName, t.getMessage());
        }
        return false;
    }

timeoutCheck先判断是否超时，接着判断是否finished，是则取消当前任务，接着更新timeoutFlag，然后通过processFuture.cancel(true)尝试打断任务；若任务未能成功打断则通过tryForceStopThread强制终止，这里用了thread.stop这个废弃方法

processTask

 
    private ProcessResult processTask() {
        executeThread.set(Thread.currentThread());
        // 设置任务开始执行的时间
        taskStartTime = System.currentTimeMillis();
        status = TaskStatus.WORKER_PROCESSING;
        // 开始执行时，提交任务判断是否超时
        ProcessResult res = null;
        do {
            Thread.currentThread().setContextClassLoader(processorBean.getClassLoader());
            if (res != null && !res.isSuccess()) {
                // 重试
                taskContext.setCurrentRetryTimes(taskContext.getCurrentRetryTimes() + 1);
                log.warn("[TaskTracker-{}] process failed, TaskTracker will have a retry,current retryTimes : {}", instanceId, taskContext.getCurrentRetryTimes());
            }
            try {
                res = processorBean.getProcessor().process(taskContext);
            } catch (InterruptedException e) {
                log.warn("[TaskTracker-{}] task has been interrupted !", instanceId, e);
                Thread.currentThread().interrupt();
                if (timeoutFlag.get()) {
                    res = new ProcessResult(false, SystemInstanceResult.INSTANCE_EXECUTE_TIMEOUT_INTERRUPTED);
                } else if (stopFlag.get()) {
                    res = new ProcessResult(false, SystemInstanceResult.USER_STOP_INSTANCE_INTERRUPTED);
                } else {
                    res = new ProcessResult(false, e.toString());
                }
            } catch (Exception e) {
                log.warn("[TaskTracker-{}] process failed !", instanceId, e);
                res = new ProcessResult(false, e.toString());
            }
            if (res == null) {
                log.warn("[TaskTracker-{}] processor return null !", instanceId);
                res = new ProcessResult(false, "Processor return null");
            }
        } while (!res.isSuccess() && taskContext.getCurrentRetryTimes() < taskContext.getMaxRetryTimes() && !timeoutFlag.get() && !stopFlag.get());
        executeThread.set(null);
        taskEndTime = System.currentTimeMillis();
        finished.set(true);
        result = res;
        status = result.isSuccess() ? TaskStatus.WORKER_PROCESS_SUCCESS : TaskStatus.WORKER_PROCESS_FAILED;
        // 取消超时检查任务
        if (timeoutCheckScheduledFuture != null) {
            timeoutCheckScheduledFuture.cancel(true);
        }
        log.info("[TaskTracker-{}] task complete ! create time:{},queue time:{},use time:{},result:{}", instanceId, createTime, taskStartTime - createTime, System.currentTimeMillis() - taskStartTime, result);
        // 执行完成后立即上报一次
        checkAndReportStatus();
        return result;
    }

processTask通过一个while循环来执行，该循环的条件是处理结果非成功，重试次数小于最大重试次数，任务未超时，stopFlag为false；循环内部执行的是processorBean.getProcessor().process(taskContext)，它会捕获InterruptedException及Exception；循环外则更新任务结束时间，取消timeoutCheckScheduledFuture，最后执行checkAndReportStatus进行上报

LightTaskTracker.create

 
    /**
     * 静态方法创建 TaskTracker
     *
     * @param req 服务端调度任务请求
     * @return LightTaskTracker
     */
    public static LightTaskTracker create(ServerScheduleJobReq req, WorkerRuntime workerRuntime) {
        try {
            return new LightTaskTracker(req, workerRuntime);
        } catch (Exception e) {
            reportCreateErrorToServer(req, workerRuntime, e);
        }
        return null;
    }

LightTaskTracker提供了静态方法create用于创建LightTaskTracker

TaskTrackerActor

tech/powerjob/worker/actors/TaskTrackerActor.java

 
@Slf4j
@Actor(path = WTT_PATH)
public class TaskTrackerActor {
 
    private final WorkerRuntime workerRuntime;
 
    public TaskTrackerActor(WorkerRuntime workerRuntime) {
        this.workerRuntime = workerRuntime;
    }
 
    /**
     * 服务器任务调度处理器
     */
    @Handler(path = WTT_HANDLER_RUN_JOB)
    public void onReceiveServerScheduleJobReq(ServerScheduleJobReq req) {
        log.debug("[TaskTrackerActor] server schedule job by request: {}.", req);
        Long instanceId = req.getInstanceId();
        // 区分轻量级任务模型以及重量级任务模型
        if (isLightweightTask(req)) {
            final LightTaskTracker taskTracker = LightTaskTrackerManager.getTaskTracker(instanceId);
            if (taskTracker != null) {
                log.warn("[TaskTrackerActor] LightTaskTracker({}) for instance(id={}) already exists.", taskTracker, instanceId);
                return;
            }
            // 判断是否已经 overload
            if (LightTaskTrackerManager.currentTaskTrackerSize() >= workerRuntime.getWorkerConfig().getMaxLightweightTaskNum() * LightTaskTrackerManager.OVERLOAD_FACTOR) {
                // ignore this request
                log.warn("[TaskTrackerActor] this worker is overload,ignore this request(instanceId={}),current size = {}!",instanceId,LightTaskTrackerManager.currentTaskTrackerSize());
                return;
            }
            if (LightTaskTrackerManager.currentTaskTrackerSize() >= workerRuntime.getWorkerConfig().getMaxLightweightTaskNum()) {
                log.warn("[TaskTrackerActor] this worker will be overload soon,current size = {}!",LightTaskTrackerManager.currentTaskTrackerSize());
            }
            // 创建轻量级任务
            LightTaskTrackerManager.atomicCreateTaskTracker(instanceId, ignore -> LightTaskTracker.create(req, workerRuntime));
        } else {
            HeavyTaskTracker taskTracker = HeavyTaskTrackerManager.getTaskTracker(instanceId);
            if (taskTracker != null) {
                log.warn("[TaskTrackerActor] HeavyTaskTracker({}) for instance(id={}) already exists.", taskTracker, instanceId);
                return;
            }
            // 判断是否已经 overload
            if (HeavyTaskTrackerManager.currentTaskTrackerSize() >= workerRuntime.getWorkerConfig().getMaxHeavyweightTaskNum()) {
                // ignore this request
                log.warn("[TaskTrackerActor] this worker is overload,ignore this request(instanceId={})! current size = {},", instanceId, HeavyTaskTrackerManager.currentTaskTrackerSize());
                return;
            }
            // 原子创建，防止多实例的存在
            HeavyTaskTrackerManager.atomicCreateTaskTracker(instanceId, ignore -> HeavyTaskTracker.create(req, workerRuntime));
        }
    }
 
    //......
}

TaskTrackerActor的path为taskTracker，它用于处理server的jobInstance请求和worker的task请求；其onReceiveServerScheduleJobReq方法的path为runJob，它接收ServerScheduleJobReq，然后通过isLightweightTask判断任务模型，是轻量级任务的话，则通过LightTaskTrackerManager.getTaskTracker(instanceId)获取taskTracker，接着判断当前实例的LightTaskTracker数量是否过多，过多则直接返回；最后通过LightTaskTrackerManager.atomicCreateTaskTracker来维护instanceId与LightTaskTracker的关系，若不存在则通过LightTaskTracker.create(req, workerRuntime)创建LightTaskTracker

isLightweightTask

 
    private boolean isLightweightTask(ServerScheduleJobReq serverScheduleJobReq) {
        final ExecuteType executeType = ExecuteType.valueOf(serverScheduleJobReq.getExecuteType());
        // 非单机执行的一定不是
        if (executeType != ExecuteType.STANDALONE){
            return false;
        }
        TimeExpressionType timeExpressionType = TimeExpressionType.valueOf(serverScheduleJobReq.getTimeExpressionType());
        // 固定频率以及固定延迟的也一定不是
        return timeExpressionType != TimeExpressionType.FIXED_DELAY && timeExpressionType != TimeExpressionType.FIXED_RATE;
    }

isLightweightTask的判断逻辑是如果executeType不是单机类型则不是轻量级任务，接着判断serverScheduleJobReq的timeExpressionType，类型不是FIXED_DELAY也不是FIXED_RATE的才是轻量级任务

小结

LightTaskTracker继承了TaskTracker，其构造器根据ServerScheduleJobReq创建ProcessorDefinition，然后使用workerRuntime.getProcessorLoader().load方法进行加载，之后调度checkAndReportStatus、timeoutCheck；最后执行processTask(没有把任务处理放到start方法，这些都在构造器里执行了)；TaskTrackerActor的path为taskTracker，它用于处理server的jobInstance请求和worker的task请求；其onReceiveServerScheduleJobReq方法的path为runJob，它接收ServerScheduleJobReq，用于根据请求的instanceId来创建和执行LightTaskTracker(通过ConcurrentHashMap来维护instanceId与LightTaskTracker的关系，避免重复执行)。

	@Slf4j
	public abstract class TaskTracker {

	/**
	* TaskTracker创建时间
	*/
	protected final long createTime;
	/**
	* 任务实例ID，使用频率过高，从 InstanceInfo 提取出来单独保存一份
	*/
	protected final long instanceId;
	/**
	* 任务实例信息
	*/
	protected final InstanceInfo instanceInfo;
	/**
	* 追加的工作流上下文数据
	*
	* @since 2021/02/05
	*/
	protected final Map<String, String> appendedWfContext;
	/**
	* worker 运行时元数据
	*/
	protected final WorkerRuntime workerRuntime;
	/**
	* 是否结束
	*/
	protected final AtomicBoolean finished;
	/**
	* 连续上报多次失败后放弃上报，视为结果不可达，TaskTracker down
	*/
	protected int reportFailedCnt = 0;

	protected static final int MAX_REPORT_FAILED_THRESHOLD = 5;

	protected TaskTracker(ServerScheduleJobReq req, WorkerRuntime workerRuntime) {
	this.createTime = System.currentTimeMillis();
	this.workerRuntime = workerRuntime;
	this.instanceId = req.getInstanceId();

	this.instanceInfo = new InstanceInfo();

	// PowerJob 值拷贝场景不多，引入三方值拷贝类库可能引入类冲突等问题，综合评估手写 ROI 最高
	instanceInfo.setJobId(req.getJobId());
	instanceInfo.setInstanceId(req.getInstanceId());
	instanceInfo.setWfInstanceId(req.getWfInstanceId());
	instanceInfo.setExecuteType(req.getExecuteType());
	instanceInfo.setProcessorType(req.getProcessorType());
	instanceInfo.setProcessorInfo(req.getProcessorInfo());
	instanceInfo.setJobParams(req.getJobParams());
	instanceInfo.setInstanceParams(req.getInstanceParams());
	instanceInfo.setThreadConcurrency(req.getThreadConcurrency());
	instanceInfo.setTaskRetryNum(req.getTaskRetryNum());
	instanceInfo.setLogConfig(req.getLogConfig());

	// 特殊处理超时时间
	if (instanceInfo.getInstanceTimeoutMS() <= 0) {
	instanceInfo.setInstanceTimeoutMS(Integer.MAX_VALUE);
	}
	// 只有工作流中的任务允许向工作流中追加上下文数据
	this.appendedWfContext = req.getWfInstanceId() == null ? Collections.emptyMap() : Maps.newConcurrentMap();
	this.finished = new AtomicBoolean(false);
	}

	/**
	* 销毁
	*/
	public abstract void destroy();

	/**
	* 停止任务
	*/
	public abstract void stopTask();


	/**
	* 查询任务实例的详细运行状态
	*
	* @return 任务实例的详细运行状态
	*/
	public abstract InstanceDetail fetchRunningStatus();

	//......
	}

	@Slf4j
	public class LightTaskTracker extends TaskTracker {
	/**
	* statusReportScheduledFuture
	*/
	private final ScheduledFuture<?> statusReportScheduledFuture;
	/**
	* timeoutCheckScheduledFuture
	*/
	private final ScheduledFuture<?> timeoutCheckScheduledFuture;
	/**
	* processFuture
	*/
	private final Future<ProcessResult> processFuture;
	/**
	* 执行线程
	*/
	private final AtomicReference<Thread> executeThread;
	/**
	* 处理器信息
	*/
	private final ProcessorBean processorBean;
	/**
	* 上下文
	*/
	private final TaskContext taskContext;
	/**
	* 任务状态
	*/
	private TaskStatus status;
	/**
	* 任务开始执行的时间
	*/
	private Long taskStartTime;
	/**
	* 任务执行结束的时间或者任务被 kill 掉的时间
	*/
	private Long taskEndTime;
	/**
	* 任务处理结果
	*/
	private ProcessResult result;

	private final AtomicBoolean timeoutFlag = new AtomicBoolean(false);

	protected final AtomicBoolean stopFlag = new AtomicBoolean(false);

	protected final AtomicBoolean destroyFlag = new AtomicBoolean(false);


	public LightTaskTracker(ServerScheduleJobReq req, WorkerRuntime workerRuntime) {
	super(req, workerRuntime);
	try {
	taskContext = constructTaskContext(req, workerRuntime);
	// 等待处理
	status = TaskStatus.WORKER_RECEIVED;
	// 加载 Processor
	processorBean = workerRuntime.getProcessorLoader().load(new ProcessorDefinition().setProcessorType(req.getProcessorType()).setProcessorInfo(req.getProcessorInfo()));
	executeThread = new AtomicReference<>();
	long delay = Integer.parseInt(System.getProperty(PowerJobDKey.WORKER_STATUS_CHECK_PERIOD, "15")) * 1000L;
	// 初始延迟加入随机值，避免在高并发场景下所有请求集中在一个时间段
	long initDelay = RandomUtils.nextInt(5000, 10000);
	// 上报任务状态
	statusReportScheduledFuture = workerRuntime.getExecutorManager().getLightweightTaskStatusCheckExecutor().scheduleWithFixedDelay(this::checkAndReportStatus, initDelay, delay, TimeUnit.MILLISECONDS);
	// 超时控制
	if (instanceInfo.getInstanceTimeoutMS() != Integer.MAX_VALUE) {
	if (instanceInfo.getInstanceTimeoutMS() < 1000L) {
	timeoutCheckScheduledFuture = workerRuntime.getExecutorManager().getLightweightTaskStatusCheckExecutor().scheduleAtFixedRate(this::timeoutCheck, instanceInfo.getInstanceTimeoutMS(), instanceInfo.getInstanceTimeoutMS() / 10, TimeUnit.MILLISECONDS);
	} else {
	// 执行时间超过 1 s 的任务，超时检测最小颗粒度为 1 s
	timeoutCheckScheduledFuture = workerRuntime.getExecutorManager().getLightweightTaskStatusCheckExecutor().scheduleAtFixedRate(this::timeoutCheck, instanceInfo.getInstanceTimeoutMS(), 1000L, TimeUnit.MILLISECONDS);
	}
	} else {
	timeoutCheckScheduledFuture = null;
	}
	// 提交任务到线程池
	processFuture = workerRuntime.getExecutorManager().getLightweightTaskExecutorService().submit(this::processTask);
	} catch (Exception e) {
	log.error("[TaskTracker-{}] fail to create TaskTracker for req:{} ", instanceId, req);
	destroy();
	throw e;
	}

	}

	//......
	}

	private synchronized void checkAndReportStatus() {
	if (destroyFlag.get()) {
	// 已经被销毁，不需要上报状态
	log.info("[TaskTracker-{}] has been destroyed,final status is {},needn't to report status!", instanceId, status);
	return;
	}
	TaskTrackerReportInstanceStatusReq reportInstanceStatusReq = new TaskTrackerReportInstanceStatusReq();
	reportInstanceStatusReq.setAppId(workerRuntime.getAppId());
	reportInstanceStatusReq.setJobId(instanceInfo.getJobId());
	reportInstanceStatusReq.setInstanceId(instanceId);
	reportInstanceStatusReq.setWfInstanceId(instanceInfo.getWfInstanceId());
	reportInstanceStatusReq.setTotalTaskNum(1);
	reportInstanceStatusReq.setReportTime(System.currentTimeMillis());
	reportInstanceStatusReq.setStartTime(createTime);
	reportInstanceStatusReq.setSourceAddress(workerRuntime.getWorkerAddress());
	reportInstanceStatusReq.setSucceedTaskNum(0);
	reportInstanceStatusReq.setFailedTaskNum(0);

	if (stopFlag.get()) {
	if (finished.get()) {
	// 已经被成功打断
	destroy();
	return;
	}
	final Thread workerThread = executeThread.get();
	if (!finished.get() && workerThread != null) {
	// 未能成功打断任务，强制停止
	try {
	if (tryForceStopThread(workerThread)) {
	finished.set(true);
	taskEndTime = System.currentTimeMillis();
	result = new ProcessResult(false, SystemInstanceResult.USER_STOP_INSTANCE_FORCE_STOP);
	log.warn("[TaskTracker-{}] task need stop, force stop thread {} success!", instanceId, workerThread.getName());
	// 被终止的任务不需要上报状态
	destroy();
	return;
	}
	} catch (Exception e) {
	log.warn("[TaskTracker-{}] task need stop,fail to stop thread {}", instanceId, workerThread.getName(), e);
	}
	}
	}
	if (finished.get()) {
	if (result.isSuccess()) {
	reportInstanceStatusReq.setSucceedTaskNum(1);
	reportInstanceStatusReq.setInstanceStatus(InstanceStatus.SUCCEED.getV());
	} else {
	reportInstanceStatusReq.setFailedTaskNum(1);
	reportInstanceStatusReq.setInstanceStatus(InstanceStatus.FAILED.getV());
	}
	// 处理工作流上下文
	if (taskContext.getWorkflowContext().getWfInstanceId() != null) {
	reportInstanceStatusReq.setAppendedWfContext(taskContext.getWorkflowContext().getAppendedContextData());
	}
	reportInstanceStatusReq.setResult(suit(result.getMsg()));
	reportInstanceStatusReq.setEndTime(taskEndTime);
	// 微操一下，上报最终状态时重新设置下时间，并且增加一小段偏移，保证在并发上报运行中状态以及最终状态时，最终状态的上报时间晚于运行中的状态
	reportInstanceStatusReq.setReportTime(System.currentTimeMillis() + 1);
	reportFinalStatusThenDestroy(workerRuntime, reportInstanceStatusReq);
	return;
	}
	// 未完成的任务，只需要上报状态
	reportInstanceStatusReq.setInstanceStatus(InstanceStatus.RUNNING.getV());
	log.info("[TaskTracker-{}] report status({}) success,real status is {}", instanceId, reportInstanceStatusReq, status);
	TransportUtils.ttReportInstanceStatus(reportInstanceStatusReq, workerRuntime.getServerDiscoveryService().getCurrentServerAddress(), workerRuntime.getTransporter());
	}

	private void timeoutCheck() {
	if (taskStartTime == null \|\| System.currentTimeMillis() - taskStartTime < instanceInfo.getInstanceTimeoutMS()) {
	return;
	}
	if (finished.get() && result != null) {
	timeoutCheckScheduledFuture.cancel(true);
	return;
	}
	// 首次判断超时
	if (timeoutFlag.compareAndSet(false, true)) {
	// 超时，仅尝试打断任务
	log.warn("[TaskTracker-{}] task timeout,taskStarTime:{},currentTime:{},runningTimeLimit:{}, try to interrupt it.", instanceId, taskStartTime, System.currentTimeMillis(), instanceInfo.getInstanceTimeoutMS());
	processFuture.cancel(true);
	return;
	}
	if (finished.get()) {
	// 已经成功被打断
	log.warn("[TaskTracker-{}] task timeout,taskStarTime:{},endTime:{}, interrupt success.", instanceId, taskStartTime, taskEndTime);
	return;
	}
	Thread workerThread = executeThread.get();
	if (workerThread == null) {
	return;
	}
	// 未能成功打断任务，强制终止
	try {
	if (tryForceStopThread(workerThread)) {
	finished.set(true);
	taskEndTime = System.currentTimeMillis();
	result = new ProcessResult(false, SystemInstanceResult.INSTANCE_EXECUTE_TIMEOUT_FORCE_STOP);
	log.warn("[TaskTracker-{}] task timeout, force stop thread {} success!", instanceId, workerThread.getName());
	}
	} catch (Exception e) {
	log.warn("[TaskTracker-{}] task timeout,fail to stop thread {}", instanceId, workerThread.getName(), e);
	}
	}

	private boolean tryForceStopThread(Thread thread) {

	String threadName = thread.getName();

	String allowStopThread = System.getProperty(PowerJobDKey.WORKER_ALLOWED_FORCE_STOP_THREAD);
	if (!StringUtils.equalsIgnoreCase(allowStopThread, Boolean.TRUE.toString())) {
	log.warn("[TaskTracker-{}] PowerJob not allowed to force stop a thread by config", instanceId);
	return false;
	}

	log.warn("[TaskTracker-{}] fail to interrupt the thread[{}], try to force stop.", instanceId, threadName);
	try {
	thread.stop();
	return true;
	} catch (Throwable t) {
	log.warn("[TaskTracker-{}] stop thread[{}] failed, msg: {}", instanceId, threadName, t.getMessage());
	}
	return false;
	}

	private ProcessResult processTask() {
	executeThread.set(Thread.currentThread());
	// 设置任务开始执行的时间
	taskStartTime = System.currentTimeMillis();
	status = TaskStatus.WORKER_PROCESSING;
	// 开始执行时，提交任务判断是否超时
	ProcessResult res = null;
	do {
	Thread.currentThread().setContextClassLoader(processorBean.getClassLoader());
	if (res != null && !res.isSuccess()) {
	// 重试
	taskContext.setCurrentRetryTimes(taskContext.getCurrentRetryTimes() + 1);
	log.warn("[TaskTracker-{}] process failed, TaskTracker will have a retry,current retryTimes : {}", instanceId, taskContext.getCurrentRetryTimes());
	}
	try {
	res = processorBean.getProcessor().process(taskContext);
	} catch (InterruptedException e) {
	log.warn("[TaskTracker-{}] task has been interrupted !", instanceId, e);
	Thread.currentThread().interrupt();
	if (timeoutFlag.get()) {
	res = new ProcessResult(false, SystemInstanceResult.INSTANCE_EXECUTE_TIMEOUT_INTERRUPTED);
	} else if (stopFlag.get()) {
	res = new ProcessResult(false, SystemInstanceResult.USER_STOP_INSTANCE_INTERRUPTED);
	} else {
	res = new ProcessResult(false, e.toString());
	}
	} catch (Exception e) {
	log.warn("[TaskTracker-{}] process failed !", instanceId, e);
	res = new ProcessResult(false, e.toString());
	}
	if (res == null) {
	log.warn("[TaskTracker-{}] processor return null !", instanceId);
	res = new ProcessResult(false, "Processor return null");
	}
	} while (!res.isSuccess() && taskContext.getCurrentRetryTimes() < taskContext.getMaxRetryTimes() && !timeoutFlag.get() && !stopFlag.get());
	executeThread.set(null);
	taskEndTime = System.currentTimeMillis();
	finished.set(true);
	result = res;
	status = result.isSuccess() ? TaskStatus.WORKER_PROCESS_SUCCESS : TaskStatus.WORKER_PROCESS_FAILED;
	// 取消超时检查任务
	if (timeoutCheckScheduledFuture != null) {
	timeoutCheckScheduledFuture.cancel(true);
	}
	log.info("[TaskTracker-{}] task complete ! create time:{},queue time:{},use time:{},result:{}", instanceId, createTime, taskStartTime - createTime, System.currentTimeMillis() - taskStartTime, result);
	// 执行完成后立即上报一次
	checkAndReportStatus();
	return result;
	}

	/**
	* 静态方法创建 TaskTracker
	*
	* @param req 服务端调度任务请求
	* @return LightTaskTracker
	*/
	public static LightTaskTracker create(ServerScheduleJobReq req, WorkerRuntime workerRuntime) {
	try {
	return new LightTaskTracker(req, workerRuntime);
	} catch (Exception e) {
	reportCreateErrorToServer(req, workerRuntime, e);
	}
	return null;
	}

	@Slf4j
	@Actor(path = WTT_PATH)
	public class TaskTrackerActor {

	private final WorkerRuntime workerRuntime;

	public TaskTrackerActor(WorkerRuntime workerRuntime) {
	this.workerRuntime = workerRuntime;
	}

	/**
	* 服务器任务调度处理器
	*/
	@Handler(path = WTT_HANDLER_RUN_JOB)
	public void onReceiveServerScheduleJobReq(ServerScheduleJobReq req) {
	log.debug("[TaskTrackerActor] server schedule job by request: {}.", req);
	Long instanceId = req.getInstanceId();
	// 区分轻量级任务模型以及重量级任务模型
	if (isLightweightTask(req)) {
	final LightTaskTracker taskTracker = LightTaskTrackerManager.getTaskTracker(instanceId);
	if (taskTracker != null) {
	log.warn("[TaskTrackerActor] LightTaskTracker({}) for instance(id={}) already exists.", taskTracker, instanceId);
	return;
	}
	// 判断是否已经 overload
	if (LightTaskTrackerManager.currentTaskTrackerSize() >= workerRuntime.getWorkerConfig().getMaxLightweightTaskNum() * LightTaskTrackerManager.OVERLOAD_FACTOR) {
	// ignore this request
	log.warn("[TaskTrackerActor] this worker is overload,ignore this request(instanceId={}),current size = {}!",instanceId,LightTaskTrackerManager.currentTaskTrackerSize());
	return;
	}
	if (LightTaskTrackerManager.currentTaskTrackerSize() >= workerRuntime.getWorkerConfig().getMaxLightweightTaskNum()) {
	log.warn("[TaskTrackerActor] this worker will be overload soon,current size = {}!",LightTaskTrackerManager.currentTaskTrackerSize());
	}
	// 创建轻量级任务
	LightTaskTrackerManager.atomicCreateTaskTracker(instanceId, ignore -> LightTaskTracker.create(req, workerRuntime));
	} else {
	HeavyTaskTracker taskTracker = HeavyTaskTrackerManager.getTaskTracker(instanceId);
	if (taskTracker != null) {
	log.warn("[TaskTrackerActor] HeavyTaskTracker({}) for instance(id={}) already exists.", taskTracker, instanceId);
	return;
	}
	// 判断是否已经 overload
	if (HeavyTaskTrackerManager.currentTaskTrackerSize() >= workerRuntime.getWorkerConfig().getMaxHeavyweightTaskNum()) {
	// ignore this request
	log.warn("[TaskTrackerActor] this worker is overload,ignore this request(instanceId={})! current size = {},", instanceId, HeavyTaskTrackerManager.currentTaskTrackerSize());
	return;
	}
	// 原子创建，防止多实例的存在
	HeavyTaskTrackerManager.atomicCreateTaskTracker(instanceId, ignore -> HeavyTaskTracker.create(req, workerRuntime));
	}
	}

	//......
	}

	private boolean isLightweightTask(ServerScheduleJobReq serverScheduleJobReq) {
	final ExecuteType executeType = ExecuteType.valueOf(serverScheduleJobReq.getExecuteType());
	// 非单机执行的一定不是
	if (executeType != ExecuteType.STANDALONE){
	return false;
	}
	TimeExpressionType timeExpressionType = TimeExpressionType.valueOf(serverScheduleJobReq.getTimeExpressionType());
	// 固定频率以及固定延迟的也一定不是
	return timeExpressionType != TimeExpressionType.FIXED_DELAY && timeExpressionType != TimeExpressionType.FIXED_RATE;
	}