@Data
public class InstanceInfo implements Serializable {

    /**
     * 基礎(chǔ)信息
     */
    private Long jobId;
    private Long instanceId;
    private Long wfInstanceId;

    /**
     * 任務(wù)執(zhí)行處理器信息
     */
    // 任務(wù)執(zhí)行類型，單機(jī)、廣播、MR
    private String executeType;
    // 處理器類型（JavaBean、Jar、腳本等）
    private String processorType;
    // 處理器信息
    private String processorInfo;
    // 定時類型
    private int timeExpressionType;

    /**
     * 超時時間
     */
    // 整個任務(wù)的總體超時時間
    private long instanceTimeoutMS;

    /**
     * 任務(wù)運行參數(shù)
     */
    // 任務(wù)級別的參數(shù)，相當(dāng)于類的static變量
    private String jobParams;
    // 實例級別的參數(shù)，相當(dāng)于類的普通變量
    private String instanceParams;


    // 每臺機(jī)器的處理線程數(shù)上限
    private int threadConcurrency;
    // 子任務(wù)重試次數(shù)（任務(wù)本身的重試機(jī)制由server控制）
    private int taskRetryNum;

    private String logConfig;
}

/**
     * 定時掃描數(shù)據(jù)庫中的task（出于內(nèi)存占用量考慮，每次最多獲取100個），并將需要執(zhí)行的任務(wù)派發(fā)出去
     */
    protected class Dispatcher implements Runnable {

        // 數(shù)據(jù)庫查詢限制，每次最多查詢幾個任務(wù)
        private static final int DB_QUERY_LIMIT = 100;

        @Override
        public void run() {

            if (finished.get()) {
                return;
            }

            Stopwatch stopwatch = Stopwatch.createStarted();

            // 1. 獲取可以派發(fā)任務(wù)的 ProcessorTracker
            List<String> availablePtIps = ptStatusHolder.getAvailableProcessorTrackers();

            // 2. 沒有可用 ProcessorTracker，本次不派發(fā)
            if (availablePtIps.isEmpty()) {
                log.debug("[TaskTracker-{}] no available ProcessorTracker now.", instanceId);
                return;
            }

            // 3. 避免大查詢，分批派發(fā)任務(wù)
            long currentDispatchNum = 0;
            long maxDispatchNum = availablePtIps.size() * instanceInfo.getThreadConcurrency() * 2L;
            AtomicInteger index = new AtomicInteger(0);

            // 4. 循環(huán)查詢數(shù)據(jù)庫，獲取需要派發(fā)的任務(wù)
            while (maxDispatchNum > currentDispatchNum) {

                int dbQueryLimit = Math.min(DB_QUERY_LIMIT, (int) maxDispatchNum);
                List<TaskDO> needDispatchTasks = taskPersistenceService.getTaskByStatus(instanceId, TaskStatus.WAITING_DISPATCH, dbQueryLimit);
                currentDispatchNum += needDispatchTasks.size();

                needDispatchTasks.forEach(task -> {
                    // 獲取 ProcessorTracker 地址，如果 Task 中自帶了 Address，則使用該 Address
                    String ptAddress = task.getAddress();
                    if (StringUtils.isEmpty(ptAddress) || RemoteConstant.EMPTY_ADDRESS.equals(ptAddress)) {
                        ptAddress = availablePtIps.get(index.getAndIncrement() % availablePtIps.size());
                    }
                    dispatchTask(task, ptAddress);
                });

                // 數(shù)量不足 或 查詢失敗，則終止循環(huán)
                if (needDispatchTasks.size() < dbQueryLimit) {
                    break;
                }
            }

            log.debug("[TaskTracker-{}] dispatched {} tasks,using time {}.", instanceId, currentDispatchNum, stopwatch.stop());
        }
    }

private int calThreadPoolSize() {
        ExecuteType executeType = ExecuteType.valueOf(instanceInfo.getExecuteType());
        ProcessorType processorType = ProcessorType.valueOf(instanceInfo.getProcessorType());

        // 腳本類自帶線程池，不過為了少一點邏輯判斷，還是象征性分配一個線程
        if (processorType == ProcessorType.PYTHON || processorType == ProcessorType.SHELL) {
            return 1;
        }

        if (executeType == ExecuteType.MAP_REDUCE || executeType == ExecuteType.MAP) {
            return instanceInfo.getThreadConcurrency();
        }
        if (TimeExpressionType.FREQUENT_TYPES.contains(instanceInfo.getTimeExpressionType())) {
            return instanceInfo.getThreadConcurrency();
        }
        return 2;
    }

