DB-Scheduler：Java持久化且支持集群的任务调度器

[LOADING...]

1. 简介

在本教程中，我们将了解 db-scheduler，它是 Quartz 的一个更简单的替代方案。我们将了解它是什么、如何使用它以及我们能用它做什么。

2. 什么是 DB-Scheduler？

DB-Scheduler 是一个支持集群、持久化的调度库。 它被设计为可以轻松集成到我们的应用程序中。我们只需要 Java 17+ 和一个 JDBC 连接，连接到的数据库中有一个用于存储和管理任务的表。

使用 db-scheduler，我们可以轻松支持：

• 一次性任务：计划在特定时间点运行一次的任务
• 简单重复任务：设置为按静态定义的固定时间表运行的任务
• 动态重复任务：按固定时间表运行，但我们可以在运行时动态调度的任务

然后，db-scheduler 库将为我们处理所有事情。这包括确保这些任务在服务集群中正确运行，既不会丢失，也不会重复执行。

3. 设置

在使用 db-scheduler 之前，我们需要先在应用程序中进行设置。这包括将其添加到项目中并让它运行起来。

3.1. 依赖

在使用 db-scheduler 之前，我们需要在构建中包含最新版本，撰写本文时该版本为 16.12.0。

如果我们使用 Maven，可以在 pom.xml 文件中包含此依赖：

<dependency>
    <groupId>com.github.kagkarlsson</groupId>
    <artifactId>db-scheduler</artifactId>
    <version>16.12.0</version>
</dependency>

这将引入我们在应用程序中使用它所需的一切。唯一必需的依赖是 JSpecify 和 SLF4J——其他一切都是自包含的。

3.2. 数据库模式

为了正常工作，db-scheduler 需要在我们连接的数据库中有一个数据库表。 大多数主流 RDBMS 引擎都提供了 DDL 脚本，不过它可能也适用于其他引擎。例如，要与 PostgreSQL 一起使用，我们可以使用如下 DDL：

CREATE TABLE scheduled_tasks (
  task_name TEXT NOT NULL,
  task_instance TEXT NOT NULL,
  task_data BYTEA,
  execution_time TIMESTAMP WITH TIME ZONE NOT NULL,
  picked BOOLEAN NOT NULL,
  picked_by TEXT,
  last_success TIMESTAMP WITH TIME ZONE,
  last_failure TIMESTAMP WITH TIME ZONE,
  consecutive_failures INT,
  last_heartbeat TIMESTAMP WITH TIME ZONE,
  version BIGINT NOT NULL,
  priority SMALLINT,
  PRIMARY KEY (task_name, task_instance)
);
CREATE INDEX execution_time_idx ON scheduled_tasks (execution_time);
CREATE INDEX last_heartbeat_idx ON scheduled_tasks (last_heartbeat);
CREATE INDEX priority_execution_time_idx on scheduled_tasks (priority desc, execution_time asc);

3.3. 调度器

最后，在使用 db-scheduler 之前，我们需要实际设置一个 Scheduler 实例。这是管理我们所有任务调度的核心类。

创建实例时，我们需要提供一个 DataSource 指向包含 scheduled_tasks 表的数据库，以及一个包含所有待管理任务的集合。我们稍后将看到如何创建这些：

Scheduler scheduler = Scheduler.create(dataSource)
  .startTasks(tasks)
  .registerShutdownHook()
  .build();

它附带了一套合理的默认配置值：

• 每 10 秒轮询一次
• 每 5 分钟心跳一次
• 如果任务错过 6 次心跳，则将其视为已死亡
• 在 10 个工作线程上运行任务

以及更多。我们可以在创建调度器时配置所有这些：

Scheduler scheduler = Scheduler.create(dataSource)
  .startTasks(tasks)
  .registerShutdownHook()
  .pollingInterval(Duration.ofSeconds(2))
  .heartbeatInterval(Duration.ofMinutes(2))
  .missedHeartbeatsLimit(10)
  .threads(5)
  .build();

同一集群中的多个实例至少使用相同的心跳设置非常重要。 否则，db-scheduler 可能会错误地将某些任务视为已死亡。

一旦我们有了调度器，就需要启动它运行：

scheduler.start();

此时，调度器将管理分配给它的所有任务。这包括确保它们被正确调度并在适当的时间运行。

4. 简单的重复任务

最容易处理的任务是简单的重复任务。为此，我们需要一个唯一的名称、任务运行的频率以及任务本身：

RecurringTask<Void> task = Tasks.recurring("my-hourly-task", FixedDelay.ofHours(1))
  .execute((instance, context) -> {
    LOG.info("Executed!");
  });

如果这是一个调度器从未见过的新任务，它将被安排立即运行。之后，调度器将根据提供的时间表运行它。在这种情况下，这意味着在上一次迭代完成后 1 小时。我们还可以使用 Daily 在每天特定时间运行，以及使用 CronSchedule 来处理更复杂的 cron 表达式。

