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

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

GitHub地址

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

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

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

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

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

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

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

支持spring-boot,支持單個(gè)任務(wù)運(yùn)行多個(gè)實(shí)例（使用擴(kuò)展后綴）。

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

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

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

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

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

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

說(shuō)明：

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

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

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

所有的任務(wù)都是基于Spring Bean的方式。可以通過(guò)定義一個(gè)或多個(gè)任務(wù)模板（Bean 的方式），通過(guò)使用任務(wù)后綴可以動(dòng)態(tài)的添加多個(gè)該模板的任務(wù)實(shí)例，你只需要傳遞不同的參數(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ù)實(shí)體
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()))) {
            // 上報(bào)的在定時(shí)任務(wù)中不執(zhí)行
            throw new SchedulerTaskExecuteException("任務(wù): " + model.getTaskName() + ", 屬于上報(bào)任務(wù), 禁止在定時(shí)任務(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 啟動(dòng)類

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

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

ConsoleManager.addScheduleTask(TaskDefine taskDefine);

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

ConsoleManager.delScheduleTask(TaskDefine taskDefine);

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

ConsoleManager.updateScheduleTask(TaskDefine taskDefine);

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

ConsoleManager.queryScheduleTask();

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

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