private static final int THREAD_POOL_QUEUE_MAX_SIZE = 128;

    private void initThreadPool() {

        int poolSize = calThreadPoolSize();
        // 待執(zhí)行隊列，為了防止對內(nèi)存造成較大壓力，內(nèi)存隊列不能太大
        BlockingQueue<Runnable> queue = new ArrayBlockingQueue<>(THREAD_POOL_QUEUE_MAX_SIZE);
        // 自定義線程池中線程名稱 (PowerJob Processor Pool -> PPP)
        ThreadFactory threadFactory = new ThreadFactoryBuilder().setNameFormat("PPP-%d").build();
        // 拒絕策略：直接拋出異常
        RejectedExecutionHandler rejectionHandler = new ThreadPoolExecutor.AbortPolicy();

        threadPool = new ThreadPoolExecutor(poolSize, poolSize, 60L, TimeUnit.SECONDS, queue, threadFactory, rejectionHandler);

        // 當(dāng)沒有任務(wù)執(zhí)行時，允許銷毀核心線程（即線程池最終存活線程個數(shù)可能為0）
        threadPool.allowCoreThreadTimeOut(true);
    }

public void submitTask(TaskDO newTask) {

        // 一旦 ProcessorTracker 出現(xiàn)異常，所有提交到此處的任務(wù)直接返回失敗，防止形成死鎖
        // 死鎖分析：TT創(chuàng)建PT，PT創(chuàng)建失敗，無法定期匯報心跳，TT長時間未收到PT心跳，認(rèn)為PT宕機(jī)（確實宕機(jī)了），無法選擇可用的PT再次派發(fā)任務(wù)，死鎖形成，GG斯密達(dá) T_T
        if (lethal) {
            ProcessorReportTaskStatusReq report = new ProcessorReportTaskStatusReq()
                    .setInstanceId(instanceId)
                    .setSubInstanceId(newTask.getSubInstanceId())
                    .setTaskId(newTask.getTaskId())
                    .setStatus(TaskStatus.WORKER_PROCESS_FAILED.getValue())
                    .setResult(lethalReason)
                    .setReportTime(System.currentTimeMillis());

            TransportUtils.ptReportTask(report, taskTrackerAddress, workerRuntime);
            return;
        }

        boolean success = false;
        // 1. 設(shè)置值并提交執(zhí)行
        newTask.setInstanceId(instanceInfo.getInstanceId());
        newTask.setAddress(taskTrackerAddress);

        HeavyProcessorRunnable heavyProcessorRunnable = new HeavyProcessorRunnable(instanceInfo, taskTrackerAddress, newTask, processorBean, omsLogger, statusReportRetryQueue, workerRuntime);
        try {
            threadPool.submit(heavyProcessorRunnable);
            success = true;
        } catch (RejectedExecutionException ignore) {
            log.warn("[ProcessorTracker-{}] submit task(taskId={},taskName={}) to ThreadPool failed due to ThreadPool has too much task waiting to process, this task will dispatch to other ProcessorTracker.",
                    instanceId, newTask.getTaskId(), newTask.getTaskName());
        } catch (Exception e) {
            log.error("[ProcessorTracker-{}] submit task(taskId={},taskName={}) to ThreadPool failed.", instanceId, newTask.getTaskId(), newTask.getTaskName(), e);
        }

        // 2. 回復(fù)接收成功
        if (success) {
            ProcessorReportTaskStatusReq reportReq = new ProcessorReportTaskStatusReq();
            reportReq.setInstanceId(instanceId);
            reportReq.setSubInstanceId(newTask.getSubInstanceId());
            reportReq.setTaskId(newTask.getTaskId());
            reportReq.setStatus(TaskStatus.WORKER_RECEIVED.getValue());
            reportReq.setReportTime(System.currentTimeMillis());

            TransportUtils.ptReportTask(reportReq, taskTrackerAddress, workerRuntime);

            log.debug("[ProcessorTracker-{}] submit task(taskId={}, taskName={}) success, current queue size: {}.",
                    instanceId, newTask.getTaskId(), newTask.getTaskName(), threadPool.getQueue().size());
        }
    }

@Handler(path = WTT_HANDLER_REPORT_TASK_STATUS)
    public AskResponse onReceiveProcessorReportTaskStatusReq(ProcessorReportTaskStatusReq req) {

        int taskStatus = req.getStatus();
        // 只有重量級任務(wù)才會有兩級任務(wù)狀態(tài)上報的機(jī)制
        HeavyTaskTracker taskTracker = HeavyTaskTrackerManager.getTaskTracker(req.getInstanceId());

        // 手動停止 TaskTracker 的情況下會出現(xiàn)這種情況
        if (taskTracker == null) {
            log.warn("[TaskTrackerActor] receive ProcessorReportTaskStatusReq({}) but system can't find TaskTracker.", req);
            return null;
        }

        if (ProcessorReportTaskStatusReq.BROADCAST.equals(req.getCmd())) {
            taskTracker.broadcast(taskStatus == TaskStatus.WORKER_PROCESS_SUCCESS.getValue(), req.getSubInstanceId(), req.getTaskId(), req.getResult());
        }

        taskTracker.updateTaskStatus(req.getSubInstanceId(), req.getTaskId(), taskStatus, req.getReportTime(), req.getResult());

        // 更新工作流上下文
        taskTracker.updateAppendedWfContext(req.getAppendedWfContext());

        // 結(jié)束狀態(tài)需要回復(fù)接受成功
        if (TaskStatus.FINISHED_STATUS.contains(taskStatus)) {
            return AskResponse.succeed(null);
        }

        return null;
    }

