java 分布式任务_一个简单的基于 Redis 的分布式任务调度器 —— Java 语言实现...
折騰了一周的 Java Quartz 集群任務(wù)調(diào)度，很遺憾沒(méi)能搞定，網(wǎng)上的相關(guān)文章也少得可憐，在多節(jié)點(diǎn)(多進(jìn)程)環(huán)境下 Quartz 似乎無(wú)法動(dòng)態(tài)增減任務(wù)，惱火。無(wú)奈之下自己擼了一個(gè)簡(jiǎn)單的任務(wù)調(diào)度器，結(jié)果只花了不到 2天時(shí)間，而且感覺(jué)非常簡(jiǎn)單好用，代碼量也不多，擴(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)方法之前，我們先來(lái)看看這個(gè)調(diào)度器是如何使用的class?Demo{
public?static?void?main(String[]?args){
var?redis?=?new?RedisStore();
//?sample?為任務(wù)分組名稱(chēng)
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)聽(tīng)器
scheduler.listener(ctx?->?{
System.out.println(ctx.task().name()?+?"?is?complete");
});
//?啟動(dòng)調(diào)度器
scheduler.start();
}
}
當(dāng)代碼升級(jí)任務(wù)需要增加減少時(shí)(或者變更調(diào)度時(shí)間)，只需要遞增全局版本號(hào)，現(xiàn)有的進(jìn)程中的任務(wù)會(huì)自動(dòng)被重新調(diào)度，那些沒(méi)有被注冊(cè)的任務(wù)(任務(wù)減少)會(huì)自動(dòng)清除。新增的任務(wù)(新任務(wù))在老代碼的進(jìn)程里是不會(huì)被調(diào)度的(沒(méi)有新任務(wù)的代碼無(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ù)分組名稱(chēng)
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)聽(tīng)器
scheduler.listener(ctx?->?{
System.out.println(ctx.task().name()?+?"?is?complete");
});
//?啟動(dòng)調(diào)度器
scheduler.start();
}
}
cron4j
it.sauronsoftware.cron4j
cron4j
2.2.5
這個(gè)開(kāi)源的 library 包含了基礎(chǔ)的 cron 表達(dá)式解析功能，它還提供了任務(wù)的調(diào)度功能，不過(guò)這里并不需要使用它的調(diào)度器。我只會(huì)用到它的表達(dá)式解析功能，以及一個(gè)簡(jiǎn)單的方法用來(lái)判斷當(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);
}
任務(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í)間差不能超過(guò) 5s，超出了這個(gè)值就會(huì)出現(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();
}
全局版本號(hào)
我們給任務(wù)列表附上一個(gè)全局的版本號(hào)，當(dāng)業(yè)務(wù)上需要增加或者減少調(diào)度任務(wù)時(shí)，通過(guò)變更版本號(hào)來(lái)觸發(fā)進(jìn)程的任務(wù)重加載。這個(gè)重加載的過(guò)程包含輪詢(xún)?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);
}
重新調(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è)計(jì)了一套任務(wù)的序列化格式，這個(gè)也很簡(jiǎn)單，使用文本符號(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"
線程池
時(shí)間調(diào)度會(huì)有一個(gè)單獨(dú)的線程(單線程線程池)，任務(wù)的運(yùn)行由另外一個(gè)線程池來(lái)完成(數(shù)量可定制)。class?DistributedScheduler{
private?ScheduledExecutorService?scheduler?=?Executors.newSingleThreadScheduledExecutor();
private?ExecutorService?executor?=?Executors.newFixedThreadPool(threads);
}
之所以要將線程池分開(kāi)，是為了避免任務(wù)的執(zhí)行(IO)影響了時(shí)間的精確調(diào)度。
支持無(wú)互斥任務(wù)
互斥任務(wù)要求任務(wù)的單進(jìn)程運(yùn)行，無(wú)互斥任務(wù)就是沒(méi)有加分布式鎖的任務(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);
}
}
增加回調(diào)接口
考慮到調(diào)度器的使用者可能需要對(duì)任務(wù)運(yùn)行狀態(tài)進(jìn)行監(jiān)控，這里增加了一個(gè)簡(jiǎn)單的回調(diào)接口，目前功能比較簡(jiǎn)單。能匯報(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);
}
支持存儲(chǔ)擴(kuò)展
目前只實(shí)現(xiàn)了 Redis 和 Memory 形式的任務(wù)存儲(chǔ)，擴(kuò)展到 zk、etcd、關(guān)系數(shù)據(jù)庫(kù)也是可行的，實(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 语言实现...的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。
- 上一篇： 有了固态硬盘怎么安装系统安装系统安装 用
- 下一篇： 深入理解Java虚拟机知乎_深入理解Ja