本身動手實現分佈式任務調度框架(續)

時間 2019-12-02

原文原文鏈接

　　以前寫過一篇:本身動手實現分佈式任務調度框架原本是用來閒來分享一下本身的思惟方式，時至今日發現竟然有些人正在使用了，本着對代碼負責任的態度，對代碼部分已知bug進行了修改，並增長了若干功能，如當即啓動，實時中止等功能，新增長的功能會在這一篇作詳細的說明。html

　　提到分佈式任務調度，市面上自己已經有一些框架工具可使用，可是我的以爲功能作的都太豐富，架構都過於複雜，因此纔有了我重複造輪子。我的喜歡把複雜的問題簡單化，利用有限的資源實現竟可能多的功能。由於有幾個朋友問部署方式，這裏再次強調下：個人這個服務能夠直接打成jar放在本身本地倉庫，而後依賴進去，或者直接copy代碼過去，當成本身項目的一部分就能夠了。也就是說跟隨大家本身的項目啓動，因此我這裏也沒有寫界面。下面先談談怎麼基於上次的代碼實現任務當即啓動吧!前端

　　調度和本身服務整合後部署圖抽象成以下:java

　　用戶在前端點擊當即請求按鈕，經過各類負載均衡軟件或者設備，到達某臺機器的某個帶有本調度框架的服務，而後進行具體的執行，也就是說這個當即啓動就是一個最多見最簡單的請求，沒有過多複雜的問題（好比多節點會不會重複執行這些）。最簡單的辦法，當用戶請求過來直接用一個線程或者線程池執行用戶點的那個任務的邏輯代碼就好了，固然我這裏沒有那麼粗暴，現有的調度代碼資源以下:node

package com.rdpaas.task.scheduler;

import com.rdpaas.task.common.Invocation;
import com.rdpaas.task.common.Node;
import com.rdpaas.task.common.NotifyCmd;
import com.rdpaas.task.common.Task;
import com.rdpaas.task.common.TaskDetail;
import com.rdpaas.task.common.TaskStatus;
import com.rdpaas.task.config.EasyJobConfig;
import com.rdpaas.task.repository.NodeRepository;
import com.rdpaas.task.repository.TaskRepository;
import com.rdpaas.task.strategy.Strategy;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * 任務調度器
 * @author rongdi
 * @date 2019-03-13 21:15
 */
@Component
public class TaskExecutor {

    private static final Logger logger = LoggerFactory.getLogger(TaskExecutor.class);

    @Autowired
    private TaskRepository taskRepository;

    @Autowired
    private NodeRepository nodeRepository;

    @Autowired
    private EasyJobConfig config;

    /**
     * 建立任務到期延時隊列
      */
    private DelayQueue<DelayItem<Task>> taskQueue = new DelayQueue<>();

    /**
     * 能夠明確知道最多隻會運行2個線程，直接使用系統自帶工具就能夠了
     */
    private ExecutorService bossPool = Executors.newFixedThreadPool(2);

    /**
     * 正在執行的任務的Future
     */
    private Map<Long,Future> doingFutures = new HashMap<>();

    /**
     * 聲明工做線程池
     */
    private ThreadPoolExecutor workerPool;
    
    /**
     * 獲取任務的策略
     */
    private Strategy strategy;


    @PostConstruct
    public void init() {
        /**
         * 根據配置選擇一個節點獲取任務的策略
         */
        strategy = Strategy.choose(config.getNodeStrategy());
        /**
         * 自定義線程池，初始線程數量corePoolSize，線程池等待隊列大小queueSize，當初始線程都有任務，而且等待隊列滿後
         * 線程數量會自動擴充最大線程數maxSize，當新擴充的線程空閒60s後自動回收.自定義線程池是由於Executors那幾個線程工具
         * 各有各的弊端，不適合生產使用
         */
        workerPool = new ThreadPoolExecutor(config.getCorePoolSize(), config.getMaxPoolSize(), 60, TimeUnit.SECONDS, new ArrayBlockingQueue<>(config.getQueueSize()));
        /**
         * 執行待處理任務加載線程
         */
        bossPool.execute(new Loader());
        /**
         * 執行任務調度線程
         */
        bossPool.execute(new Boss());
    
    }

    class Loader implements Runnable {

