折騰了一周的 Java Quartz 集群任務調度，很遺憾沒能搞定，網上的相關文章也少得可憐，在多節點(多進程)環境下 Quartz 似乎無法動態增減任務，惱火。無奈之下自己擼了一個簡單的任務調度器，結果只花了不到 2天時間，而且感覺非常簡單好用，代碼量也不多，擴展性很好。

實現一個分布式的任務調度器有幾個關鍵的考慮點單次任務和循環任務好做，難的是 cron 表達式的解析和時間計算怎么做？

多進程同一時間如何保證一個任務的互斥性？

如何動態變更增加和減少任務？

代碼實例

在深入講解實現方法之前，我們先來看看這個調度器是如何使用的class?Demo{

public?static?void?main(String[]?args){

var?redis?=?new?RedisStore();

//?sample?為任務分組名稱

var?store?=?new?RedisTaskStore(redis,?"sample");

//?5s?為任務鎖壽命

var?scheduler?=?new?DistributedScheduler(store,?5);

//?注冊一個單次任務

scheduler.register(Trigger.onceOfDelay(5),?Task.of("once1",?()?->?{

System.out.println("once1");

}));

//?注冊一個循環任務

scheduler.register(Trigger.periodOfDelay(5,?5),?Task.of("period2",?()?->?{

System.out.println("period2");

}));

//?注冊一個?CRON?任務

scheduler.register(Trigger.cronOfMinutes(1),?Task.of("cron3",?()?->?{

System.out.println("cron3");

}));

//?設置全局版本號

scheduler.version(1);

//?注冊監聽器

scheduler.listener(ctx?->?{

System.out.println(ctx.task().name()?+?"?is?complete");

});

//?啟動調度器

scheduler.start();

}

}

當代碼升級任務需要增加減少時(或者變更調度時間)，只需要遞增全局版本號，現有的進程中的任務會自動被重新調度，那些沒有被注冊的任務(任務減少)會自動清除。新增的任務(新任務)在老代碼的進程里是不會被調度的(沒有新任務的代碼無法調度)，被清除的任務(老任務)在老代碼的進程里會被取消調度。

比如我們要取消 period2 任務，增加 period4 任務class?Demo{

public?static?void?main(String[]?args){

var?redis?=?new?RedisStore();

//?sample?為任務分組名稱

var?store?=?new?RedisTaskStore(redis,?"sample");

//?5s?為任務鎖壽命

var?scheduler?=?new?DistributedScheduler(store,?5);

//?注冊一個單次任務

scheduler.register(Trigger.onceOfDelay(5),?Task.of("once1",?()?->?{

System.out.println("once1");

}));

//?注冊一個?CRON?任務

scheduler.register(Trigger.cronOfMinutes(1),?Task.of("cron3",?()?->?{

System.out.println("cron3");

}));

//?注冊一個循環任務

scheduler.register(Trigger.periodOfDelay(5,?10),?Task.of("period4",?()?->?{

System.out.println("period4");

}));

//?遞增全局版本號

scheduler.version(2);

//?注冊監聽器

scheduler.listener(ctx?->?{

System.out.println(ctx.task().name()?+?"?is?complete");

});

//?啟動調度器

scheduler.start();

}

}

cron4j

it.sauronsoftware.cron4j

cron4j

2.2.5

這個開源的 library 包含了基礎的 cron 表達式解析功能，它還提供了任務的調度功能，不過這里并不需要使用它的調度器。我只會用到它的表達式解析功能，以及一個簡單的方法用來判斷當前的時間是否匹配表達式(是否該運行任務了)。

我們對 cron 的時間精度要求很低，1 分鐘判斷一次當前的時間是否到了該運行任務的時候就可以了。class?SchedulingPattern{

//?表達式是否有效

boolean?validate(String?cronExpr);

//?是否應該運行任務了(一分鐘判斷一次)

boolean?match(long?nowTs);

}

任務的互斥性

因為是分布式任務調度器，多進程環境下要控制同一個任務在調度的時間點只能有一個進程運行。使用 Redis 分布式鎖很容易就可以搞定。鎖需要保持一定的時間(比如默認 5s)。

所有的進程都會在同一時間調度這個任務，但是只有一個進程可以搶到鎖。因為分布式環境下時間的不一致性，不同機器上的進程會有較小的時間差異窗口，鎖必須保持一個窗口時間，這里我默認設置為 5s(可定制)，這就要求不同機器的時間差不能超過 5s，超出了這個值就會出現重復調度。public?boolean?grabTask(String?name){

var?holder?=?new?Holder();

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();

}

全局版本號

我們給任務列表附上一個全局的版本號，當業務上需要增加或者減少調度任務時，通過變更版本號來觸發進程的任務重加載。這個重加載的過程包含輪詢全局版本號(Redis 的一個key)，如果發現版本號變動，立即重新加載任務列表配置并重新調度所有的任務。private?void?scheduleReload()?{

//?1s?對比一次

this.scheduler.scheduleWithFixedDelay(()?->?{

try?{

if?(this.reloadIfChanged())?{

this.rescheduleTasks();

}

}?catch?(Exception?e)?{

LOG.error("reloading?tasks?error",?e);

}

},?0,?1,?TimeUnit.SECONDS);

}

重新調度任務先要取消當前所有正在調度的任務，然后調度剛剛加載的所有任務。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();

}

因為需要將任務持久化，所以設計了一套任務的序列化格式，這個也很簡單，使用文本符號分割任務配置屬性就行。//?一次性任務(startTime)

ONCE@2019-04-29T15:26:29.946+0800

//?循環任務，(startTime,endTime,period)，這里任務的結束時間是天荒地老

PERIOD@2019-04-29T15:26:29.949+0800|292278994-08-17T15:12:55.807+0800|5

//?cron?任務，一分鐘一次

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"

線程池

時間調度會有一個單獨的線程(單線程線程池)，任務的運行由另外一個線程池來完成(數量可定制)。class?DistributedScheduler{

private?ScheduledExecutorService?scheduler?=?Executors.newSingleThreadScheduledExecutor();

private?ExecutorService?executor?=?Executors.newFixedThreadPool(threads);

}

之所以要將線程池分開，是為了避免任務的執行(IO)影響了時間的精確調度。

支持無互斥任務

互斥任務要求任務的單進程運行，無互斥任務就是沒有加分布式鎖的任務，可以多進程同時運行。默認需要互斥。class?Task{

/**

*?是否需要考慮多進程互斥(true表示不互斥，多進程能同時跑)

*/

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);

}

}

增加回調接口

考慮到調度器的使用者可能需要對任務運行狀態進行監控，這里增加了一個簡單的回調接口，目前功能比較簡單。能匯報運行結果(成功還是異常)和運行的耗時class?TaskContext{

private?Task?task;

private?long?cost;??//?運行時間

private?boolean?ok;

private?Throwable?e;

}

interface?ISchedulerListener{

public?void?onComplete(TaskContext?ctx);

}

支持存儲擴展

目前只實現了 Redis 和 Memory 形式的任務存儲，擴展到 zk、etcd、關系數據庫也是可行的，實現下面的接口即可。interface?ITaskStore{

public?long?getRemoteVersion();

public?Map?getAllTriggers();

public?void?saveAllTriggers(long?version,?Map?triggers);

public?boolean?grabTask(String?name);

}

代碼地址

https://github.com/pyloque/taskin

總結

以上是生活随笔為你收集整理的java 分布式任务_一个简单的基于 Redis 的分布式任务调度器 —— Java 语言实现...的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。