為什么要使用分布式調(diào)度器

分布式調(diào)度器主要應(yīng)用于系統(tǒng)中一些任務(wù)定時(shí)調(diào)度處理既们。通常我們?cè)O(shè)計(jì)一個(gè)定時(shí)任務(wù)晚伙，最簡單的就是直接使用@scheduled注解配置好定時(shí)任務(wù)，這樣開發(fā)工作也簡單。但是也許會(huì)有一種情況，如果發(fā)生在生產(chǎn)環(huán)境上串远，需要不重啟就去變更定時(shí)任務(wù)時(shí)間宏多，或者可能由于某些原因我們需要關(guān)閉某個(gè)定時(shí)任務(wù)，那么這時(shí)候就無法做到動(dòng)態(tài)化澡罚。分布式調(diào)度器就能很好的解決這些疑難雜癥伸但。

有的人可能會(huì)問：現(xiàn)在開源的調(diào)度器也有一些很流行的，比如xxl-job留搔，為什么還要自己設(shè)計(jì)一套更胖。其實(shí)我們也不能說開源的設(shè)計(jì)不好，原因是它的功能太完善隔显，如果要用好還要有人專門運(yùn)維處理却妨，功能過于強(qiáng)大，大部分功能都是雞肋括眠，所以自研一套簡單的調(diào)度服務(wù)有些時(shí)候還是很有必要的管呵。

分布式調(diào)度流程

首先分布式調(diào)度器需要依賴數(shù)據(jù)庫配置，主要配置調(diào)度服務(wù)接口和調(diào)度時(shí)間哺窄。通過調(diào)度服務(wù)集群獲取數(shù)據(jù)庫配置，解析完需要進(jìn)行調(diào)度的任務(wù)账锹，由于是job服務(wù)的一個(gè)集群（也可以單機(jī)部署）所以也要考慮到加鎖萌业，防止多個(gè)job服務(wù)同時(shí)對(duì)一個(gè)任務(wù)多次調(diào)度。最終job服務(wù)將解析完的服務(wù)接口奸柬，檢測到觸發(fā)時(shí)間點(diǎn)就對(duì)應(yīng)用服務(wù)接口發(fā)起任務(wù)調(diào)度生年。

分布式調(diào)度細(xì)節(jié)設(shè)計(jì)分析

數(shù)據(jù)庫設(shè)計(jì)

job_info表設(shè)計(jì)：主要記錄一些job任務(wù)的配置，下面分析一下主要字段：

• job_cron：定時(shí)任務(wù)觸發(fā)時(shí)間配置
• config_id：關(guān)聯(lián)調(diào)度任務(wù)服務(wù)接口配置主鍵
• execute_timeout：任務(wù)調(diào)度超時(shí)時(shí)間配置廓奕，防止調(diào)度時(shí)間過長無結(jié)果
• execute_fail_retry_count：如果任務(wù)調(diào)度失敗重試次數(shù)
• job_status：調(diào)度任務(wù)狀態(tài)開關(guān)配置
• trigger_last_time：最后一次調(diào)度時(shí)間
• trigger_next_time：下一次執(zhí)行時(shí)間

job_config表設(shè)計(jì)：主要配置一些任務(wù)對(duì)應(yīng)需要調(diào)度的服務(wù)接口信息抱婉。

• execute_servier：所需調(diào)度的應(yīng)用服務(wù)
• execute_method：調(diào)度應(yīng)用服務(wù)接口
• execute_param：調(diào)度參數(shù)配置

• service_type：服務(wù)類型（GET/POST）

  
  

job服務(wù)調(diào)度流程設(shè)計(jì)

• 讀取配置：首先job服務(wù)需要不斷的讀取數(shù)據(jù)庫配置，從而得知有哪一些任務(wù)需要進(jìn)行調(diào)度桌粉≌艏ǎ可以通過一個(gè)while循環(huán)加上休眠一段時(shí)間不斷讀取配置，下面就用簡短的偽代碼做個(gè)思路分析：

while(true) {
    // PRE_READ_TIME每次刷新時(shí)間間隔
    TimeUnit.MILLISECONDS.sleep(PRE_READ_TIME - System.currentTimeMillis() % 1000);
    // 讀取配置铃肯，給定一個(gè)時(shí)間患亿，獲取這段時(shí)間內(nèi)要執(zhí)行調(diào)度的任務(wù)以及初次配置的任務(wù)trigger_next_time=0
    List<JobInfo> jobInfos = jobInfoMapper.select(time);
    // 循環(huán)對(duì)任務(wù)一一進(jìn)行解析
    // 1.job應(yīng)用對(duì)獲取到的任務(wù)進(jìn)行加鎖，防止job集群其他服務(wù)同時(shí)調(diào)用押逼，如果確定只會(huì)有單機(jī)部署可不加鎖
    int resultCount = jobInfoMapper.updateByOptimisticLock(jobInfo);
    // 2.加鎖成功繼續(xù)執(zhí)行下一步步藕，對(duì)首次配置的任務(wù)(trigger_next_time=0)需要獲取job_cron進(jìn)行解析，計(jì)算出真實(shí)的下次執(zhí)行時(shí)間trigger_next_time
     refreshNextValidTime(jobInfo, new Date(nowTime));
    // 3.即將要執(zhí)行的任務(wù)加入隊(duì)列
   checkHighFrequency(jobInfo, nowTime);
}