        @Override
        public void run() {
            for(;;) {
                try { 
                    /**
                     * 先獲取可用的節點列表
                     */
                    List<Node> nodes = nodeRepository.getEnableNodes(config.getHeartBeatSeconds() * 2);
                    if(nodes == null || nodes.isEmpty()) {
                        continue;
                    }
                    /**
                     * 查找還有指定時間(單位秒)纔開始的主任務列表
                     */
                    List<Task> tasks = taskRepository.listNotStartedTasks(config.getFetchDuration());
                    if(tasks == null || tasks.isEmpty()) {
                        continue;
                    }
                    for(Task task:tasks) {
                        
                        boolean accept = strategy.accept(nodes, task, config.getNodeId());
                        /**
                         * 不應本身拿就不要搶
                         */
                        if(!accept) {
                            continue;
                        }
                        /**
                         * 先設置成待執行
                         */
                        task.setStatus(TaskStatus.PENDING);
                        task.setNodeId(config.getNodeId());
                        /**
                         * 使用樂觀鎖嘗試更新狀態，若是更新成功，其餘節點就不會更新成功。若是其它節點也正在查詢未完成的
                         * 任務列表和當前這段時間有節點已經更新了這個任務，version必然和查出來時候的version不同了,這裏更新
                         * 必然會返回0了
                         */
                        int n = taskRepository.updateWithVersion(task);
                        Date nextStartTime = task.getNextStartTime();
                        if(n == 0 || nextStartTime == null) {
                            continue;
                        }
                        /**
                         * 封裝成延時對象放入延時隊列,這裏再查一次是由於上面樂觀鎖已經更新了版本，會致使後面結束任務更新不成功
                         */
                        task = taskRepository.get(task.getId());
                        DelayItem<Task> delayItem = new DelayItem<Task>(nextStartTime.getTime() - new Date().getTime(), task);
                        taskQueue.offer(delayItem);
                        
                    }
                    Thread.sleep(config.getFetchPeriod());
                } catch(Exception e) {
                    logger.error("fetch task list failed,cause by:{}", e);
                }
            }
        }
        
    }
    
    class Boss implements Runnable {
        @Override
        public void run() {
            for (;;) {
                try {
                     /**
                     * 時間到了就能夠從延時隊列拿出任務對象,而後交給worker線程池去執行
                     */
                    DelayItem<Task> item = taskQueue.take();
                    if(item != null && item.getItem() != null) {
                        Task task = item.getItem();
                        /**
                         * 真正開始執行了設置成執行中
                         */
                        task.setStatus(TaskStatus.DOING);
                        /**
                         * loader線程中已經使用樂觀鎖控制了，這裏不必了
                         */
                        taskRepository.update(task);
                        /**
                         * 提交到線程池
                         */
                        Future future = workerPool.submit(new Worker(task));
                        /**
                         * 暫存在doingFutures
                         */
                        doingFutures.put(task.getId(),future);
                    }
                     
                } catch (Exception e) {
                    logger.error("fetch task failed,cause by:{}", e);
                }
            }
        }

    }

    class Worker implements Callable<String> {

        private Task task;

        public Worker(Task task) {
            this.task = task;
        }

        @Override
        public String call() {
            logger.info("Begin to execute task:{}",task.getId());
            TaskDetail detail = null;
            try {
                //開始任務
                detail = taskRepository.start(task);
                if(detail == null) return null;
                //執行任務
                task.getInvokor().invoke();
                //完成任務
                finish(task,detail);
                logger.info("finished execute task:{}",task.getId());
                /**
                 * 執行完後刪了
                 */
                doingFutures.remove(task.getId());
            } catch (Exception e) {
                logger.error("execute task:{} error,cause by:{}",task.getId(), e);
                try {
                    taskRepository.fail(task,detail,e.getCause().getMessage());
                } catch(Exception e1) {
                    logger.error("fail task:{} error,cause by:{}",task.getId(), e);
                }
            }
            return null;
        }

    }

