Java中怎么利用Redis 實現(xiàn)一個分布式任務調(diào)度器

Java中怎么利用Redis 實現(xiàn)一個分布式任務調(diào)度器，針對這個問題，這篇文章詳細介紹了相對應的分析和解答，希望可以幫助更多想解決這個問題的小伙伴找到更簡單易行的方法。

代碼實例

在深入講解實現(xiàn)方法之前，我們先來看看這個調(diào)度器是如何使用的

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");
        }));
        // 注冊一個循環(huán)任務
        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);
        // 注冊監(jiān)聽器
        scheduler.listener(ctx -> {
            System.out.println(ctx.task().name() + " is complete");
        });
        // 啟動調(diào)度器
        scheduler.start();
    }
}

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

比如我們要取消 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");
        }));
        // 注冊一個循環(huán)任務
        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

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

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

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

class SchedulingPattern {
    // 表達式是否有效
    boolean validate(String cronExpr);
    // 是否應該運行任務了(一分鐘判斷一次)
    boolean match(long nowTs);
}

任務的互斥性

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

所有的進程都會在同一時間調(diào)度這個任務，但是只有一個進程可以搶到鎖。因為分布式環(huán)境下時間的不一致性，不同機器上的進程會有較小的時間差異窗口，鎖必須保持一個窗口時間，這里我默認設置為 5s（可定制），這就要求不同機器的時間差不能超過 5s，超出了這個值就會出現(xiàn)重復調(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();
}

全局版本號

我們給任務列表附上一個全局的版本號，當業(yè)務上需要增加或者減少調(diào)度任務時，通過變更版本號來觸發(fā)進程的任務重加載。這個重加載的過程包含輪詢?nèi)职姹咎枺≧edis 的一個key），如果發(fā)現(xiàn)版本號變動，立即重新加載任務列表配置并重新調(diào)度所有的任務。

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

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
// 循環(huán)任務，(startTime,endTime,period)，這里任務的結(jié)束時間是天荒地老
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"

線程池

時間調(diào)度會有一個單獨的線程（單線程線程池），任務的運行由另外一個線程池來完成（數(shù)量可定制）。

class DistributedScheduler {
    private ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();
    private ExecutorService executor = Executors.newFixedThreadPool(threads);
}

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

支持無互斥任務

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

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

增加回調(diào)接口

考慮到調(diào)度器的使用者可能需要對任務運行狀態(tài)進行監(jiān)控，這里增加了一個簡單的回調(diào)接口，目前功能比較簡單。能匯報運行結(jié)果（成功還是異常）和運行的耗時

class TaskContext {
    private Task task;
    private long cost;  // 運行時間
    private boolean ok;
    private Throwable e;
}

interface ISchedulerListener {
    public void onComplete(TaskContext ctx);
}

支持存儲擴展

目前只實現(xiàn)了 Redis 和 Memory 形式的任務存儲，擴展到 zk、etcd、關系數(shù)據(jù)庫也是可行的，實現(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);
}

關于Java中怎么利用Redis 實現(xiàn)一個分布式任務調(diào)度器問題的解答就分享到這里了，希望以上內(nèi)容可以對大家有一定的幫助，如果你還有很多疑惑沒有解開，可以關注億速云行業(yè)資訊頻道了解更多相關知識。