我们的任务本身作为 VoidExecutionHandler<T> 的实例提供，它本身是一个函数式接口，如果我们愿意，可以用 lambda 实现。该接口的唯一方法接收两个参数：

• TaskInstance<T> instance —— 关于任务实例的详细信息。如果我们有多个使用相同任务实现的时间表，这会很有用。
• ExecutionContext context —— 关于执行器本身的详细信息。如果我们想从任务内部访问调度器本身，这会很有用。

请注意，这里我们使用的是 VoidExecutionHandler 和 RecurringTask<Void>。这是因为我们的示例任务是无状态的。我们还可以支持有状态的任务，这些任务有一个初始状态，可以根据任务执行的结果进行更新。然而，这些超出了本文的范围。

5. 一次性任务

一次性任务的工作方式略有不同。在这种情况下，我们需要区分任务本身和标识任务的描述符：

TaskDescriptor<String> taskDescriptor = TaskDescriptor.of("my-onetime-task", String.class);
Task<String> task = Tasks.oneTime(taskDescriptor)
  .execute((inst, ctx) -> {
    LOG.info("Executed! Custom data {}, Instance {}", inst.getData(), inst.getId());
  });

此任务定义接受我们在调度任务时提供的一些数据，并可以适当地处理这些数据。在这种情况下，数据是一个 String，但它可以是 db-scheduler 能够序列化的任何类型。默认情况下，它使用 Java 序列化，但 db-scheduler 也支持其他机制，例如 Jackson 和 Gson。

然后，我们需要在调度器中注册该任务。但是，我们通过将任务传递给 Scheduler.create() 方法来做到这一点，这样 db-scheduler 就知道它，但不会立即开始运行它：

Scheduler scheduler = Scheduler.create(dataSource, task)
  .build();

此时，我们可以使用调度器和任务描述符随时运行任务：

scheduler.schedule(taskDescriptor.instance(UUID.randomUUID().toString())
  .data("Hello")
  .scheduledTo(Instant.now().plusSeconds(5))
);

在这里，我们必须为任务提供唯一的实例 ID，并指明任务应何时运行。我们还提供了一些数据供任务处理。然后，调度器将确保该任务在所需时间正确运行。

6. 动态重复任务

动态重复任务是在运行时动态注册的任务，类似于我们的一次性任务。然而，一旦注册，它们将类似于我们的简单重复任务，按计划继续运行。

我们创建和启动这些任务的方式与一次性任务完全相同，只是使用 Tasks.recurring() 代替：

TaskDescriptor<String> taskDescriptor = TaskDescriptor.of("my-dynamic-recurring-task", String.class);
Task<String> task = Tasks.recurring(taskDescriptor, new CronSchedule("*/5 * * * * ?", ZoneId.of("UTC")))
  .execute((inst, ctx) -> {
    LOG.info("Executed! Custom data {}, Instance {}", inst.getData(), inst.getId());
  });
scheduler.schedule(taskDescriptor.instance(UUID.randomUUID().toString())
  .data("Hello")
  .scheduledTo(Instant.now().plusSeconds(15))
);

这将按照给定时间启动我们的任务，然后一旦启动，它就会按照给定的时间表运行。在这种情况下，它会在当前时间 15 秒后首次运行，之后每 5 秒运行一次。

如果我们愿意，可以根据需要调度任意数量的不同实例，它们都会按照自己的计划运行，而不会相互干扰。

7. 与 Spring Boot 一起使用

如果我们使用 Spring Boot，db-scheduler 提供了一个我们可以使用的 starter 依赖：

<dependency>
    <groupId>com.github.kagkarlsson</groupId>
    <artifactId>db-scheduler-spring-boot-4-starter</artifactId>
    <version>16.12.0</version>
</dependency>

如果我们使用的是 Spring Boot 4.x，那么我们需要使用 db-scheduler-spring-boot-4-starter。如果我们仍然使用 Spring Boot 3.x，那么应该使用 db-scheduler-spring-boot-starter。

这将自动创建并启动我们的 Scheduler 实例作为 Spring bean。这样做时，它还会自动发现已创建为 Spring bean 的任何任务，并将它们注册到此调度器中。这意味着我们唯一需要做的就是创建任务本身。

如果我们想以任何方式配置 Scheduler，我们可以用标准方式使用属性来做到这一点：

db-scheduler.enabled=true
db-scheduler.polling-interval=5s
db-scheduler.heartbeat-interval=2m
db-scheduler.missed-heartbeats-limit=10
db-scheduler.threads=5

关于使用 db-scheduler 的其他一切，包括我们如何创建和调度任务，都与之前完全相同。 starter 只是让一开始的入门变得更简单。

8. 结论

在本文中，我们快速了解了 db-scheduler，包括如何调度重复任务和一次性任务。我们还可以用它做更多事情。下次需要为应用程序管理计划任务时，为什么不试试呢？

与往常一样，本文的所有代码都可以在 GitHub 上获取。