    /**
     * 完成子任務，若是父任務失敗了，子任務不會執行
     * @param task
     * @param detail
     * @throws Exception
     */
    private void finish(Task task,TaskDetail detail) throws Exception {

        //查看是否有子類任務
        List<Task> childTasks = taskRepository.getChilds(task.getId());
        if(childTasks == null || childTasks.isEmpty()) {
            //當沒有子任務時完成父任務
            taskRepository.finish(task,detail);
            return;
        } else {
            for (Task childTask : childTasks) {
                //開始任務
                TaskDetail childDetail = null;
                try {
                    //將子任務狀態改爲執行中
                    childTask.setStatus(TaskStatus.DOING);
                    childTask.setNodeId(config.getNodeId());
                    //開始子任務
                    childDetail = taskRepository.startChild(childTask,detail);
                    //使用樂觀鎖更新下狀態，否則這裏可能和恢復線程產生併發問題
                    int n = taskRepository.updateWithVersion(childTask);
                    if (n > 0) {
                        //再從數據庫取一下，避免上面update修改後version不一樣步
                        childTask = taskRepository.get(childTask.getId());
                        //執行子任務
                        childTask.getInvokor().invoke();
                        //完成子任務
                        finish(childTask, childDetail);
                    }
                } catch (Exception e) {
                    logger.error("execute child task error,cause by:{}", e);
                    try {
                        taskRepository.fail(childTask, childDetail, e.getCause().getMessage());
                    } catch (Exception e1) {
                        logger.error("fail child task error,cause by:{}", e);
                    }
                }
            }
            /**
             * 當有子任務時完成子任務後再完成父任務
             */
            taskRepository.finish(task,detail);

        }

    }

    /**
     * 添加任務
     * @param name
     * @param cronExp
     * @param invockor
     * @return
     * @throws Exception
     */
    public long addTask(String name, String cronExp, Invocation invockor) throws Exception {
        Task task = new Task(name,cronExp,invockor);
        return taskRepository.insert(task);
    }

    /**
     * 添加子任務
     * @param pid
     * @param name
     * @param cronExp
     * @param invockor
     * @return
     * @throws Exception
     */
    public long addChildTask(Long pid,String name, String cronExp, Invocation invockor) throws Exception {
        Task task = new Task(name,cronExp,invockor);
        task.setPid(pid);
        return taskRepository.insert(task);
    }

   
}

　　上面主要就是三組線程，Loader負責加載將要執行的任務放入本地的任務隊列，Boss線程負責取出任務隊列的任務，而後分配Worker線程池的一個線程去執行。由上面的代碼能夠看到若是要當即執行，其實只須要把一個延時爲0的任務放入任務隊列，等着Boss線程去取而後分配給worker執行就能夠實現了，代碼以下:git

    /**
     * 當即執行任務，就是設置一下延時爲0加入任務隊列就行了，這個能夠外部直接調用
     * @param taskId
     * @return
     */
    public boolean startNow(Long taskId) {
        Task task = taskRepository.get(taskId);
        task.setStatus(TaskStatus.DOING);
        taskRepository.update(task);
        DelayItem<Task> delayItem = new DelayItem<Task>(0L, task);
        return taskQueue.offer(delayItem);
    }

　　啓動不用再多說，下面介紹一下中止任務，根據面向對象的思惟，用戶要想中止一個任務，最終執行中止任務的就是正在執行任務的那個節點。中止任務有兩種狀況，第一種任務沒有正在運行如何中止，第二種是任務正在運行如何中止。第一種其實直接改變一下任務對象的狀態爲中止就好了，沒必要多說。下面主要考慮如何中止正在運行的任務，細心的朋友可能已經發現上面代碼和以前那一篇代碼有點區別，以前用的Runnble做爲線程實現接口，這個用了Callable，其實在java中中止線程池中正在運行的線程最經常使用的就是直接調用future的cancel方法了，要想獲取到這個future對象就須要將之前實現Runnbale改爲實現Callable，而後提交到線程池由execute改爲submit就能夠了，而後每次提交到線程池獲得的future對象使用taskId一塊兒保存在一個map中，方便根據taskId隨時找到。固然任務執行完後要及時刪除這個map裏的任務，以避免常駐其中致使內存溢出。中止任務的請求流程以下程序員

