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

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

GitHub地址

基于zookeeper+spring task的分布任務(wù)調(diào)度系統(tǒng)。

確保每個任務(wù)在集群中不同節(jié)點上不重復(fù)的執(zhí)行。

單個任務(wù)節(jié)點故障時自動轉(zhuǎn)移到其他任務(wù)節(jié)點繼續(xù)執(zhí)行。

任務(wù)節(jié)點啟動時必須保證zookeeper可用，任務(wù)節(jié)點運行期zookeeper集群不可用時任務(wù)節(jié)點保持可用前狀態(tài)運行，zookeeper集群恢復(fù)正常運行。

支持動態(tài)添加、修改和刪除任務(wù)，支持任務(wù)暫停和重新啟動。

添加ip黑名單，過濾不需要執(zhí)行任務(wù)的節(jié)點。

后臺管理和任務(wù)執(zhí)行監(jiān)控。

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

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

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

參照Alibaba代碼規(guī)范對代碼進行了大量重構(gòu)優(yōu)化，更具有可讀性。

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

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

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

說明：

單節(jié)點故障時需要業(yè)務(wù)保障數(shù)據(jù)完整性或冪等性。

Spring Task是Spring 3.0之后自主開發(fā)的定時任務(wù)工具。

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

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

@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 !=========");
    }
}
    

// 定義任務(wù)實體
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是我預(yù)先定義好的任務(wù)模板，下面是代碼片段

@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()))) {
            // 上報的在定時任務(wù)中不執(zhí)行
            throw new SchedulerTaskExecuteException("任務(wù): " + model.getTaskName() + ", 屬于上報任務(wù), 禁止在定時任務(wù)中執(zhí)行");
        }
        executeCollect(TaskType.forCode(model.getTaskType()), model.getAfns(), model.getOrgId(), model.getTermType());
        LOGGER.debug("任務(wù): " + model.getTaskName() + ", 執(zhí)行完成");
    }

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
// 這個也是可選的,如果你不需要默認的任務(wù)管理界面的話
// 強烈建議自己去實現(xiàn)這個任務(wù)管理功能
@ServletComponentScan("cn.uncode.schedule")
public class UncodeScheduleApplication {
    public static void main(String[] agrs){
        SpringApplication.run(UncodeScheduleApplication.class,agrs);
    }
}

1 動態(tài)添加任務(wù)

ConsoleManager.addScheduleTask(TaskDefine taskDefine);

2 動態(tài)刪除任務(wù)

ConsoleManager.delScheduleTask(TaskDefine taskDefine);

3 動態(tài)更新任務(wù)

ConsoleManager.updateScheduleTask(TaskDefine taskDefine);

4 查詢?nèi)蝿?wù)列表

ConsoleManager.queryScheduleTask();

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

作者：劉惠濤
轉(zhuǎn)載請注明出處
2017-10-23