Java中怎么利用Redis 實現一個分布式任務調度器

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

代碼實例

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

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)
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<String, String> getAllTriggers();
  public void saveAllTriggers(long version, Map<String, String> triggers);
  public boolean grabTask(String name);
}

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