　　圖仍是原來的圖，可是這時候狀況不同了，由於中止任務的時候假如當前正在執行這個任務的節點處於服務1，負載均衡是不知道要去把你引到服務1的，他可能會引入到服務2，那就悲劇了，因此通用的作法就是中止請求過來無論落到哪一個節點上，那個節點就往一個公用的mq上發一個帶有中止任務業務含義的消息，各個節點訂閱這個消息，而後判斷都判斷任務在不在本身這裏執行，若是在就執行中止操做。可是這樣勢必讓咱們的調度服務又要依賴一個外部的消息隊列服務，就算很方便的就能夠引入一個外部的消息隊列，可是你真的能夠駕馭的了嗎，消息丟了咋辦，重複發送了咋辦，消息服務掛了咋辦，網絡斷了咋辦，又引入了一大堆問題，那我是否是又要寫n篇文章來分別解決這些問題。每每現實倒是就是這麼殘酷，你解決了一個問題，引入了更多的問題，這就是爲何bug永遠改不完的道理了。固然這不是個人風格，個人風格是利用有限的資源作儘量多的事情（多是因爲我工做的企業都是那種資源貧瘠的，養成了我這種習慣，土豪公司的程序員請繞道，哈哈）。github

　　簡化一下問題:目前的問題就是如何讓正在執行任務的節點知道，而後中止正在執行的這個任務，其實就是這個中止通知如何實現。這難免讓我想起了12306網站上買票，其實咱們做爲老百姓多麼但願12306能夠在有票的時候發個短信通知一下咱們，而後咱們上去搶，可是現實倒是，你要麼使用軟件一直刷，要麼是本身隔一段時間上去瞄一下有沒有票。若是把有票了給咱們發短信通知定義爲異步通知，那麼這種咱們要隔一段時間本身去瞄一下的方式就是同步輪訓。這兩種方式都能達到告知的目的，關鍵的區別在於你到底有沒有時間去一直去瞄，不過相比於能夠回家，這些時間都是值得的。我的認爲軟件的設計其實就是一個權衡是否值得的過程。若是約定了不使用外部消息隊列這種異步通知的方式，那麼咱們只能使用同步輪訓的方式了。不過正好咱們的任務調度自己已經有一個心跳機制，沒隔一段時間就去更新一下節點狀態，若是咱們把用戶的中止請求做爲命令信息更新到每一個節點的上，而後隨着心跳獲取到這個節點的信息，而後判斷這個命令，作相應的處理是否是就能夠完美解決這個問題。值得嗎？很明顯是值得的，咱們只是在心跳邏輯上加一個小小的反作用就實現了通知功能了。代碼以下spring

package com.rdpaas.task.common;

/**
 * @author rongdi
 * @date 2019/11/26
 */
public enum NotifyCmd {

    //沒有通知，默認狀態
    NO_NOTIFY(0),
    //開啓任務(Task)
    START_TASK(1),
    //修改任務(Task)
    EDIT_TASK(2),
    //中止任務(Task)
    STOP_TASK(3);

    int id;

    NotifyCmd(int id) {
        this.id = id;
    }

    public int getId() {
        return id;
    }

    public static NotifyCmd valueOf(int id) {
        switch (id) {
            case 1:
                return START_TASK;
            case 2:
                return EDIT_TASK;
            case 3:
                return STOP_TASK;
            default:
                return NO_NOTIFY;
        }
    }

}

package com.rdpaas.task.handles;

import com.rdpaas.task.common.NotifyCmd;
import com.rdpaas.task.utils.SpringContextUtil;

/**
 * @author: rongdi
 * @date:
 */
public interface NotifyHandler<T> {

    static NotifyHandler chooseHandler(NotifyCmd notifyCmd) {
        return SpringContextUtil.getByTypeAndName(NotifyHandler.class,notifyCmd.toString());
    }

    public void update(T t);

}

package com.rdpaas.task.handles;

import com.rdpaas.task.scheduler.TaskExecutor;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

/**
 * @author: rongdi
 * @date:
 */
@Component("STOP_TASK")
public class StopTaskHandler implements NotifyHandler<Long> {

    @Autowired
    private TaskExecutor taskExecutor;

    @Override
    public void update(Long taskId) {
        taskExecutor.stop(taskId);
    }

}

