java 分布式任务_一个简单的基于 Redis 的分布式任务调度器 —— Java 语言实现...
折騰了一周的 Java Quartz 集群任務(wù)調(diào)度，很遺憾沒能搞定，網(wǎng)上的相關(guān)文章也少得可憐，在多節(jié)點(diǎn)(多進(jìn)程)環(huán)境下 Quartz 似乎無法動態(tài)增減任務(wù)，惱火。無奈之下自己擼了一個簡單的任務(wù)調(diào)度器，結(jié)果只花了不到 2天時間，而且感覺非常簡單好用，代碼量也不多，擴(kuò)展性很好。
實(shí)現(xiàn)一個分布式的任務(wù)調(diào)度器有幾個關(guān)鍵的考慮點(diǎn)單次任務(wù)和循環(huán)任務(wù)好做，難的是 cron 表達(dá)式的解析和時間計算怎么做？
多進(jìn)程同一時間如何保證一個任務(wù)的互斥性？
如何動態(tài)變更增加和減少任務(wù)？
代碼實(shí)例
在深入講解實(shí)現(xiàn)方法之前，我們先來看看這個調(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);
//?注冊一個單次任務(wù)
scheduler.register(Trigger.onceOfDelay(5),?Task.of("once1",?()?->?{
System.out.println("once1");
}));
//?注冊一個循環(huán)任務(wù)
scheduler.register(Trigger.periodOfDelay(5,?5),?Task.of("period2",?()?->?{
System.out.println("period2");
}));
//?注冊一個?CRON?任務(wù)
scheduler.register(Trigger.cronOfMinutes(1),?Task.of("cron3",?()?->?{
System.out.println("cron3");
}));
//?設(shè)置全局版本號
scheduler.version(1);
//?注冊監(jiān)聽器
scheduler.listener(ctx?->?{
System.out.println(ctx.task().name()?+?"?is?complete");
});
//?啟動調(diào)度器
scheduler.start();
}
}
當(dāng)代碼升級任務(wù)需要增加減少時(或者變更調(diào)度時間)，只需要遞增全局版本號，現(xiàn)有的進(jìn)程中的任務(wù)會自動被重新調(diào)度，那些沒有被注冊的任務(wù)(任務(wù)減少)會自動清除。新增的任務(wù)(新任務(wù))在老代碼的進(jìn)程里是不會被調(diào)度的(沒有新任務(wù)的代碼無法調(diào)度)，被清除的任務(wù)(老任務(wù))在老代碼的進(jìn)程里會被取消調(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);
//?注冊一個單次任務(wù)
scheduler.register(Trigger.onceOfDelay(5),?Task.of("once1",?()?->?{
System.out.println("once1");
}));
//?注冊一個?CRON?任務(wù)
scheduler.register(Trigger.cronOfMinutes(1),?Task.of("cron3",?()?->?{
System.out.println("cron3");
}));
//?注冊一個循環(huán)任務(wù)
scheduler.register(Trigger.periodOfDelay(5,?10),?Task.of("period4",?()?->?{
System.out.println("period4");
}));
//?遞增全局版本號
scheduler.version(2);
//?注冊監(jiān)聽器
scheduler.listener(ctx?->?{
System.out.println(ctx.task().name()?+?"?is?complete");
});
//?啟動調(diào)度器
scheduler.start();
}
}
cron4j
it.sauronsoftware.cron4j
cron4j
2.2.5
這個開源的 library 包含了基礎(chǔ)的 cron 表達(dá)式解析功能，它還提供了任務(wù)的調(diào)度功能，不過這里并不需要使用它的調(diào)度器。我只會用到它的表達(dá)式解析功能，以及一個簡單的方法用來判斷當(dāng)前的時間是否匹配表達(dá)式(是否該運(yùn)行任務(wù)了)。
我們對 cron 的時間精度要求很低，1 分鐘判斷一次當(dāng)前的時間是否到了該運(yùn)行任務(wù)的時候就可以了。class?SchedulingPattern{
//?表達(dá)式是否有效
boolean?validate(String?cronExpr);
//?是否應(yīng)該運(yùn)行任務(wù)了(一分鐘判斷一次)
boolean?match(long?nowTs);
}
任務(wù)的互斥性
因?yàn)槭欠植际饺蝿?wù)調(diào)度器，多進(jìn)程環(huán)境下要控制同一個任務(wù)在調(diào)度的時間點(diǎn)只能有一個進(jìn)程運(yùn)行。使用 Redis 分布式鎖很容易就可以搞定。鎖需要保持一定的時間(比如默認(rèn) 5s)。
所有的進(jìn)程都會在同一時間調(diào)度這個任務(wù)，但是只有一個進(jìn)程可以搶到鎖。因?yàn)榉植际江h(huán)境下時間的不一致性，不同機(jī)器上的進(jìn)程會有較小的時間差異窗口，鎖必須保持一個窗口時間，這里我默認(rèn)設(shè)置為 5s(可定制)，這就要求不同機(jī)器的時間差不能超過 5s，超出了這個值就會出現(xiàn)重復(fù)調(diào)度。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();
}
全局版本號
我們給任務(wù)列表附上一個全局的版本號，當(dāng)業(yè)務(wù)上需要增加或者減少調(diào)度任務(wù)時，通過變更版本號來觸發(fā)進(jìn)程的任務(wù)重加載。這個重加載的過程包含輪詢?nèi)职姹咎?Redis 的一個key)，如果發(fā)現(xiàn)版本號變動，立即重新加載任務(wù)列表配置并重新調(diào)度所有的任務(wù)。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);
}
重新調(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();
}
因?yàn)樾枰獙⑷蝿?wù)持久化，所以設(shè)計了一套任務(wù)的序列化格式，這個也很簡單，使用文本符號分割任務(wù)配置屬性就行。//?一次性任務(wù)(startTime)
ONCE@2019-04-29T15:26:29.946+0800
//?循環(huán)任務(wù)，(startTime,endTime,period)，這里任務(wù)的結(jié)束時間是天荒地老
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"
線程池
時間調(diào)度會有一個單獨(dú)的線程(單線程線程池)，任務(wù)的運(yùn)行由另外一個線程池來完成(數(shù)量可定制)。class?DistributedScheduler{
private?ScheduledExecutorService?scheduler?=?Executors.newSingleThreadScheduledExecutor();
private?ExecutorService?executor?=?Executors.newFixedThreadPool(threads);
}
之所以要將線程池分開，是為了避免任務(wù)的執(zhí)行(IO)影響了時間的精確調(diào)度。
支持無互斥任務(wù)
互斥任務(wù)要求任務(wù)的單進(jìn)程運(yùn)行，無互斥任務(wù)就是沒有加分布式鎖的任務(wù)，可以多進(jìn)程同時運(yùn)行。默認(rèn)需要互斥。class?Task{
/**
*?是否需要考慮多進(jìn)程互斥(true表示不互斥，多進(jìn)程能同時跑)
*/
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);
}
}
增加回調(diào)接口
考慮到調(diào)度器的使用者可能需要對任務(wù)運(yùn)行狀態(tài)進(jìn)行監(jiān)控，這里增加了一個簡單的回調(diào)接口，目前功能比較簡單。能匯報運(yùn)行結(jié)果(成功還是異常)和運(yùn)行的耗時class?TaskContext{
private?Task?task;
private?long?cost;??//?運(yùn)行時間
private?boolean?ok;
private?Throwable?e;
}
interface?ISchedulerListener{
public?void?onComplete(TaskContext?ctx);
}
支持存儲擴(kuò)展
目前只實(shí)現(xiàn)了 Redis 和 Memory 形式的任務(wù)存儲，擴(kuò)展到 zk、etcd、關(guān)系數(shù)據(jù)庫也是可行的，實(shí)現(xiàn)下面的接口即可。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
總結(jié)
以上是生活随笔為你收集整理的java 分布式任务_一个简单的基于 Redis 的分布式任务调度器 —— Java 语言实现...的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： 有了固态硬盘怎么安装系统安装系统安装 用
- 下一篇： 深入理解Java虚拟机知乎_深入理解Ja