開發

一個簡單的基于 Redis 的分布式任務調度器 —— Java 語言實現

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

實現一個分布式的任務調度器有幾個關鍵的考慮點

  1. 單次任務和循環任務好做,難的是 cron 表達式的解析和時間計算怎么做?
  2. 多進程同一時間如何保證一個任務的互斥性?
  3. 如何動態變更增加和減少任務?

代碼實例

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

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

<dependency>????<groupId>it.sauronsoftware.cron4j</groupId>????<artifactId>cron4j</artifactId>????<version>2.2.5</version></dependency>

這個開源的 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<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();}

全局版本號

我們給任務列表附上一個全局的版本號,當業務上需要增加或者減少調度任務時,通過變更版本號來觸發進程的任務重加載。這個重加載的過程包含輪詢全局版本號(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)[email protected]:26:29.946+0800//?循環任務,(startTime,endTime,period),這里任務的結束時間是天荒地老[email protected]:26:29.949+0800|292278994-08-17T15:12:55.807+0800|5//?cron?任務,一分鐘一次[email protected]*/1?*?*?*?*$?redis-cli127.0.0.1:6379>?hgetall?sample_triggers1)?"task3"2)?"[email protected]*/1?*?*?*?*"3)?"task2"4)?"[email protected]:26:29.949+0800|292278994-08-17T15:12:55.807+0800|5"5)?"task1"6)?"[email protected]:26:29.946+0800"7)?"task4"8)?"[email protected]: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<String,?String>?getAllTriggers();??public?void?saveAllTriggers(long?version,?Map<String,?String>?triggers);??public?boolean?grabTask(String?name);}

代碼地址

https://github.com/pyloque/taskin
我還沒有學會寫個人說明!

侵入式服務治理方案,讀這一篇就夠

上一篇

黑客鎖定市政系統勒索比特幣,政府拒付贖金!全美最危險城市陷入癱瘓的第三周……

下一篇

你也可能喜歡

一個簡單的基于 Redis 的分布式任務調度器 —— Java 語言實現

長按儲存圖像,分享給朋友

ITPUB 每周精要將以郵件的形式發放至您的郵箱


微信掃一掃

微信掃一掃
天津11选五开奖结果手 贵州十一选五前三直选和 对平安银行股票分析 江西11选5推荐号 排列走势图五 股票分析软件排行榜 股票分析软件哪个好 快乐12前三直选遗漏号 山西快乐十分停售 河南22选5推茬号 今曰股市行情 什么是股票涨跌幅