class HeartBeat implements Runnable {
        @Override
        public void run() {
            for(;;) {
                try {
                    /**
                     * 時間到了就能夠從延時隊列拿出節點對象，而後更新時間和序號，
                     * 最後再新建一個超時時間爲心跳時間的節點對象放入延時隊列，造成循環的心跳
                     */
                    DelayItem<Node> item = heartBeatQueue.take();
                    if(item != null && item.getItem() != null) {
                        Node node = item.getItem();
                        handHeartBeat(node);
                    }
                    heartBeatQueue.offer(new DelayItem<>(config.getHeartBeatSeconds() * 1000,new Node(config.getNodeId())));
                } catch (Exception e) {
                    logger.error("task heart beat error,cause by:{} ",e);
                }
            }
        }
    }

    /**
     * 處理節點心跳
     * @param node
     */
    private void handHeartBeat(Node node) {
        if(node == null) {
            return;
        }
        /**
         * 先看看數據庫是否存在這個節點
         * 若是不存在：先查找下一個序號，而後設置到node對象中，最後插入
         * 若是存在：直接根據nodeId更新當前節點的序號和時間
         */
        Node currNode= nodeRepository.getByNodeId(node.getNodeId());
        if(currNode == null) {
            node.setRownum(nodeRepository.getNextRownum());
            nodeRepository.insert(node);
        } else  {
            nodeRepository.updateHeartBeat(node.getNodeId());
            NotifyCmd cmd = currNode.getNotifyCmd();
            String notifyValue = currNode.getNotifyValue();
            if(cmd != null && cmd != NotifyCmd.NO_NOTIFY) {
                /**
                 * 藉助心跳作一下通知的事情，好比及時中止正在執行的任務
                 * 根據指令名稱查找Handler
                 */
                NotifyHandler handler = NotifyHandler.chooseHandler(currNode.getNotifyCmd());
                if(handler == null || StringUtils.isEmpty(notifyValue)) {
                    return;
                }
                /**
                 * 執行操做
                 */
                handler.update(Long.valueOf(notifyValue));
            }
            
        }


    }

　　最終的任務調度代碼以下:數據庫

package com.rdpaas.task.scheduler;

import com.rdpaas.task.common.Invocation;
import com.rdpaas.task.common.Node;
import com.rdpaas.task.common.NotifyCmd;
import com.rdpaas.task.common.Task;
import com.rdpaas.task.common.TaskDetail;
import com.rdpaas.task.common.TaskStatus;
import com.rdpaas.task.config.EasyJobConfig;
import com.rdpaas.task.repository.NodeRepository;
import com.rdpaas.task.repository.TaskRepository;
import com.rdpaas.task.strategy.Strategy;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * 任務調度器
 * @author rongdi
 * @date 2019-03-13 21:15
 */
@Component
public class TaskExecutor {

    private static final Logger logger = LoggerFactory.getLogger(TaskExecutor.class);

    @Autowired
    private TaskRepository taskRepository;

    @Autowired
    private NodeRepository nodeRepository;

    @Autowired
    private EasyJobConfig config;

    /**
     * 建立任務到期延時隊列
      */
    private DelayQueue<DelayItem<Task>> taskQueue = new DelayQueue<>();

    /**
     * 能夠明確知道最多隻會運行2個線程，直接使用系統自帶工具就能夠了
     */
    private ExecutorService bossPool = Executors.newFixedThreadPool(2);

    /**
     * 正在執行的任務的Future
     */
    private Map<Long,Future> doingFutures = new HashMap<>();

    /**
     * 聲明工做線程池
     */
    private ThreadPoolExecutor workerPool;
    
    /**
     * 獲取任務的策略
     */
    private Strategy strategy;


    @PostConstruct
    public void init() {
        /**
         * 根據配置選擇一個節點獲取任務的策略
         */
        strategy = Strategy.choose(config.getNodeStrategy());
        /**
         * 自定義線程池，初始線程數量corePoolSize，線程池等待隊列大小queueSize，當初始線程都有任務，而且等待隊列滿後
         * 線程數量會自動擴充最大線程數maxSize，當新擴充的線程空閒60s後自動回收.自定義線程池是由於Executors那幾個線程工具
         * 各有各的弊端，不適合生產使用
         */
        workerPool = new ThreadPoolExecutor(config.getCorePoolSize(), config.getMaxPoolSize(), 60, TimeUnit.SECONDS, new ArrayBlockingQueue<>(config.getQueueSize()));
        /**
         * 執行待處理任務加載線程
         */
        bossPool.execute(new Loader());
        /**
         * 執行任務調度線程
         */
        bossPool.execute(new Boss());
    
    }

