折騰了一周的 Java Quartz 集群任務(wù)調(diào)度，很遺憾沒能搞定拌滋，網(wǎng)上的相關(guān)文章也少得可憐，在多節(jié)點(diǎn)（多進(jìn)程）環(huán)境下 Quartz 似乎無法動(dòng)態(tài)增減任務(wù)，惱火读第。無奈之下自己擼了一個(gè)簡單的任務(wù)調(diào)度器，結(jié)果只花了不到 2天時(shí)間稻扬，而且感覺非常簡單好用卦方，代碼量也不多，擴(kuò)展性很好泰佳。
實(shí)現(xiàn)一個(gè)分布式的任務(wù)調(diào)度器有幾個(gè)關(guān)鍵的考慮點(diǎn)
- 單次任務(wù)和循環(huán)任務(wù)好做盼砍，難的是 cron 表達(dá)式的解析和時(shí)間計(jì)算怎么做？
- 多進(jìn)程同一時(shí)間如何保證一個(gè)任務(wù)的互斥性逝她？
- 如何動(dòng)態(tài)變更增加和減少任務(wù)浇坐？
代碼實(shí)例
在深入講解實(shí)現(xiàn)方法之前，我們先來看看這個(gè)調(diào)度器是如何使用的
class Demo {
public static void main(String[] args) {
var redis = new RedisStore();
// sample 為任務(wù)分組名稱
var store = new RedisTaskStore(redis, "sample");
// 5s 為任務(wù)鎖壽命
var scheduler = new DistributedScheduler(store, 5);
// 注冊(cè)一個(gè)單次任務(wù)
scheduler.register(Trigger.onceOfDelay(5), Task.of("once1", () -> {
System.out.println("once1");
}));
// 注冊(cè)一個(gè)循環(huán)任務(wù)
scheduler.register(Trigger.periodOfDelay(5, 5), Task.of("period2", () -> {
System.out.println("period2");
}));
// 注冊(cè)一個(gè) CRON 任務(wù)
scheduler.register(Trigger.cronOfMinutes(1), Task.of("cron3", () -> {
System.out.println("cron3");
}));
// 設(shè)置全局版本號(hào)
scheduler.version(1);
// 注冊(cè)監(jiān)聽器
scheduler.listener(ctx -> {
System.out.println(ctx.task().name() + " is complete");
});
// 啟動(dòng)調(diào)度器
scheduler.start();
}
}
復(fù)制代碼
當(dāng)代碼升級(jí)任務(wù)需要增加減少時(shí)（或者變更調(diào)度時(shí)間）黔宛，只需要遞增全局版本號(hào)近刘，現(xiàn)有的進(jìn)程中的任務(wù)會(huì)自動(dòng)被重新調(diào)度，那些沒有被注冊(cè)的任務(wù)（任務(wù)減少）會(huì)自動(dòng)清除臀晃。新增的任務(wù)（新任務(wù)）在老代碼的進(jìn)程里是不會(huì)被調(diào)度的（沒有新任務(wù)的代碼無法調(diào)度）觉渴，被清除的任務(wù)（老任務(wù)）在老代碼的進(jìn)程里會(huì)被取消調(diào)度。
比如我們要取消 period2 任務(wù)徽惋，增加 period4 任務(wù)
class Demo {
public static void main(String[] args) {
var redis = new RedisStore();
// sample 為任務(wù)分組名稱
var store = new RedisTaskStore(redis, "sample");
// 5s 為任務(wù)鎖壽命
var scheduler = new DistributedScheduler(store, 5);
// 注冊(cè)一個(gè)單次任務(wù)
scheduler.register(Trigger.onceOfDelay(5), Task.of("once1", () -> {
System.out.println("once1");
}));
// 注冊(cè)一個(gè) CRON 任務(wù)
scheduler.register(Trigger.cronOfMinutes(1), Task.of("cron3", () -> {
System.out.println("cron3");
}));
// 注冊(cè)一個(gè)循環(huán)任務(wù)
scheduler.register(Trigger.periodOfDelay(5, 10), Task.of("period4", () -> {
System.out.println("period4");
}));
// 遞增全局版本號(hào)
scheduler.version(2);
// 注冊(cè)監(jiān)聽器
scheduler.listener(ctx -> {
System.out.println(ctx.task().name() + " is complete");
});
// 啟動(dòng)調(diào)度器
scheduler.start();
}
}
復(fù)制代碼
cron4j
<dependency>
<groupId>it.sauronsoftware.cron4j</groupId>
<artifactId>cron4j</artifactId>
<version>2.2.5</version>
</dependency>
復(fù)制代碼
這個(gè)開源的 library 包含了基礎(chǔ)的 cron 表達(dá)式解析功能案淋，它還提供了任務(wù)的調(diào)度功能，不過這里并不需要使用它的調(diào)度器险绘。我只會(huì)用到它的表達(dá)式解析功能踢京，以及一個(gè)簡單的方法用來判斷當(dāng)前的時(shí)間是否匹配表達(dá)式（是否該運(yùn)行任務(wù)了）。
我們對(duì) cron 的時(shí)間精度要求很低宦棺，1 分鐘判斷一次當(dāng)前的時(shí)間是否到了該運(yùn)行任務(wù)的時(shí)候就可以了瓣距。
class SchedulingPattern {
// 表達(dá)式是否有效
boolean validate(String cronExpr);
// 是否應(yīng)該運(yùn)行任務(wù)了(一分鐘判斷一次)
boolean match(long nowTs);
}
復(fù)制代碼
任務(wù)的互斥性
因?yàn)槭欠植际饺蝿?wù)調(diào)度器，多進(jìn)程環(huán)境下要控制同一個(gè)任務(wù)在調(diào)度的時(shí)間點(diǎn)只能有一個(gè)進(jìn)程運(yùn)行代咸。使用 Redis 分布式鎖很容易就可以搞定蹈丸。鎖需要保持一定的時(shí)間（比如默認(rèn) 5s）。
所有的進(jìn)程都會(huì)在同一時(shí)間調(diào)度這個(gè)任務(wù)侣背，但是只有一個(gè)進(jìn)程可以搶到鎖白华。因?yàn)榉植际江h(huán)境下時(shí)間的不一致性，不同機(jī)器上的進(jìn)程會(huì)有較小的時(shí)間差異窗口贩耐，鎖必須保持一個(gè)窗口時(shí)間，這里我默認(rèn)設(shè)置為 5s（可定制）厦取，這就要求不同機(jī)器的時(shí)間差不能超過 5s潮太，超出了這個(gè)值就會(huì)出現(xiàn)重復(fù)調(diào)度。
public boolean grabTask(String name) {
var holder = new Holder<Boolean>();
redis.execute(jedis -> {
var lockKey = keyFor("task_lock", name);
var ok = jedis.set(lockKey, "true", SetParams.setParams().nx().ex(lockAge));
holder.value(ok != null);
});
return holder.value();
}
復(fù)制代碼
全局版本號(hào)
我們給任務(wù)列表附上一個(gè)全局的版本號(hào)，當(dāng)業(yè)務(wù)上需要增加或者減少調(diào)度任務(wù)時(shí)铡买，通過變更版本號(hào)來觸發(fā)進(jìn)程的任務(wù)重加載更鲁。這個(gè)重加載的過程包含輪詢?nèi)职姹咎?hào)（Redis 的一個(gè)key），如果發(fā)現(xiàn)版本號(hào)變動(dòng)奇钞，立即重新加載任務(wù)列表配置并重新調(diào)度所有的任務(wù)澡为。
private void scheduleReload() {
// 1s 對(duì)比一次
this.scheduler.scheduleWithFixedDelay(() -> {
try {
if (this.reloadIfChanged()) {
this.rescheduleTasks();
}
} catch (Exception e) {
LOG.error("reloading tasks error", e);
}
}, 0, 1, TimeUnit.SECONDS);
}
復(fù)制代碼
重新調(diào)度任務(wù)先要取消當(dāng)前所有正在調(diào)度的任務(wù)，然后調(diào)度剛剛加載的所有任務(wù)景埃。
private void rescheduleTasks() {
this.cancelAllTasks();
this.scheduleTasks();
}
private void cancelAllTasks() {
this.futures.forEach((name, future) -> {
LOG.warn("cancelling task {}", name);
future.cancel(false);
});
this.futures.clear();
}
復(fù)制代碼
因?yàn)樾枰獙⑷蝿?wù)持久化媒至，所以設(shè)計(jì)了一套任務(wù)的序列化格式，這個(gè)也很簡單谷徙，使用文本符號(hào)分割任務(wù)配置屬性就行拒啰。
// 一次性任務(wù)(startTime)
ONCE@2019-04-29T15:26:29.946+0800
// 循環(huán)任務(wù)，(startTime,endTime,period)完慧，這里任務(wù)的結(jié)束時(shí)間是天荒地老
PERIOD@2019-04-29T15:26:29.949+0800|292278994-08-17T15:12:55.807+0800|5
// cron 任務(wù)谋旦，一分鐘一次
CRON@*/1 * * * *
$ redis-cli
127.0.0.1:6379> hgetall sample_triggers
1) "task3"
2) "CRON@*/1 * * * *"
3) "task2"
4) "PERIOD@2019-04-29T15:26:29.949+0800|292278994-08-17T15:12:55.807+0800|5"
5) "task1"
6) "ONCE@2019-04-29T15:26:29.946+0800"
7) "task4"
8) "PERIOD@2019-04-29T15:26:29.957+0800|292278994-08-17T15:12:55.807+0800|10"
復(fù)制代碼
線程池
時(shí)間調(diào)度會(huì)有一個(gè)單獨(dú)的線程（單線程線程池），任務(wù)的運(yùn)行由另外一個(gè)線程池來完成（數(shù)量可定制）屈尼。
class DistributedScheduler {
private ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();
private ExecutorService executor = Executors.newFixedThreadPool(threads);
}
復(fù)制代碼
之所以要將線程池分開册着，是為了避免任務(wù)的執(zhí)行（IO）影響了時(shí)間的精確調(diào)度。
FixedDelay vs FixedRate
Java 的內(nèi)置調(diào)度器提供兩種調(diào)度策略 FixedDelay 和 FixedRate脾歧。FixedDelay 保證同一個(gè)任務(wù)的連續(xù)兩次運(yùn)行有相等的時(shí)延（nextRun.startTime - lastRun.endTime）甲捏，F(xiàn)ixedRate 保證同一個(gè)任務(wù)的連續(xù)運(yùn)行有確定的間隔（nextRun.startTime - lastRun.startTime）。
[圖片上傳中...(image-21b05b-1558600573416-56)]
<figcaption style="display: block; text-align: center; font-size: 1rem; line-height: 1.6; color: rgb(144, 144, 144); margin-top: 2px;"></figcaption>
FixedDelay 就好比你加班到深夜12點(diǎn)涨椒，可以第二天12點(diǎn)再來上班(保證固定的休息時(shí)間)摊鸡，而 FixedRate 就沒那么體貼了，第二天你繼續(xù) 9點(diǎn)過來上班蚕冬。如果你不走運(yùn)到第二天 9 點(diǎn)了還在加班免猾，那你今天就沒有休息時(shí)間了，繼續(xù)上班吧囤热。
class ScheduledExecutorService {
void scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit);
void scheduleAtFixedDelay(Runnable command, long initialDelay, long delay, TimeUnit unit);
}
復(fù)制代碼
分布式調(diào)度器要求有精確的調(diào)度時(shí)間猎提，所以必須采用 FixedRate 模式，保證多節(jié)點(diǎn)同一個(gè)任務(wù)在同一時(shí)間被爭搶旁蔼。如果采用 FixedDelay 模式锨苏，會(huì)導(dǎo)致不同進(jìn)程的調(diào)度時(shí)間錯(cuò)開了，分布式鎖的默認(rèn) 5s 時(shí)間窗口將起不到互斥作用棺聊。
支持無互斥任務(wù)
互斥任務(wù)要求任務(wù)的單進(jìn)程運(yùn)行伞租，無互斥任務(wù)就是沒有加分布式鎖的任務(wù)，可以多進(jìn)程同時(shí)運(yùn)行限佩。默認(rèn)需要互斥葵诈。
class Task {
/**
* 是否需要考慮多進(jìn)程互斥（true表示不互斥裸弦，多進(jìn)程能同時(shí)跑）
*/
private boolean concurrent;
private String name;
private Runnable runner;
...
public static Task of(String name, Runnable runner) {
return new Task(name, false, runner);
}
public static Task concurrent(String name, Runnable runner) {
return new Task(name, true, runner);
}
}
復(fù)制代碼
增加回調(diào)接口
考慮到調(diào)度器的使用者可能需要對(duì)任務(wù)運(yùn)行狀態(tài)進(jìn)行監(jiān)控，這里增加了一個(gè)簡單的回調(diào)接口作喘，目前功能比較簡單理疙。能匯報(bào)運(yùn)行結(jié)果（成功還是異常）和運(yùn)行的耗時(shí)
class TaskContext {
private Task task;
private long cost; // 運(yùn)行時(shí)間
private boolean ok;
private Throwable e;
}
interface ISchedulerListener {
public void onComplete(TaskContext ctx);
}
復(fù)制代碼
支持存儲(chǔ)擴(kuò)展
目前只實(shí)現(xiàn)了 Redis 和 Memory 形式的任務(wù)存儲(chǔ)，擴(kuò)展到 zk泞坦、etcd窖贤、關(guān)系數(shù)據(jù)庫也是可行的，實(shí)現(xiàn)下面的接口即可贰锁。
interface ITaskStore {
public long getRemoteVersion();
public Map<String, String> getAllTriggers();
public void saveAllTriggers(long version, Map<String, String> triggers);
public boolean grabTask(String name);
}
復(fù)制代碼