private void checkHighFrequency(JobInfo jobInfo, Long nowTime) throws ParseException {
    // PRE_READ_TIME = 5000挑格，即提前預(yù)留5秒咙冗，將任務(wù)加入隊(duì)列
    if (jobInfo.getTriggerNextTime() < (nowTime + PRE_READ_TIME)) {
        // 將任務(wù)放入待執(zhí)行隊(duì)列
        triggerPoolHelper.triggerJob(jobInfo, jobInfo.getTriggerNextTime() - nowTime);
        // 任務(wù)加入隊(duì)列后，再次更新計(jì)算下次調(diào)度job的時(shí)間
        refreshNextValidTime(jobInfo, new Date(jobInfo.getTriggerNextTime()));
        // 判斷是否是超高頻繁任務(wù)漂彤，即調(diào)度周期小于5s一次
        checkHighFrequency(jobInfo, nowTime);
    }
}
// 計(jì)算下次執(zhí)行時(shí)間
private void refreshNextValidTime(JobInfo jobInfo, Date fromTime) throws ParseException {
    // 時(shí)間表達(dá)式轉(zhuǎn)換計(jì)算下次觸發(fā)時(shí)間
    Date nextValidTime = new CronExpression(jobInfo.getJobCron()).getNextValidTimeAfter(fromTime);
    if (nextValidTime != null) {
        jobInfo.setTriggerLastTime(jobInfo.getTriggerNextTime());
        jobInfo.setTriggerNextTime(nextValidTime.getTime());
    }
}

• 線程池隊(duì)列執(zhí)行任務(wù)調(diào)度

public void triggerJob(JobInfo jobInfo, long delay) {
    JobInfo copyOf = new JobInfo();
    BeanUtils.copyProperties(jobInfo, copyOf);
    JobTriggerThread triggerThread = new JobTriggerThread(copyOf, tinyJobExecutor.get(jobInfo.getJobType()));
    // 小于0說明是延期的任務(wù)雾消，立即執(zhí)行
    if (delay <= 0) {
        // 加入線程池
        triggerPool.execute(triggerThread);
    }
    // 大于0說明還未到調(diào)度時(shí)間,延遲調(diào)度
    else {
        triggerPool.schedule(triggerThread, delay, TimeUnit.MILLISECONDS);
    }
}

• 任務(wù)調(diào)度流程：SpringCloud服務(wù)使用DiscoveryClient灾搏，根據(jù)job_config表配置的服務(wù)名獲取集群服務(wù)列表，再根據(jù)隨機(jī)（或自定義算法）計(jì)算獲取一個(gè)服務(wù)實(shí)例仪或，用該實(shí)例創(chuàng)建請(qǐng)求并發(fā)起服務(wù)接口調(diào)用确镊，最終再根據(jù)調(diào)用結(jié)果進(jìn)行日志記錄，以及失敗后續(xù)是否進(jìn)行重試處理范删。

List<ServiceInstance> serviceInstanceList = discoveryClient.getInstances(jobConfig.getExecuteService());
// 隨機(jī)獲取服務(wù)列表（可自定義算法）
ServiceInstance serviceInstance = getRandomInstrance(serviceInstanceList);
CloseableHttpClient httpClient = HttpClients.createDefault();
// 創(chuàng)建請(qǐng)求
HttpPost httpPost = new HttpPost(serviceInstance.getUri() + "/" + jobConfig.getExecuteMethod() + "?" + jobConfig.getExecuteParam());
// http發(fā)起調(diào)用
CloseableHttpResponse response = httpClient.execute(httpPost);

案例配置說明

• 添加兩個(gè)定時(shí)任務(wù)配置蕾域，此配置如有需要也可開發(fā)個(gè)簡單的頁面方便配置添加與更改。

• 定時(shí)任務(wù)對(duì)應(yīng)的調(diào)度服務(wù)接口配置

根據(jù)以上的配置到旦，定時(shí)刷新獲取任務(wù)列表旨巷，任務(wù)首次配置trigger_next_time=0，需解析成具體執(zhí)行時(shí)間點(diǎn)添忘，任務(wù)調(diào)度判斷該時(shí)間點(diǎn)是否達(dá)到可執(zhí)行時(shí)間采呐，在達(dá)到指定時(shí)間點(diǎn)job服務(wù)將對(duì)該接口發(fā)起調(diào)用并記錄調(diào)度日志。

總結(jié)

使用分布式調(diào)度器能夠很好的管理我們的定時(shí)任務(wù)接口搁骑，開發(fā)人員也只需專注開發(fā)業(yè)務(wù)接口斧吐，讓業(yè)務(wù)與配置完全分離。定時(shí)配置還可以根據(jù)業(yè)務(wù)場景統(tǒng)一進(jìn)行時(shí)間協(xié)調(diào)管理仲器，以免在有些時(shí)間點(diǎn)多任務(wù)同時(shí)處理煤率，可以將時(shí)間配置的分散點(diǎn)以減輕CPU的壓力。如果系統(tǒng)業(yè)務(wù)量少乏冀，定時(shí)任務(wù)也不多的情況也沒必要多浪費(fèi)時(shí)間開發(fā)一個(gè)調(diào)度系統(tǒng)蝶糯。