@Getter
@AllArgsConstructor
public enum TaskStatus {

    WAITING_DISPATCH(1, "等待調(diào)度器調(diào)度"),
    DISPATCH_SUCCESS_WORKER_UNCHECK(2, "調(diào)度成功（但不保證worker收到）"),
    WORKER_RECEIVED(3, "worker接收成功，但未開始執(zhí)行"),
    WORKER_PROCESSING(4, "worker正在執(zhí)行"),
    WORKER_PROCESS_FAILED(5, "worker執(zhí)行失敗"),
    WORKER_PROCESS_SUCCESS(6, "worker執(zhí)行成功");

    public static final Set<Integer> FINISHED_STATUS = Sets.newHashSet(WORKER_PROCESS_FAILED.value, WORKER_PROCESS_SUCCESS.value);

    private final int value;
    private final String des;

    public static TaskStatus of(int v) {
        for (TaskStatus taskStatus : values()) {
            if (v == taskStatus.value) {
                return taskStatus;
            }
        }
        throw new IllegalArgumentException("no TaskStatus match the value of " + v);
    }
}

public void run() {
        // 切換線程上下文類加載器（否則用的是 Worker 類加載器，不存在容器類，在序列化/反序列化時會報 ClassNotFoundException）
        Thread.currentThread().setContextClassLoader(processorBean.getClassLoader());
        try {
            innerRun();
        } catch (InterruptedException ignore) {
            // ignore
        } catch (Throwable e) {
            reportStatus(TaskStatus.WORKER_PROCESS_FAILED, e.toString(), null, null);
            log.error("[ProcessorRunnable-{}] execute failed, please contact the author(@KFCFans) to fix the bug!", task.getInstanceId(), e);
        } finally {
            ThreadLocalStore.clear();
        }
    }

    public void innerRun() throws InterruptedException {

        final BasicProcessor processor = processorBean.getProcessor();

        String taskId = task.getTaskId();
        Long instanceId = task.getInstanceId();

        log.debug("[ProcessorRunnable-{}] start to run task(taskId={}&taskName={})", instanceId, taskId, task.getTaskName());
        ThreadLocalStore.setTask(task);
        ThreadLocalStore.setRuntimeMeta(workerRuntime);

        // 0. 構(gòu)造任務(wù)上下文
        WorkflowContext workflowContext = constructWorkflowContext();
        TaskContext taskContext = constructTaskContext();
        taskContext.setWorkflowContext(workflowContext);
        // 1. 上報執(zhí)行信息
        reportStatus(TaskStatus.WORKER_PROCESSING, null, null, null);

        ProcessResult processResult;
        ExecuteType executeType = ExecuteType.valueOf(instanceInfo.getExecuteType());

        // 2. 根任務(wù) & 廣播執(zhí)行 特殊處理
        if (TaskConstant.ROOT_TASK_NAME.equals(task.getTaskName()) && executeType == ExecuteType.BROADCAST) {
            // 廣播執(zhí)行：先選本機(jī)執(zhí)行 preProcess，完成后 TaskTracker 再為所有 Worker 生成子 Task
            handleBroadcastRootTask(instanceId, taskContext);
            return;
        }

        // 3. 最終任務(wù)特殊處理（一定和 TaskTracker 處于相同的機(jī)器）
        if (TaskConstant.LAST_TASK_NAME.equals(task.getTaskName())) {
            handleLastTask(taskId, instanceId, taskContext, executeType);
            return;
        }

        // 4. 正式提交運行
        try {
            processResult = processor.process(taskContext);
            if (processResult == null) {
                processResult = new ProcessResult(false, "ProcessResult can't be null");
            }
        } catch (Throwable e) {
            log.warn("[ProcessorRunnable-{}] task(id={},name={}) process failed.", instanceId, taskContext.getTaskId(), taskContext.getTaskName(), e);
            processResult = new ProcessResult(false, e.toString());
        }
        reportStatus(processResult.isSuccess() ? TaskStatus.WORKER_PROCESS_SUCCESS : TaskStatus.WORKER_PROCESS_FAILED, suit(processResult.getMsg()), null, workflowContext.getAppendedContextData());
    }