摘要：基于的分布式任務調度組件，非常小巧，使用簡單，只需要引入包。單個任務節點故障時自動轉移到其他任務節點繼續執行。和都是一整套的定時任務框架，沒有必要強行將集成進來，專注做的分布式以及動態任務的封裝。是之后自主開發的定時任務工具。

基于Spring Task + Zookeeper的分布式任務調度組件，非常小巧，使用簡單，只需要引入jar包。不需要多帶帶部署服務端。確保所有任務在集群中不重復，不遺漏的執行。支持動態添加和刪除任務。

GitHub地址

基于zookeeper+spring task的分布任務調度系統。

確保每個任務在集群中不同節點上不重復的執行。

單個任務節點故障時自動轉移到其他任務節點繼續執行。

任務節點啟動時必須保證zookeeper可用，任務節點運行期zookeeper集群不可用時任務節點保持可用前狀態運行，zookeeper集群恢復正常運行。

支持動態添加、修改和刪除任務，支持任務暫停和重新啟動。

添加ip黑名單，過濾不需要執行任務的節點。

后臺管理和任務執行監控。

支持spring-boot,支持單個任務運行多個實例（使用擴展后綴）。

刪除了quartz的集成。Spring Task和quartz都是一整套的定時任務框架，沒有必要強行將quartz集成進來，專注做Spring Task的分布式以及動態任務的封裝。刪除quartz后，組件更加輕便。且所有功能依舊保留。

對于Spring Boot的支持更加智能化。通過spring.factories的方式自動加載配置類UncodeScheduleAutoConfiguration。只需要引入jar包依賴，無須顯示的添加配置類掃描。

參照Alibaba代碼規范對代碼進行了大量重構優化，更具有可讀性。

刪除了默認1s的心跳機制（主要作用：刷新server、重新分配任務、檢查當前serve可執行的任務），采用watcher的方式，對server節點和task節點進行動態監聽，進一步提升性能。

對于非動態添加的任務，也就是注解或配置文件配置的任務會在容器啟動通過組件定義的方式啟動。但是在刪除此類任務時，沒有真正的刪除，taskWrapper任然會定時的執行。 解決了這個bug。

關于UncodeScheduleAutoConfiguration中SchedulerTaskManager的定義。將SchedulerTaskManager的Bean名稱定義為taskScheduler，這樣可以阻止Spring Task初始化名為taskScheduler的bean，以免重復加載。當然你也可以不這么做，因為SchedulerTaskManager繼承了ThreadPoolTaskScheduler，我們動態添加的任務都是通過SchedulerTaskManager添加的。

說明：

單節點故障時需要業務保障數據完整性或冪等性。

Spring Task是Spring 3.0之后自主開發的定時任務工具。

Spring Task默認不是并行執行，需要添加一個名為taskScheduler的Bean，采用ThreadPoolTaskScheduler或其他線程池的Scheduler實現。Spring Task默認采用ThreadPoolTaskScheduler

所有的任務都是基于Spring Bean的方式。可以通過定義一個或多個任務模板（Bean 的方式），通過使用任務后綴可以動態的添加多個該模板的任務實例，你只需要傳遞不同的參數即可。

@Component    
public class SimpleTask {

    private static int i = 0;
    
    @Scheduled(fixedDelay = 5000)
    public void print() {
        System.out.println("===========start!=========");
        System.out.println("I:"+i);i++;
        System.out.println("=========== end !=========");
    }
    
    @Scheduled(cron = "0/5 * * * * ?")
    public void print1() {
        System.out.println("===========start!=========");
        System.out.println("I:"+i);i++;
        System.out.println("=========== end !=========");
    }
    
    @Scheduled(fixedRate = 3000)
    public void print3() {
        System.out.println("===========start!=========");
        System.out.println("I:"+i);i++;
        System.out.println("=========== end !=========");
    }
}
    

// 定義任務實體
TaskDefine task = new TaskDefine();
task.setTargetBean(SchedulerTaskForward.BEAN_NAME);
task.setTargetMethod(SchedulerTaskForward.METHOD);
task.setExtKeySuffix(SUFFIX + model.getId());
task.setCronExpression(model.getCronExpression());
task.setParams(gson.toJson(model));
if (ConsoleManager.isExistsTask(task)) {
    // 更新
    ConsoleManager.updateScheduleTask(task);
} else {
    // 新增
    task.setStartTime(new Date());
    ConsoleManager.addScheduleTask(task);
}

SchedulerTaskForward是我預先定義好的任務模板，下面是代碼片段

@SuppressWarnings("unchecked")
@Component("schedulerTaskForward")
public class SchedulerTaskForward {
    public static final String BEAN_NAME = "schedulerTaskForward";
    public static final String METHOD = "forward";
    private Gson gson = new GsonBuilder().setDateFormat("yyyy-MM-dd HH:mm:ss").create();
    private static final Logger LOGGER = LoggerFactory.getLogger(SchedulerTaskForward.class);

    @Autowired
    private EpmManagerTerminalFeign terminalFeign;

    @Autowired
    private AmqpClientService mqService;

    @Value("${collectd.batch}")
    protected int batch;

    public void forward(String json) {
        SchedulerTaskModel model = gson.fromJson(json, SchedulerTaskModel.class);
        if (ExecuteType.report.getCode().equals(Integer.valueOf(model.getExecuteType()))) {
            // 上報的在定時任務中不執行
            throw new SchedulerTaskExecuteException("任務: " + model.getTaskName() + ", 屬于上報任務, 禁止在定時任務中執行");
        }
        executeCollect(TaskType.forCode(model.getTaskType()), model.getAfns(), model.getOrgId(), model.getTermType());
        LOGGER.debug("任務: " + model.getTaskName() + ", 執行完成");
    }

application.yml

uncode:
  schedule:
    zkConnect: ${spring.cloud.zookeeper.connectString}
    rootPath: /uncode/schedule
    zkSessionTimeout: 60000
    zkUsername: admin
    zkPassword: admin
    poolSize: 10
#    ipBlackList[0]: 127.0.0.2 #server黑名單可選
#    ipBlackList[1]: 127.0.0.3 #server黑名單可選

2 啟動類

@SpringBootApplication
@EnableScheduling
// 這個也是可選的,如果你不需要默認的任務管理界面的話
// 強烈建議自己去實現這個任務管理功能
@ServletComponentScan("cn.uncode.schedule")
public class UncodeScheduleApplication {
    public static void main(String[] agrs){
        SpringApplication.run(UncodeScheduleApplication.class,agrs);
    }
}

1 動態添加任務

ConsoleManager.addScheduleTask(TaskDefine taskDefine);

2 動態刪除任務

ConsoleManager.delScheduleTask(TaskDefine taskDefine);

3 動態更新任務

ConsoleManager.updateScheduleTask(TaskDefine taskDefine);

4 查詢任務列表

ConsoleManager.queryScheduleTask();

通過獲得我們定義的SchedulerTaskManager這個bean,依然可以動態的添加任務。這里就不展示了。

作者：劉惠濤
轉載請注明出處
2017-10-23