    class Loader implements Runnable {

        @Override
        public void run() {
            for(;;) {
                try { 
                    /**
                     * 先獲取可用的節點列表
                     */
                    List<Node> nodes = nodeRepository.getEnableNodes(config.getHeartBeatSeconds() * 2);
                    if(nodes == null || nodes.isEmpty()) {
                        continue;
                    }
                    /**
                     * 查找還有指定時間(單位秒)纔開始的主任務列表
                     */
                    List<Task> tasks = taskRepository.listNotStartedTasks(config.getFetchDuration());
                    if(tasks == null || tasks.isEmpty()) {
                        continue;
                    }
                    for(Task task:tasks) {
                        
                        boolean accept = strategy.accept(nodes, task, config.getNodeId());
                        /**
                         * 不應本身拿就不要搶
                         */
                        if(!accept) {
                            continue;
                        }
                        /**
                         * 先設置成待執行
                         */
                        task.setStatus(TaskStatus.PENDING);
                        task.setNodeId(config.getNodeId());
                        /**
                         * 使用樂觀鎖嘗試更新狀態，若是更新成功，其餘節點就不會更新成功。若是其它節點也正在查詢未完成的
                         * 任務列表和當前這段時間有節點已經更新了這個任務，version必然和查出來時候的version不同了,這裏更新
                         * 必然會返回0了
                         */
                        int n = taskRepository.updateWithVersion(task);
                        Date nextStartTime = task.getNextStartTime();
                        if(n == 0 || nextStartTime == null) {
                            continue;
                        }
                        /**
                         * 封裝成延時對象放入延時隊列,這裏再查一次是由於上面樂觀鎖已經更新了版本，會致使後面結束任務更新不成功
                         */
                        task = taskRepository.get(task.getId());
                        DelayItem<Task> delayItem = new DelayItem<Task>(nextStartTime.getTime() - new Date().getTime(), task);
                        taskQueue.offer(delayItem);
                        
                    }
                    Thread.sleep(config.getFetchPeriod());
                } catch(Exception e) {
                    logger.error("fetch task list failed,cause by:{}", e);
                }
            }
        }
        
    }
    
    class Boss implements Runnable {
        @Override
        public void run() {
            for (;;) {
                try {
                     /**
                     * 時間到了就能夠從延時隊列拿出任務對象,而後交給worker線程池去執行
                     */
                    DelayItem<Task> item = taskQueue.take();
                    if(item != null && item.getItem() != null) {
                        Task task = item.getItem();
                        /**
                         * 真正開始執行了設置成執行中
                         */
                        task.setStatus(TaskStatus.DOING);
                        /**
                         * loader線程中已經使用樂觀鎖控制了，這裏不必了
                         */
                        taskRepository.update(task);
                        /**
                         * 提交到線程池
                         */
                        Future future = workerPool.submit(new Worker(task));
                        /**
                         * 暫存在doingFutures
                         */
                        doingFutures.put(task.getId(),future);
                    }
                     
                } catch (Exception e) {
                    logger.error("fetch task failed,cause by:{}", e);
                }
            }
        }

    }

    class Worker implements Callable<String> {

        private Task task;

        public Worker(Task task) {
            this.task = task;
        }

        @Override
        public String call() {
            logger.info("Begin to execute task:{}",task.getId());
            TaskDetail detail = null;
            try {
                //開始任務
                detail = taskRepository.start(task);
                if(detail == null) return null;
                //執行任務
                task.getInvokor().invoke();
                //完成任務
                finish(task,detail);
                logger.info("finished execute task:{}",task.getId());
                /**
                 * 執行完後刪了
                 */
                doingFutures.remove(task.getId());
            } catch (Exception e) {
                logger.error("execute task:{} error,cause by:{}",task.getId(), e);
                try {
                    taskRepository.fail(task,detail,e.getCause().getMessage());
                } catch(Exception e1) {
                    logger.error("fail task:{} error,cause by:{}",task.getId(), e);
                }
            }
            return null;
        }

    }

    /**
     * 完成子任務，若是父任務失敗了，子任務不會執行
     * @param task
     * @param detail
     * @throws Exception
     */
    private void finish(Task task,TaskDetail detail) throws Exception {

        //查看是否有子類任務
        List<Task> childTasks = taskRepository.getChilds(task.getId());
        if(childTasks == null || childTasks.isEmpty()) {
            //當沒有子任務時完成父任務
            taskRepository.finish(task,detail);
            return;
        } else {
            for (Task childTask : childTasks) {
                //開始任務
                TaskDetail childDetail = null;
                try {
                    //將子任務狀態改爲執行中
                    childTask.setStatus(TaskStatus.DOING);
                    childTask.setNodeId(config.getNodeId());
                    //開始子任務
                    childDetail = taskRepository.startChild(childTask,detail);
                    //使用樂觀鎖更新下狀態，否則這裏可能和恢復線程產生併發問題
                    int n = taskRepository.updateWithVersion(childTask);
                    if (n > 0) {
                        //再從數據庫取一下，避免上面update修改後version不一樣步
                        childTask = taskRepository.get(childTask.getId());
                        //執行子任務
                        childTask.getInvokor().invoke();
                        //完成子任務
                        finish(childTask, childDetail);
                    }
                } catch (Exception e) {
                    logger.error("execute child task error,cause by:{}", e);
                    try {
                        taskRepository.fail(childTask, childDetail, e.getCause().getMessage());
                    } catch (Exception e1) {
                        logger.error("fail child task error,cause by:{}", e);
                    }
                }
            }
            /**
             * 當有子任務時完成子任務後再完成父任務
             */
            taskRepository.finish(task,detail);

        }

    }

    /**
     * 添加任務
     * @param name
     * @param cronExp
     * @param invockor
     * @return
     * @throws Exception
     */
    public long addTask(String name, String cronExp, Invocation invockor) throws Exception {
        Task task = new Task(name,cronExp,invockor);
        return taskRepository.insert(task);
    }

    /**
     * 添加子任務
     * @param pid
     * @param name
     * @param cronExp
     * @param invockor
     * @return
     * @throws Exception
     */
    public long addChildTask(Long pid,String name, String cronExp, Invocation invockor) throws Exception {
        Task task = new Task(name,cronExp,invockor);
        task.setPid(pid);
        return taskRepository.insert(task);
    }

    /**
     * 當即執行任務，就是設置一下延時爲0加入任務隊列就行了，這個能夠外部直接調用
     * @param taskId
     * @return
     */
    public boolean startNow(Long taskId) {
        Task task = taskRepository.get(taskId);
        task.setStatus(TaskStatus.DOING);
        taskRepository.update(task);
        DelayItem<Task> delayItem = new DelayItem<Task>(0L, task);
        return taskQueue.offer(delayItem);
    }

    /**
     * 當即中止正在執行的任務，留給外部調用的方法
     * @param taskId
     * @return
     */
    public boolean stopNow(Long taskId) {
        Task task = taskRepository.get(taskId);
        if(task == null) {
            return false;
        }
        /**
         * 該任務不是正在執行，直接修改task狀態爲已完成便可
         */
        if(task.getStatus() != TaskStatus.DOING) {
            task.setStatus(TaskStatus.STOP);
            taskRepository.update(task);
            return true;
        }
        /**
         * 該任務正在執行，使用節點配合心跳發佈停用通知
         */
        int n = nodeRepository.updateNotifyInfo(NotifyCmd.STOP_TASK,String.valueOf(taskId));
        return n > 0;
    }

    /**
     * 當即中止正在執行的任務，這個不須要本身調用，是給心跳線程調用
     * @param taskId
     * @return
     */
    public boolean stop(Long taskId) {
        Task task = taskRepository.get(taskId);
        /**
         * 不是本身節點的任務，本節點不能執行停用
         */
        if(task == null || !config.getNodeId().equals(task.getNodeId())) {
            return false;
        }
        /**
         * 拿到正在執行任務的future，而後強制停用，並刪除doingFutures的任務
         */
        Future future = doingFutures.get(taskId);
        boolean flag =  future.cancel(true);
        if(flag) {
            doingFutures.remove(taskId);
            /**
             * 修改狀態爲已停用
             */
            task.setStatus(TaskStatus.STOP);
            taskRepository.update(task);
        }
        /**
         * 重置通知信息，避免重複執行停用通知
         */
        nodeRepository.resetNotifyInfo(NotifyCmd.STOP_TASK);
        return flag;
    }
}