feat: 新增嵌入式分布式调度底座

- 提供通用调度 API、Quartz JDBC Provider 与独立 Starter

- 补充 MySQL、PostgreSQL、H2 建表脚本与接入校验

- 同步完善 Federation 与 Scheduler 模块说明
This commit is contained in:
2026-08-26 18:15:13 +08:00
parent 02b8fdd3ae
commit b4bdc392ee
47 changed files with 5169 additions and 1 deletions

View File

@@ -0,0 +1,43 @@
# Quartz 2.5.2 数据库脚本
`easy-agents-scheduler-quartz` 不会自动创建、删除或修改数据库结构。引用方需要选择与数据库匹配的脚本,并将其内容纳入自己的 Flyway、Liquibase 或初始化流程。
## 脚本与 Delegate
| 数据库 | 类路径资源 | `driverDelegateClass` |
|---|---|---|
| MySQL / MariaDB | `quartz-schema/mysql-2.5.2.sql` | `org.quartz.impl.jdbcjobstore.StdJDBCDelegate` |
| PostgreSQL | `quartz-schema/postgresql-2.5.2.sql` | `org.quartz.impl.jdbcjobstore.PostgreSQLDelegate` |
| H2 2.x开发与测试 | `quartz-schema/h2-2.5.2.sql` | `org.quartz.impl.jdbcjobstore.StdJDBCDelegate` |
默认表前缀是 `QRTZ_`。若调用方调整表名或 Schema需要同步设置 `tablePrefix`,例如 `scheduler.QRTZ_`
## 来源与变更
脚本来自 Quartz 官方仓库 `v2.5.2` 标签:
- [MySQL InnoDB 原始脚本](https://github.com/quartz-scheduler/quartz/blob/v2.5.2/quartz/src/main/resources/org/quartz/impl/jdbcjobstore/tables_mysql_innodb.sql)
- [PostgreSQL 原始脚本](https://github.com/quartz-scheduler/quartz/blob/v2.5.2/quartz/src/main/resources/org/quartz/impl/jdbcjobstore/tables_postgres.sql)
- [H2 原始脚本](https://github.com/quartz-scheduler/quartz/blob/v2.5.2/quartz/src/main/resources/org/quartz/impl/jdbcjobstore/tables_h2.sql)
为避免初始化脚本误删调用方已有任务,本模块从 MySQL 和 PostgreSQL 脚本中移除了全部 `DROP TABLE`。H2 版本另将旧式 `IMAGE` 类型替换为 H2 2.x 支持的 `BLOB`,并移除了已经失效的 `MVCC=TRUE` 提示。三套脚本同时移除了显式事务提交和行尾空白,使事务边界由调用方的迁移工具管理,其余建表结构沿用官方脚本。
## SHA-256
| 文件 | 官方原始文件 | 本模块资源 |
|---|---|---|
| MySQL | `90453cda26f4aad8abd35195a38ec36306561f0265f766dd12fbb42e185d5f82` | `1bfadd7762251836d6e15360e85f3c996a5c942a997ff0e9df30b59c50b0e97d` |
| PostgreSQL | `60d22e05e00203702f18f835916c75270aeb6828e7051bb81dd87ebd7bf6a943` | `911d058d415dbb9b29b344997c952e29d2e0b510ce4fe0b69f38adc5d5afffdb` |
| H2 | `21d806abef330e7c7175491b55b730acffa3c2e17a910840b32eee3fe2bf9fed` | `7750ac7b6c29d91072a5f8f98e4489d79469120effa9f613a7d9f89613beba5b` |
脚本随 Quartz 版本固化。后续升级 Quartz 时应新增版本化资源并完成数据库升级验证,不回改已经发布的脚本。
## 引用方接入要求
1. 在目标数据库中执行所选脚本,或复制到引用方自己的迁移文件中。
2. 保持同一集群节点的 `schedulerName``tablePrefix` 和数据库一致。
3. 每个节点使用唯一 `instanceId`,常规部署可配置为 `AUTO`
4. 生产集群启用 `clustered=true`,业务 Handler 按至少一次执行语义实现幂等。
5. 保留启动结构校验;表或关键列缺失时底座会拒绝启动并报告 `SCHEMA_INVALID`
本模块只包含 Quartz 的 `QRTZ_*` 技术表。任务定义、执行流水、审计和管理界面等业务表由引用方按自身需求设计。

View File

@@ -0,0 +1,40 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>com.easyagents</groupId>
<artifactId>easy-agents-scheduler</artifactId>
<version>${revision}</version>
</parent>
<artifactId>easy-agents-scheduler-quartz</artifactId>
<name>easy-agents-scheduler-quartz</name>
<dependencies>
<dependency>
<groupId>com.easyagents</groupId>
<artifactId>easy-agents-scheduler-core</artifactId>
</dependency>
<dependency>
<groupId>org.quartz-scheduler</groupId>
<artifactId>quartz</artifactId>
</dependency>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-api</artifactId>
</dependency>
<dependency>
<groupId>com.h2database</groupId>
<artifactId>h2</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
</project>

View File

@@ -0,0 +1,49 @@
package com.easyagents.scheduler.quartz;
import org.quartz.InterruptableJob;
import org.quartz.JobExecutionContext;
import org.quartz.JobExecutionException;
import org.quartz.SchedulerException;
/**
* 将 Quartz Job 委派给当前节点 {@link QuartzRuntime} 的内部基类。
*/
abstract class AbstractDispatchJob implements InterruptableJob {
private volatile Thread executionThread;
/**
* 从 SchedulerContext 定位运行时并分发任务。
*
* @param context Quartz 执行上下文
* @throws JobExecutionException 运行时缺失或 Handler 执行失败时抛出
*/
@Override
public final void execute(JobExecutionContext context) throws JobExecutionException {
executionThread = Thread.currentThread();
try {
Object runtime = context.getScheduler().getContext().get(
QuartzRuntime.SCHEDULER_CONTEXT_KEY
);
if (!(runtime instanceof QuartzRuntime quartzRuntime)) {
throw new JobExecutionException("easy-agents scheduler runtime is not available");
}
quartzRuntime.execute(context);
} catch (SchedulerException exception) {
throw new JobExecutionException("failed to access scheduler runtime", exception, false);
} finally {
executionThread = null;
}
}
/**
* 向当前 Handler 执行线程发送协作式中断信号。
*/
@Override
public final void interrupt() {
Thread currentExecutionThread = executionThread;
if (currentExecutionThread != null) {
currentExecutionThread.interrupt();
}
}
}

View File

@@ -0,0 +1,13 @@
package com.easyagents.scheduler.quartz;
/**
* 允许同一任务并发执行的内部 Quartz Job。
*/
public final class ConcurrentDispatchJob extends AbstractDispatchJob {
/**
* Quartz 反射创建 Job 所需的公共无参构造器。
*/
public ConcurrentDispatchJob() {
}
}

View File

@@ -0,0 +1,57 @@
package com.easyagents.scheduler.quartz;
import org.quartz.utils.ConnectionProvider;
import javax.sql.DataSource;
import java.sql.Connection;
import java.sql.SQLException;
import java.util.Objects;
/**
* 将调用方管理的 {@link DataSource} 暴露给 Quartz且不接管其生命周期。
*/
final class DataSourceConnectionProvider implements ConnectionProvider {
private volatile DataSource dataSource;
/**
* 创建外部 DataSource 连接提供器。
*
* @param dataSource 调用方管理的 DataSource
*/
DataSourceConnectionProvider(DataSource dataSource) {
this.dataSource = Objects.requireNonNull(dataSource, "dataSource must not be null");
}
/**
* 从调用方 DataSource 获取连接。
*
* @return JDBC 连接
* @throws SQLException 获取连接失败时抛出
*/
@Override
public Connection getConnection() throws SQLException {
DataSource currentDataSource = dataSource;
if (currentDataSource == null) {
throw new SQLException("Quartz DataSource provider is already shut down");
}
return currentDataSource.getConnection();
}
/**
* 外部 DataSource 不需要由 Quartz 初始化。
*/
@Override
public void initialize() {
// DataSource 的创建和初始化由调用方负责。
}
/**
* 外部 DataSource 不由 Quartz 关闭。
*/
@Override
public void shutdown() {
// Quartz 的全局注册表没有公开移除方法;解除引用以免长期持有调用方连接池。
dataSource = null;
}
}

View File

@@ -0,0 +1,16 @@
package com.easyagents.scheduler.quartz;
import org.quartz.DisallowConcurrentExecution;
/**
* 禁止同一 JobKey 并发执行的内部 Quartz Job。
*/
@DisallowConcurrentExecution
public final class DisallowConcurrentDispatchJob extends AbstractDispatchJob {
/**
* Quartz 反射创建 Job 所需的公共无参构造器。
*/
public DisallowConcurrentDispatchJob() {
}
}

View File

@@ -0,0 +1,57 @@
package com.easyagents.scheduler.quartz;
import com.easyagents.scheduler.ScheduleException;
import com.easyagents.scheduler.ScheduleId;
import org.quartz.Trigger;
import org.quartz.listeners.TriggerListenerSupport;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.time.Instant;
/**
* 将 Quartz Misfire 投影到通用执行监听器。
*/
final class QuartzMisfireListener extends TriggerListenerSupport {
private static final Logger log = LoggerFactory.getLogger(QuartzMisfireListener.class);
private final QuartzRuntime runtime;
/**
* 创建 Misfire 监听器。
*
* @param runtime 当前节点运行时
*/
QuartzMisfireListener(QuartzRuntime runtime) {
this.runtime = runtime;
}
/**
* 返回 Quartz Listener 名称。
*
* @return 稳定 Listener 名称
*/
@Override
public String getName() {
return "easyAgentsScheduleMisfireListener";
}
/**
* 转发 Misfire 事件,持久数据损坏时记录明确错误。
*
* @param trigger 发生 Misfire 的 Trigger
*/
@Override
public void triggerMisfired(Trigger trigger) {
try {
ScheduleId scheduleId = QuartzScheduleMapper.toScheduleId(trigger.getJobDataMap());
Instant expected = trigger.getNextFireTime() == null
? null
: trigger.getNextFireTime().toInstant();
runtime.notifyMisfired(scheduleId, expected);
} catch (ScheduleException | IllegalArgumentException exception) {
log.error("Failed to decode misfired Quartz trigger {}", trigger.getKey(), exception);
}
}
}

View File

@@ -0,0 +1,178 @@
package com.easyagents.scheduler.quartz;
import com.easyagents.scheduler.ScheduleDefinition;
import com.easyagents.scheduler.ScheduleErrorCode;
import com.easyagents.scheduler.ScheduleException;
import com.easyagents.scheduler.ScheduleExecutionListener;
import com.easyagents.scheduler.ScheduleFireContext;
import com.easyagents.scheduler.ScheduleHandler;
import com.easyagents.scheduler.ScheduleId;
import org.quartz.JobDataMap;
import org.quartz.JobExecutionContext;
import org.quartz.JobExecutionException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.time.Duration;
import java.time.Instant;
import java.util.Collection;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
/**
* 当前 Scheduler 节点的 Handler 和监听器运行时。
*/
final class QuartzRuntime {
static final String SCHEDULER_CONTEXT_KEY = "easyAgentsSchedulerRuntime";
private static final Logger log = LoggerFactory.getLogger(QuartzRuntime.class);
private final Map<String, ScheduleHandler> handlers;
private final List<ScheduleExecutionListener> listeners;
/**
* 创建节点本地运行时。
*
* @param handlers 当前节点 Handler
* @param listeners 当前节点执行监听器
*/
QuartzRuntime(
Collection<? extends ScheduleHandler> handlers,
Collection<? extends ScheduleExecutionListener> listeners
) {
this.handlers = registerHandlers(handlers);
this.listeners = listeners == null ? List.of() : List.copyOf(listeners);
}
/**
* 执行 Quartz 触发并保留 Handler 原始失败原因。
*
* @param quartzContext Quartz 执行上下文
* @throws JobExecutionException Handler 或运行时执行失败时抛出
*/
void execute(JobExecutionContext quartzContext) throws JobExecutionException {
JobDataMap data = quartzContext.getMergedJobDataMap();
ScheduleDefinition definition = QuartzScheduleMapper.toDefinition(data);
Instant actualFireTime = toInstant(quartzContext.getFireTime(), Instant.now());
ScheduleFireContext context = new ScheduleFireContext(
definition.id(),
definition.handlerCode(),
toInstant(quartzContext.getScheduledFireTime(), actualFireTime),
actualFireTime,
quartzContext.getFireInstanceId(),
stringValue(data, QuartzScheduleMapper.KEY_INVOCATION),
quartzContext.isRecovering(),
definition.parameters()
);
ScheduleHandler handler = handlers.get(definition.handlerCode());
if (handler == null) {
ScheduleException failure = new ScheduleException(
ScheduleErrorCode.HANDLER_NOT_FOUND,
"schedule handler not found: " + definition.handlerCode()
);
notifyFailed(context, Duration.ZERO, failure);
throw new JobExecutionException(failure, false);
}
notifyStarted(context);
long startedAt = System.nanoTime();
try {
handler.execute(context);
notifySucceeded(context, elapsed(startedAt));
} catch (Exception exception) {
notifyFailed(context, elapsed(startedAt), exception);
throw new JobExecutionException(exception, false);
}
}
/**
* 向监听器发布 Misfire 事件。
*
* @param scheduleId 任务标识
* @param expectedFireTime 原计划时间,可为空
*/
void notifyMisfired(ScheduleId scheduleId, Instant expectedFireTime) {
for (ScheduleExecutionListener listener : listeners) {
try {
listener.onMisfired(scheduleId, expectedFireTime);
} catch (RuntimeException exception) {
log.warn("Schedule listener failed on misfire: {}", scheduleId, exception);
}
}
}
private void notifyStarted(ScheduleFireContext context) {
for (ScheduleExecutionListener listener : listeners) {
try {
listener.onStarted(context);
} catch (RuntimeException exception) {
log.warn("Schedule listener failed on start: {}", context.scheduleId(), exception);
}
}
}
private void notifySucceeded(ScheduleFireContext context, Duration duration) {
for (ScheduleExecutionListener listener : listeners) {
try {
listener.onSucceeded(context, duration);
} catch (RuntimeException exception) {
log.warn("Schedule listener failed on success: {}", context.scheduleId(), exception);
}
}
}
private void notifyFailed(
ScheduleFireContext context,
Duration duration,
Throwable failure
) {
for (ScheduleExecutionListener listener : listeners) {
try {
listener.onFailed(context, duration, failure);
} catch (RuntimeException exception) {
log.warn("Schedule listener failed on failure: {}", context.scheduleId(), exception);
}
}
}
private static Map<String, ScheduleHandler> registerHandlers(
Collection<? extends ScheduleHandler> handlers
) {
Map<String, ScheduleHandler> registered = new LinkedHashMap<>();
if (handlers == null) {
return Map.of();
}
for (ScheduleHandler handler : handlers) {
String rawCode = handler == null ? null : handler.code();
if (rawCode == null || rawCode.isBlank()) {
throw new ScheduleException(
ScheduleErrorCode.INVALID_DEFINITION,
"schedule handler and handler code must not be blank"
);
}
String code = rawCode.trim();
ScheduleHandler previous = registered.putIfAbsent(code, handler);
if (previous != null) {
throw new ScheduleException(
ScheduleErrorCode.DUPLICATE_HANDLER,
"duplicate schedule handler code: " + code
);
}
}
return Map.copyOf(registered);
}
private static Duration elapsed(long startedAt) {
return Duration.ofNanos(Math.max(0L, System.nanoTime() - startedAt));
}
private static Instant toInstant(java.util.Date value, Instant fallback) {
return value == null ? fallback : value.toInstant();
}
private static String stringValue(JobDataMap data, String key) {
Object value = data.get(key);
return value == null ? null : value.toString();
}
}

View File

@@ -0,0 +1,378 @@
package com.easyagents.scheduler.quartz;
import com.easyagents.scheduler.ConcurrencyPolicy;
import com.easyagents.scheduler.CronSchedulePlan;
import com.easyagents.scheduler.MisfirePolicy;
import com.easyagents.scheduler.OnceSchedulePlan;
import com.easyagents.scheduler.ScheduleDefinition;
import com.easyagents.scheduler.ScheduleErrorCode;
import com.easyagents.scheduler.ScheduleException;
import com.easyagents.scheduler.ScheduleId;
import com.easyagents.scheduler.SchedulePlan;
import org.quartz.CronScheduleBuilder;
import org.quartz.JobBuilder;
import org.quartz.JobDataMap;
import org.quartz.JobDetail;
import org.quartz.JobKey;
import org.quartz.SimpleScheduleBuilder;
import org.quartz.Trigger;
import org.quartz.TriggerBuilder;
import org.quartz.TriggerKey;
import java.nio.charset.StandardCharsets;
import java.time.Instant;
import java.time.ZoneId;
import java.util.Base64;
import java.util.HashMap;
import java.util.Map;
import java.util.TimeZone;
/**
* Core 调度模型与 Quartz 持久模型之间的单一映射器。
*/
final class QuartzScheduleMapper {
static final String SCHEMA_VERSION = "1";
static final String KEY_SCHEMA_VERSION = "ea.schemaVersion";
static final String KEY_NAMESPACE = "ea.namespace";
static final String KEY_NAME = "ea.name";
static final String KEY_HANDLER = "ea.handler";
static final String KEY_PLAN_TYPE = "ea.planType";
static final String KEY_CRON = "ea.cron";
static final String KEY_ZONE = "ea.zone";
static final String KEY_FIRE_AT = "ea.fireAt";
static final String KEY_MISFIRE = "ea.misfire";
static final String KEY_CONCURRENCY = "ea.concurrency";
static final String KEY_RECOVER = "ea.recover";
static final String KEY_DESCRIPTION = "ea.description";
static final String KEY_INVOCATION = "ea.invocationId";
static final String KEY_IMMEDIATE_PARAMETER_SNAPSHOT = "ea.immediateParameterSnapshot";
static final String PARAMETER_PREFIX = "ea.parameter.";
static final String IMMEDIATE_PARAMETER_PREFIX = "ea.immediateParameter.";
static final String GROUP_PREFIX = "ea.scheduler.";
/** 禁止实例化映射器。 */
private QuartzScheduleMapper() {
}
/**
* 映射 Quartz Job Key。
*
* @param scheduleId 调度标识
* @return Job Key
*/
static JobKey jobKey(ScheduleId scheduleId) {
return new JobKey(scheduleId.name(), GROUP_PREFIX + scheduleId.namespace());
}
/**
* 映射 Quartz Trigger Key。
*
* @param scheduleId 调度标识
* @return Trigger Key
*/
static TriggerKey triggerKey(ScheduleId scheduleId) {
return new TriggerKey(scheduleId.name(), GROUP_PREFIX + scheduleId.namespace());
}
/**
* 将公共定义映射为持久 Job。
*
* @param definition 调度定义
* @return Job 详情
*/
static JobDetail toJobDetail(ScheduleDefinition definition) {
Class<? extends org.quartz.Job> jobClass =
definition.concurrencyPolicy() == ConcurrencyPolicy.DISALLOW
? DisallowConcurrentDispatchJob.class
: ConcurrentDispatchJob.class;
return JobBuilder.newJob(jobClass)
.withIdentity(jobKey(definition.id()))
.withDescription(emptyToNull(definition.description()))
.storeDurably(true)
.requestRecovery(definition.recoverOnNodeFailure())
.usingJobData(toJobData(definition))
.build();
}
/**
* 将公共定义映射为主 Trigger。
*
* @param definition 调度定义
* @return Quartz Trigger
*/
static Trigger toTrigger(ScheduleDefinition definition) {
TriggerBuilder<Trigger> builder = TriggerBuilder.newTrigger()
.withIdentity(triggerKey(definition.id()))
.forJob(jobKey(definition.id()))
.usingJobData(identityData(definition.id()));
SchedulePlan plan = definition.plan();
if (plan instanceof CronSchedulePlan cronPlan) {
CronScheduleBuilder schedule = CronScheduleBuilder
.cronSchedule(cronPlan.expression())
.inTimeZone(TimeZone.getTimeZone(cronPlan.zoneId()));
schedule = definition.misfirePolicy() == MisfirePolicy.SKIP
? schedule.withMisfireHandlingInstructionDoNothing()
: schedule.withMisfireHandlingInstructionFireAndProceed();
return builder.withSchedule(schedule).build();
}
if (plan instanceof OnceSchedulePlan oncePlan) {
SimpleScheduleBuilder schedule = SimpleScheduleBuilder.simpleSchedule()
.withRepeatCount(0);
schedule = definition.misfirePolicy() == MisfirePolicy.SKIP
? schedule.withMisfireHandlingInstructionNextWithExistingCount()
: schedule.withMisfireHandlingInstructionFireNow();
return builder
.startAt(oncePlan.fireAt())
.withSchedule(schedule)
.build();
}
throw new ScheduleException(
ScheduleErrorCode.INVALID_DEFINITION,
"unsupported schedule plan: " + plan.getClass().getName()
);
}
/**
* 构造仅对本次立即触发生效的数据。
*
* @param scheduleId 调度标识
* @param invocationId 调用标识
* @param parameters 已合并并校验的参数快照
* @return Trigger 数据
*/
static JobDataMap immediateData(
ScheduleId scheduleId,
String invocationId,
Map<String, String> parameters
) {
JobDataMap data = identityData(scheduleId);
data.put(KEY_INVOCATION, invocationId);
data.put(KEY_IMMEDIATE_PARAMETER_SNAPSHOT, Boolean.TRUE.toString());
putParameters(data, parameters, IMMEDIATE_PARAMETER_PREFIX);
return data;
}
/**
* 从持久 JobData 恢复公共定义。
*
* @param data 持久数据
* @return 调度定义
*/
static ScheduleDefinition toDefinition(JobDataMap data) {
try {
String version = required(data, KEY_SCHEMA_VERSION);
if (!SCHEMA_VERSION.equals(version)) {
throw new ScheduleException(
ScheduleErrorCode.PROVIDER_FAILURE,
"unsupported persisted schedule schema version: " + version
);
}
ScheduleId scheduleId = toScheduleId(data);
SchedulePlan plan = switch (required(data, KEY_PLAN_TYPE)) {
case "CRON" -> new CronSchedulePlan(
required(data, KEY_CRON),
ZoneId.of(required(data, KEY_ZONE))
);
case "ONCE" -> new OnceSchedulePlan(
Instant.ofEpochMilli(Long.parseLong(required(data, KEY_FIRE_AT)))
);
default -> throw new IllegalArgumentException("unknown plan type");
};
return new ScheduleDefinition(
scheduleId,
required(data, KEY_HANDLER),
plan,
MisfirePolicy.valueOf(required(data, KEY_MISFIRE)),
ConcurrencyPolicy.valueOf(required(data, KEY_CONCURRENCY)),
strictBoolean(data, KEY_RECOVER),
parameters(data),
stringValue(data, KEY_DESCRIPTION)
);
} catch (ScheduleException exception) {
if (exception.errorCode() == ScheduleErrorCode.PROVIDER_FAILURE) {
throw exception;
}
throw new ScheduleException(
ScheduleErrorCode.PROVIDER_FAILURE,
"invalid persisted schedule data",
exception
);
} catch (RuntimeException exception) {
throw new ScheduleException(
ScheduleErrorCode.PROVIDER_FAILURE,
"invalid persisted schedule data",
exception
);
}
}
/**
* 从 JobData 恢复调度标识。
*
* @param data 持久数据
* @return 调度标识
*/
static ScheduleId toScheduleId(JobDataMap data) {
return new ScheduleId(required(data, KEY_NAMESPACE), required(data, KEY_NAME));
}
/**
* 解码 JobData 中的业务参数。
*
* @param data 合并后的 JobData
* @return 不可变参数
*/
static Map<String, String> parameters(JobDataMap data) {
String parameterPrefix = immediateParameterSnapshot(data)
? IMMEDIATE_PARAMETER_PREFIX
: PARAMETER_PREFIX;
Map<String, String> parameters = new HashMap<>();
for (String key : data.getKeys()) {
if (!key.startsWith(parameterPrefix)) {
continue;
}
String encodedKey = key.substring(parameterPrefix.length());
String parameterKey = new String(
Base64.getUrlDecoder().decode(encodedKey),
StandardCharsets.UTF_8
);
parameters.put(parameterKey, stringValue(data, key));
}
return ScheduleDefinition.copyAndValidateParameters(parameters);
}
/**
* 将公共定义编码为纯字符串 JobData。
*
* @param definition 调度定义
* @return 持久数据
*/
private static JobDataMap toJobData(ScheduleDefinition definition) {
JobDataMap data = identityData(definition.id());
data.put(KEY_SCHEMA_VERSION, SCHEMA_VERSION);
data.put(KEY_HANDLER, definition.handlerCode());
data.put(KEY_MISFIRE, definition.misfirePolicy().name());
data.put(KEY_CONCURRENCY, definition.concurrencyPolicy().name());
data.put(KEY_RECOVER, Boolean.toString(definition.recoverOnNodeFailure()));
data.put(KEY_DESCRIPTION, definition.description());
if (definition.plan() instanceof CronSchedulePlan cronPlan) {
data.put(KEY_PLAN_TYPE, "CRON");
data.put(KEY_CRON, cronPlan.expression());
data.put(KEY_ZONE, cronPlan.zoneId().getId());
} else if (definition.plan() instanceof OnceSchedulePlan oncePlan) {
data.put(KEY_PLAN_TYPE, "ONCE");
data.put(KEY_FIRE_AT, Long.toString(oncePlan.fireAt().toEpochMilli()));
}
putParameters(data, definition.parameters(), PARAMETER_PREFIX);
return data;
}
/**
* 构造调度标识数据。
*
* @param scheduleId 调度标识
* @return 标识数据
*/
private static JobDataMap identityData(ScheduleId scheduleId) {
JobDataMap data = new JobDataMap();
data.put(KEY_NAMESPACE, scheduleId.namespace());
data.put(KEY_NAME, scheduleId.name());
return data;
}
/**
* 将参数编码到 JobData。
*
* @param data 目标 JobData
* @param parameters 参数
* @param parameterPrefix 参数键前缀
*/
private static void putParameters(
JobDataMap data,
Map<String, String> parameters,
String parameterPrefix
) {
for (Map.Entry<String, String> entry : parameters.entrySet()) {
String encodedKey = Base64.getUrlEncoder().withoutPadding().encodeToString(
entry.getKey().getBytes(StandardCharsets.UTF_8)
);
data.put(parameterPrefix + encodedKey, entry.getValue());
}
}
/**
* 判断当前数据是否包含立即触发参数快照。
*
* @param data 合并后的 JobData
* @return 存在完整快照时为 true
*/
private static boolean immediateParameterSnapshot(JobDataMap data) {
Object marker = data.get(KEY_IMMEDIATE_PARAMETER_SNAPSHOT);
if (marker == null) {
return false;
}
if (Boolean.TRUE.toString().equals(marker.toString())) {
return true;
}
throw new ScheduleException(
ScheduleErrorCode.PROVIDER_FAILURE,
"invalid immediate parameter snapshot marker"
);
}
/**
* 读取必填字符串字段。
*
* @param data JobData
* @param key 字段键
* @return 非空字段值
*/
private static String required(JobDataMap data, String key) {
String value = stringValue(data, key);
if (value == null || value.isBlank()) {
throw new IllegalArgumentException("missing job data: " + key);
}
return value;
}
/**
* 将 JobData 字段读取为字符串。
*
* @param data JobData
* @param key 字段键
* @return 字符串值,可为空
*/
private static String stringValue(JobDataMap data, String key) {
Object value = data.get(key);
return value == null ? null : value.toString();
}
/**
* 严格读取布尔字符串,避免损坏值静默降级为 false。
*
* @param data JobData
* @param key 字段键
* @return 布尔值
*/
private static boolean strictBoolean(JobDataMap data, String key) {
String value = required(data, key);
if ("true".equals(value)) {
return true;
}
if ("false".equals(value)) {
return false;
}
throw new IllegalArgumentException("invalid boolean job data: " + key);
}
/**
* 将空字符串转换为空值。
*
* @param value 原始值
* @return 非空文本或 null
*/
private static String emptyToNull(String value) {
return value == null || value.isEmpty() ? null : value;
}
}

View File

@@ -0,0 +1,571 @@
package com.easyagents.scheduler.quartz;
import com.easyagents.scheduler.CronSchedulePlan;
import com.easyagents.scheduler.OnceSchedulePlan;
import com.easyagents.scheduler.ScheduleDefinition;
import com.easyagents.scheduler.ScheduleErrorCode;
import com.easyagents.scheduler.ScheduleException;
import com.easyagents.scheduler.ScheduleFireReceipt;
import com.easyagents.scheduler.ScheduleId;
import com.easyagents.scheduler.SchedulePlan;
import com.easyagents.scheduler.ScheduleService;
import com.easyagents.scheduler.ScheduleStatus;
import com.easyagents.scheduler.ScheduleView;
import org.quartz.CronExpression;
import org.quartz.JobDataMap;
import org.quartz.JobDetail;
import org.quartz.ObjectAlreadyExistsException;
import org.quartz.Scheduler;
import org.quartz.SchedulerException;
import org.quartz.Trigger;
import org.quartz.Trigger.TriggerState;
import org.quartz.TriggerKey;
import java.text.ParseException;
import java.time.Instant;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.atomic.AtomicBoolean;
/**
* 基于单个 Quartz Scheduler 的调度服务实现。
*/
public final class QuartzScheduleService implements ScheduleService, AutoCloseable {
/** 立即触发调用标识的最大长度。 */
public static final int MAX_INVOCATION_ID_LENGTH = 190;
/** 单次触发时间预览的最大数量。 */
public static final int MAX_PREVIEW_LIMIT = 100;
private final Scheduler scheduler;
private final boolean waitForJobsToCompleteOnShutdown;
private final long shutdownWaitTimeoutMillis;
private final AtomicBoolean closed = new AtomicBoolean();
/**
* 创建 Quartz 调度服务。
*
* @param scheduler 已完成运行时装配、尚未启动的 Scheduler
* @param waitForJobsToCompleteOnShutdown 关闭时是否等待在途任务
* @param shutdownWaitTimeoutMillis 关闭等待最长毫秒数
*/
QuartzScheduleService(
Scheduler scheduler,
boolean waitForJobsToCompleteOnShutdown,
long shutdownWaitTimeoutMillis
) {
this.scheduler = Objects.requireNonNull(scheduler, "scheduler must not be null");
this.waitForJobsToCompleteOnShutdown = waitForJobsToCompleteOnShutdown;
if (shutdownWaitTimeoutMillis < 1) {
throw new IllegalArgumentException("shutdownWaitTimeoutMillis must be positive");
}
this.shutdownWaitTimeoutMillis = shutdownWaitTimeoutMillis;
}
/**
* 启动底层 Quartz Scheduler。
*
* @throws ScheduleException Scheduler 启动失败或已经关闭时抛出
*/
public void start() {
ensureOpen();
try {
if (!scheduler.isStarted()) {
scheduler.start();
}
} catch (SchedulerException exception) {
ScheduleException failure = providerFailure(
"failed to start Quartz scheduler",
exception
);
closed.set(true);
try {
if (!scheduler.isShutdown()) {
scheduler.shutdown(false);
}
} catch (SchedulerException cleanupFailure) {
failure.addSuppressed(cleanupFailure);
}
throw failure;
}
}
/**
* {@inheritDoc}
*/
@Override
public ScheduleView create(ScheduleDefinition definition) {
Objects.requireNonNull(definition, "definition must not be null");
ensureOpen();
JobDetail job = mapJob(definition);
Trigger trigger = mapTrigger(definition);
try {
scheduler.scheduleJob(job, trigger);
return requireView(definition.id());
} catch (ObjectAlreadyExistsException exception) {
Optional<ScheduleView> existing = get(definition.id());
if (existing.isPresent() && existing.get().definition().equals(definition)) {
return existing.get();
}
throw new ScheduleException(
ScheduleErrorCode.SCHEDULE_CONFLICT,
"schedule already exists with a different definition: " + definition.id(),
exception
);
} catch (SchedulerException exception) {
throw providerFailure("failed to create schedule " + definition.id(), exception);
}
}
/**
* {@inheritDoc}
*/
@Override
public ScheduleView replace(ScheduleDefinition definition) {
Objects.requireNonNull(definition, "definition must not be null");
ensureOpen();
JobDetail job = mapJob(definition);
Trigger trigger = mapTrigger(definition);
try {
// Quartz 在一个 JobStore 事务中替换 Job 与其唯一 Trigger避免中间态。
scheduler.scheduleJob(job, Set.of(trigger), true);
return requireView(definition.id());
} catch (SchedulerException exception) {
throw providerFailure("failed to replace schedule " + definition.id(), exception);
}
}
/**
* {@inheritDoc}
*/
@Override
public ScheduleView pause(ScheduleId scheduleId) {
Objects.requireNonNull(scheduleId, "scheduleId must not be null");
ensureOpen();
try {
if (!scheduler.checkExists(QuartzScheduleMapper.jobKey(scheduleId))) {
throw notFound(scheduleId);
}
TriggerKey triggerKey = QuartzScheduleMapper.triggerKey(scheduleId);
if (scheduler.checkExists(triggerKey)) {
scheduler.pauseTrigger(triggerKey);
}
return requireView(scheduleId);
} catch (ScheduleException exception) {
throw exception;
} catch (SchedulerException exception) {
throw providerFailure("failed to pause schedule " + scheduleId, exception);
}
}
/**
* {@inheritDoc}
*/
@Override
public ScheduleView resume(ScheduleId scheduleId) {
Objects.requireNonNull(scheduleId, "scheduleId must not be null");
ensureOpen();
try {
if (!scheduler.checkExists(QuartzScheduleMapper.jobKey(scheduleId))) {
throw notFound(scheduleId);
}
TriggerKey triggerKey = QuartzScheduleMapper.triggerKey(scheduleId);
if (scheduler.checkExists(triggerKey)) {
scheduler.resumeTrigger(triggerKey);
}
return requireView(scheduleId);
} catch (ScheduleException exception) {
throw exception;
} catch (SchedulerException exception) {
throw providerFailure("failed to resume schedule " + scheduleId, exception);
}
}
/**
* {@inheritDoc}
*/
@Override
public boolean delete(ScheduleId scheduleId) {
Objects.requireNonNull(scheduleId, "scheduleId must not be null");
ensureOpen();
try {
// 批量接口对单个 Job 也直接委派 JobStore避免逐 Trigger 删除时的完成竞态。
return scheduler.deleteJobs(List.of(QuartzScheduleMapper.jobKey(scheduleId)));
} catch (SchedulerException exception) {
throw providerFailure("failed to delete schedule " + scheduleId, exception);
}
}
/**
* {@inheritDoc}
*/
@Override
public ScheduleFireReceipt triggerNow(
ScheduleId scheduleId,
String invocationId,
Map<String, String> parameters
) {
Objects.requireNonNull(scheduleId, "scheduleId must not be null");
String normalizedInvocationId = normalizeInvocationId(invocationId);
ensureOpen();
try {
JobDetail job = scheduler.getJobDetail(QuartzScheduleMapper.jobKey(scheduleId));
if (job == null) {
throw notFound(scheduleId);
}
Map<String, String> mergedParameters = new HashMap<>(
QuartzScheduleMapper.toDefinition(job.getJobDataMap()).parameters()
);
if (parameters != null) {
mergedParameters.putAll(parameters);
}
Map<String, String> safeParameters = ScheduleDefinition
.copyAndValidateParameters(mergedParameters);
JobDataMap data = QuartzScheduleMapper.immediateData(
scheduleId,
normalizedInvocationId,
safeParameters
);
scheduler.triggerJob(QuartzScheduleMapper.jobKey(scheduleId), data);
return new ScheduleFireReceipt(scheduleId, normalizedInvocationId, Instant.now());
} catch (ScheduleException exception) {
throw exception;
} catch (SchedulerException exception) {
throw providerFailure("failed to trigger schedule " + scheduleId, exception);
}
}
/**
* {@inheritDoc}
*/
@Override
public Optional<ScheduleView> get(ScheduleId scheduleId) {
Objects.requireNonNull(scheduleId, "scheduleId must not be null");
ensureOpen();
try {
JobDetail job = scheduler.getJobDetail(QuartzScheduleMapper.jobKey(scheduleId));
if (job == null) {
return Optional.empty();
}
ScheduleDefinition definition = QuartzScheduleMapper.toDefinition(job.getJobDataMap());
TriggerKey triggerKey = QuartzScheduleMapper.triggerKey(scheduleId);
Trigger trigger = scheduler.getTrigger(triggerKey);
TriggerState triggerState = trigger == null
? TriggerState.NONE
: scheduler.getTriggerState(triggerKey);
return Optional.of(new ScheduleView(
definition,
toStatus(triggerState, trigger == null),
toInstant(trigger == null ? null : trigger.getPreviousFireTime()),
toInstant(trigger == null ? null : trigger.getNextFireTime())
));
} catch (SchedulerException exception) {
throw providerFailure("failed to read schedule " + scheduleId, exception);
}
}
/**
* {@inheritDoc}
*/
@Override
public List<Instant> nextFireTimes(SchedulePlan plan, int limit) {
Objects.requireNonNull(plan, "plan must not be null");
if (limit < 1 || limit > MAX_PREVIEW_LIMIT) {
throw new ScheduleException(
ScheduleErrorCode.INVALID_DEFINITION,
"limit must be between 1 and " + MAX_PREVIEW_LIMIT
);
}
if (plan instanceof OnceSchedulePlan oncePlan) {
return oncePlan.fireAt().isAfter(Instant.now())
? List.of(oncePlan.fireAt())
: List.of();
}
if (plan instanceof CronSchedulePlan cronPlan) {
return nextCronFireTimes(cronPlan, limit);
}
throw new ScheduleException(
ScheduleErrorCode.INVALID_DEFINITION,
"unsupported schedule plan: " + plan.getClass().getName()
);
}
/**
* 关闭 Scheduler 并释放 Provider 对外部 DataSource 的强引用。
*
* <p>外部 DataSource 的连接池所有权仍属于调用方,本方法不会关闭连接池。</p>
*
* @throws ScheduleException Scheduler 关闭失败时抛出
*/
@Override
public void close() {
if (!closed.compareAndSet(false, true)) {
return;
}
ScheduleException failure = null;
try {
if (!scheduler.isShutdown()) {
scheduler.standby();
if (
waitForJobsToCompleteOnShutdown
&& !awaitExecutingJobs(shutdownWaitTimeoutMillis)
) {
failure = new ScheduleException(
ScheduleErrorCode.PROVIDER_FAILURE,
"timed out after " + shutdownWaitTimeoutMillis
+ " ms waiting for Quartz jobs to complete"
);
}
}
} catch (InterruptedException exception) {
Thread.currentThread().interrupt();
failure = providerFailure(
"interrupted while waiting for Quartz jobs to complete",
exception
);
} catch (SchedulerException exception) {
failure = providerFailure("failed to prepare Quartz scheduler shutdown", exception);
}
try {
if (!scheduler.isShutdown()) {
// false 保证 close 本身有界Quartz 会向内部 InterruptableJob 发送中断。
scheduler.shutdown(false);
}
} catch (SchedulerException exception) {
if (failure == null) {
failure = providerFailure("failed to close Quartz scheduler", exception);
} else {
failure.addSuppressed(exception);
}
}
if (failure != null) {
throw failure;
}
}
/**
* 返回底层 Scheduler供 Provider 诊断和测试使用。
*
* @return 当前 Quartz Scheduler
*/
Scheduler quartzScheduler() {
return scheduler;
}
/**
* 在给定时限内等待当前节点任务完成。
*
* @param timeoutMillis 最长等待毫秒数
* @return 全部完成时为 true超时时为 false
* @throws SchedulerException 读取执行状态失败时抛出
* @throws InterruptedException 等待线程被中断时抛出
*/
private boolean awaitExecutingJobs(long timeoutMillis)
throws SchedulerException, InterruptedException {
long deadline = System.nanoTime() + java.util.concurrent.TimeUnit.MILLISECONDS.toNanos(
timeoutMillis
);
while (!scheduler.getCurrentlyExecutingJobs().isEmpty()) {
long remainingNanos = deadline - System.nanoTime();
if (remainingNanos <= 0L) {
return false;
}
long sleepMillis = Math.max(
1L,
Math.min(
100L,
java.util.concurrent.TimeUnit.NANOSECONDS.toMillis(remainingNanos)
)
);
Thread.sleep(sleepMillis);
}
return true;
}
/**
* 查询必然存在的任务视图。
*
* @param scheduleId 调度标识
* @return 任务视图
*/
private ScheduleView requireView(ScheduleId scheduleId) {
return get(scheduleId).orElseThrow(() -> notFound(scheduleId));
}
/**
* 映射 Job 并统一定义错误。
*
* @param definition 调度定义
* @return Job 详情
*/
private static JobDetail mapJob(ScheduleDefinition definition) {
try {
return QuartzScheduleMapper.toJobDetail(definition);
} catch (ScheduleException exception) {
throw exception;
} catch (RuntimeException exception) {
throw invalidDefinition(definition.id(), exception);
}
}
/**
* 映射 Trigger 并统一定义错误。
*
* @param definition 调度定义
* @return Quartz Trigger
*/
private static Trigger mapTrigger(ScheduleDefinition definition) {
try {
return QuartzScheduleMapper.toTrigger(definition);
} catch (ScheduleException exception) {
throw exception;
} catch (RuntimeException exception) {
throw invalidDefinition(definition.id(), exception);
}
}
/**
* 计算 Cron 后续触发时间。
*
* @param plan Cron 计划
* @param limit 最大数量
* @return 触发时间
*/
private static List<Instant> nextCronFireTimes(CronSchedulePlan plan, int limit) {
try {
CronExpression expression = new CronExpression(plan.expression());
expression.setTimeZone(java.util.TimeZone.getTimeZone(plan.zoneId()));
List<Instant> fireTimes = new ArrayList<>(limit);
Date cursor = Date.from(Instant.now());
for (int index = 0; index < limit; index++) {
Date next = expression.getNextValidTimeAfter(cursor);
if (next == null) {
break;
}
fireTimes.add(next.toInstant());
cursor = next;
}
return List.copyOf(fireTimes);
} catch (ParseException exception) {
throw new ScheduleException(
ScheduleErrorCode.INVALID_DEFINITION,
"invalid Quartz cron expression: " + plan.expression(),
exception
);
}
}
/**
* 映射 Quartz Trigger 状态。
*
* @param state Quartz 状态
* @param completed 是否已无主 Trigger
* @return 公共状态
*/
private static ScheduleStatus toStatus(TriggerState state, boolean completed) {
if (completed) {
return ScheduleStatus.COMPLETE;
}
return switch (state) {
case NORMAL -> ScheduleStatus.SCHEDULED;
case PAUSED -> ScheduleStatus.PAUSED;
case BLOCKED -> ScheduleStatus.BLOCKED;
case ERROR -> ScheduleStatus.ERROR;
case COMPLETE, NONE -> ScheduleStatus.COMPLETE;
};
}
/**
* 将 JDBC/Quartz 时间转换为 Instant。
*
* @param value 日期,可为空
* @return Instant可为空
*/
private static Instant toInstant(Date value) {
return value == null ? null : value.toInstant();
}
/**
* 规范化立即触发调用标识。
*
* @param invocationId 原始调用标识
* @return 规范化标识
*/
private static String normalizeInvocationId(String invocationId) {
if (invocationId == null || invocationId.isBlank()) {
throw new ScheduleException(
ScheduleErrorCode.INVALID_DEFINITION,
"invocationId must not be blank"
);
}
String normalized = invocationId.trim();
if (normalized.length() > MAX_INVOCATION_ID_LENGTH) {
throw new ScheduleException(
ScheduleErrorCode.INVALID_DEFINITION,
"invocationId length must not exceed " + MAX_INVOCATION_ID_LENGTH
);
}
return normalized;
}
/**
* 确认服务仍可接受操作。
*/
private void ensureOpen() {
if (closed.get()) {
throw new ScheduleException(
ScheduleErrorCode.SCHEDULER_CLOSED,
"scheduler is already closed"
);
}
}
/**
* 构造任务不存在异常。
*
* @param scheduleId 调度标识
* @return 稳定异常
*/
private static ScheduleException notFound(ScheduleId scheduleId) {
return new ScheduleException(
ScheduleErrorCode.SCHEDULE_NOT_FOUND,
"schedule not found: " + scheduleId
);
}
/**
* 构造定义非法异常。
*
* @param scheduleId 调度标识
* @param exception 原始原因
* @return 稳定异常
*/
private static ScheduleException invalidDefinition(
ScheduleId scheduleId,
RuntimeException exception
) {
return new ScheduleException(
ScheduleErrorCode.INVALID_DEFINITION,
"invalid schedule definition: " + scheduleId,
exception
);
}
/**
* 构造 Provider 失败异常。
*
* @param message 失败摘要
* @param exception 原始原因
* @return 稳定异常
*/
private static ScheduleException providerFailure(String message, Exception exception) {
return new ScheduleException(ScheduleErrorCode.PROVIDER_FAILURE, message, exception);
}
}

View File

@@ -0,0 +1,114 @@
package com.easyagents.scheduler.quartz;
/**
* 使用外部 DataSource 创建 Quartz JDBC 调度器的不可变配置。
*
* @param schedulerName 集群内保持一致的 Scheduler Name
* @param instanceId 当前实例标识,通常使用 AUTO
* @param tablePrefix Quartz 表前缀,可包含 Schema 前缀
* @param driverDelegateClass Quartz JDBC DriverDelegate 类名
* @param clustered 是否启用 JDBC 集群
* @param threadCount Quartz Worker 线程数
* @param threadPriority Quartz Worker 线程优先级
* @param clusterCheckinIntervalMillis 集群心跳间隔,单位毫秒
* @param misfireThresholdMillis Misfire 判定阈值,单位毫秒
* @param waitForJobsToCompleteOnShutdown 关闭时是否等待运行中任务完成
* @param shutdownWaitTimeoutMillis 关闭等待运行中任务的最长时间,单位毫秒
* @param validateSchema 启动前是否执行只读表结构校验
*/
public record QuartzSchedulerConfig(
String schedulerName,
String instanceId,
String tablePrefix,
String driverDelegateClass,
boolean clustered,
int threadCount,
int threadPriority,
long clusterCheckinIntervalMillis,
long misfireThresholdMillis,
boolean waitForJobsToCompleteOnShutdown,
long shutdownWaitTimeoutMillis,
boolean validateSchema
) {
/** Quartz 官方通用 JDBC Delegate。 */
public static final String STANDARD_JDBC_DELEGATE =
"org.quartz.impl.jdbcjobstore.StdJDBCDelegate";
/**
* 校验并创建 Quartz 配置。
*
* @throws IllegalArgumentException 任一配置越界时抛出
*/
public QuartzSchedulerConfig {
schedulerName = requireText(schedulerName, "schedulerName", 120);
instanceId = requireText(instanceId, "instanceId", 190);
tablePrefix = requireText(tablePrefix, "tablePrefix", 180);
driverDelegateClass = requireText(
driverDelegateClass,
"driverDelegateClass",
250
);
if (threadCount < 1 || threadCount > 1000) {
throw new IllegalArgumentException("threadCount must be between 1 and 1000");
}
if (threadPriority < Thread.MIN_PRIORITY || threadPriority > Thread.MAX_PRIORITY) {
throw new IllegalArgumentException("threadPriority must be between 1 and 10");
}
if (clusterCheckinIntervalMillis < 1000) {
throw new IllegalArgumentException(
"clusterCheckinIntervalMillis must be at least 1000"
);
}
if (misfireThresholdMillis < 1) {
throw new IllegalArgumentException("misfireThresholdMillis must be positive");
}
if (shutdownWaitTimeoutMillis < 1 || shutdownWaitTimeoutMillis > 600_000L) {
throw new IllegalArgumentException(
"shutdownWaitTimeoutMillis must be between 1 and 600000"
);
}
}
/**
* 创建适合常规 JDBC 集群的安全默认配置。
*
* @param schedulerName Scheduler Name
* @return 默认配置
*/
public static QuartzSchedulerConfig clusteredDefaults(String schedulerName) {
return new QuartzSchedulerConfig(
schedulerName,
"AUTO",
"QRTZ_",
STANDARD_JDBC_DELEGATE,
true,
8,
Thread.NORM_PRIORITY,
15_000L,
60_000L,
true,
30_000L,
true
);
}
/**
* 规范化必填配置文本。
*
* @param value 原始值
* @param field 字段名
* @param maxLength 最大长度
* @return 规范化值
*/
private static String requireText(String value, String field, int maxLength) {
if (value == null || value.isBlank()) {
throw new IllegalArgumentException(field + " must not be blank");
}
String normalized = value.trim();
if (normalized.length() > maxLength) {
throw new IllegalArgumentException(field + " length must not exceed " + maxLength);
}
return normalized;
}
}

View File

@@ -0,0 +1,258 @@
package com.easyagents.scheduler.quartz;
import com.easyagents.scheduler.ScheduleErrorCode;
import com.easyagents.scheduler.ScheduleException;
import com.easyagents.scheduler.ScheduleExecutionListener;
import com.easyagents.scheduler.ScheduleHandler;
import org.quartz.Scheduler;
import org.quartz.SchedulerException;
import org.quartz.impl.SchedulerRepository;
import org.quartz.impl.StdSchedulerFactory;
import org.quartz.utils.DBConnectionManager;
import javax.sql.DataSource;
import java.sql.SQLException;
import java.util.Collection;
import java.util.List;
import java.util.Objects;
import java.util.Properties;
/**
* 使用调用方 DataSource 创建独立 Quartz Scheduler 的工厂。
*/
public final class QuartzSchedulerFactory {
/** 禁止实例化工厂。 */
private QuartzSchedulerFactory() {
}
/**
* 创建采用 Quartz JDBC JobStore 的调度服务。
*
* <p>返回的服务尚未启动,调用方应在应用生命周期就绪时调用
* {@link QuartzScheduleService#start()}。</p>
*
* @param dataSource 调用方管理生命周期的 DataSource
* @param config Quartz Scheduler 配置
* @param handlers 当前节点注册的 Handler
* @param listeners 当前节点注册的监听器
* @return 尚未启动的调度服务
* @throws ScheduleException 结构校验或 Scheduler 初始化失败时抛出
*/
public static synchronized QuartzScheduleService createJdbc(
DataSource dataSource,
QuartzSchedulerConfig config,
Collection<? extends ScheduleHandler> handlers,
Collection<? extends ScheduleExecutionListener> listeners
) {
Objects.requireNonNull(dataSource, "dataSource must not be null");
Objects.requireNonNull(config, "config must not be null");
if (config.validateSchema()) {
try {
QuartzSchemaValidator.validate(dataSource, config.tablePrefix());
} catch (ScheduleException exception) {
throw new ScheduleException(
exception.errorCode(),
"Quartz schema validation failed for scheduler " + config.schedulerName()
+ " and prefix " + config.tablePrefix() + ": "
+ exception.getMessage(),
exception
);
}
}
rejectDuplicateSchedulerName(config.schedulerName());
// Quartz 的全局连接注册表不支持移除,稳定名称可避免应用内重建时持续增长。
String dataSourceName = "easyAgentsScheduler-" + config.schedulerName();
DBConnectionManager connectionManager = DBConnectionManager.getInstance();
connectionManager.addConnectionProvider(
dataSourceName,
new DataSourceConnectionProvider(dataSource)
);
Scheduler scheduler = null;
try {
scheduler = new StdSchedulerFactory(properties(config, dataSourceName)).getScheduler();
return attach(
scheduler,
config.waitForJobsToCompleteOnShutdown(),
config.shutdownWaitTimeoutMillis(),
handlers,
listeners
);
} catch (Exception exception) {
cleanupAfterFailure(scheduler, dataSourceName, exception);
if (exception instanceof ScheduleException scheduleException) {
throw scheduleException;
}
throw new ScheduleException(
ScheduleErrorCode.PROVIDER_FAILURE,
"failed to initialize Quartz scheduler " + config.schedulerName(),
exception
);
}
}
/**
* 将 Handler 运行时装配到已有 Scheduler供无数据库的单元测试复用。
*
* @param scheduler 已创建且未关闭的 Scheduler
* @param handlers 当前节点 Handler
* @param listeners 当前节点监听器
* @return 尚未启动的调度服务
* @throws ScheduleException 运行时装配失败时抛出
*/
static QuartzScheduleService attach(
Scheduler scheduler,
Collection<? extends ScheduleHandler> handlers,
Collection<? extends ScheduleExecutionListener> listeners
) {
return attach(scheduler, true, 30_000L, handlers, listeners);
}
/**
* 使用显式关闭策略装配已有 Scheduler。
*
* @param scheduler 已创建的 Scheduler
* @param waitForJobsToCompleteOnShutdown 是否等待在途任务
* @param shutdownWaitTimeoutMillis 最长等待毫秒数
* @param handlers 当前节点 Handler
* @param listeners 当前节点监听器
* @return 尚未启动的调度服务
*/
static QuartzScheduleService attach(
Scheduler scheduler,
boolean waitForJobsToCompleteOnShutdown,
long shutdownWaitTimeoutMillis,
Collection<? extends ScheduleHandler> handlers,
Collection<? extends ScheduleExecutionListener> listeners
) {
try {
QuartzRuntime runtime = new QuartzRuntime(
handlers == null ? List.of() : handlers,
listeners == null ? List.of() : listeners
);
scheduler.getContext().put(QuartzRuntime.SCHEDULER_CONTEXT_KEY, runtime);
scheduler.getListenerManager().addTriggerListener(new QuartzMisfireListener(runtime));
return new QuartzScheduleService(
scheduler,
waitForJobsToCompleteOnShutdown,
shutdownWaitTimeoutMillis
);
} catch (SchedulerException exception) {
throw new ScheduleException(
ScheduleErrorCode.PROVIDER_FAILURE,
"failed to attach Easy Agents scheduler runtime",
exception
);
}
}
/**
* 构造隔离的原生 Quartz 配置。
*
* @param config Provider 配置
* @param dataSourceName Quartz 内部 DataSource 名称
* @return Quartz 属性
*/
private static Properties properties(
QuartzSchedulerConfig config,
String dataSourceName
) {
Properties properties = new Properties();
properties.setProperty("org.quartz.scheduler.instanceName", config.schedulerName());
properties.setProperty("org.quartz.scheduler.instanceId", config.instanceId());
properties.setProperty(
"org.quartz.threadPool.class",
"org.quartz.simpl.SimpleThreadPool"
);
properties.setProperty(
"org.quartz.threadPool.threadCount",
Integer.toString(config.threadCount())
);
properties.setProperty(
"org.quartz.threadPool.threadPriority",
Integer.toString(config.threadPriority())
);
properties.setProperty(
"org.quartz.jobStore.class",
"org.quartz.impl.jdbcjobstore.JobStoreTX"
);
properties.setProperty(
"org.quartz.jobStore.driverDelegateClass",
config.driverDelegateClass()
);
properties.setProperty("org.quartz.jobStore.useProperties", "true");
properties.setProperty("org.quartz.jobStore.dataSource", dataSourceName);
properties.setProperty("org.quartz.jobStore.tablePrefix", config.tablePrefix());
properties.setProperty(
"org.quartz.jobStore.isClustered",
Boolean.toString(config.clustered())
);
properties.setProperty(
"org.quartz.jobStore.clusterCheckinInterval",
Long.toString(config.clusterCheckinIntervalMillis())
);
properties.setProperty(
"org.quartz.jobStore.misfireThreshold",
Long.toString(config.misfireThresholdMillis())
);
properties.setProperty("org.quartz.scheduler.interruptJobsOnShutdown", "true");
return properties;
}
/**
* 拒绝同 JVM 中仍存活的同名 Scheduler。
*
* @param schedulerName Scheduler Name
*/
private static void rejectDuplicateSchedulerName(String schedulerName) {
Scheduler existing = SchedulerRepository.getInstance().lookup(schedulerName);
if (existing == null) {
return;
}
try {
if (existing.isShutdown()) {
SchedulerRepository.getInstance().remove(schedulerName);
return;
}
} catch (SchedulerException exception) {
throw new ScheduleException(
ScheduleErrorCode.PROVIDER_FAILURE,
"failed to inspect existing Quartz scheduler " + schedulerName,
exception
);
}
throw new ScheduleException(
ScheduleErrorCode.PROVIDER_FAILURE,
"Quartz scheduler name is already in use in this JVM: " + schedulerName
);
}
/**
* 清理初始化失败留下的 Scheduler 与 DataSource 引用。
*
* @param scheduler 已创建的 Scheduler可为空
* @param dataSourceName Quartz 内部 DataSource 名称
* @param originalFailure 原始失败
*/
private static void cleanupAfterFailure(
Scheduler scheduler,
String dataSourceName,
Exception originalFailure
) {
if (scheduler != null) {
try {
if (!scheduler.isShutdown()) {
scheduler.shutdown(false);
}
} catch (SchedulerException cleanupFailure) {
originalFailure.addSuppressed(cleanupFailure);
}
}
try {
DBConnectionManager.getInstance().shutdown(dataSourceName);
} catch (SQLException cleanupFailure) {
originalFailure.addSuppressed(cleanupFailure);
}
}
}

View File

@@ -0,0 +1,261 @@
package com.easyagents.scheduler.quartz;
import com.easyagents.scheduler.ScheduleErrorCode;
import com.easyagents.scheduler.ScheduleException;
import javax.sql.DataSource;
import java.sql.Connection;
import java.sql.DatabaseMetaData;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
/**
* Quartz JDBC JobStore 必需结构的只读校验器。
*/
public final class QuartzSchemaValidator {
private static final List<String> REQUIRED_SUFFIXES = List.of(
"JOB_DETAILS",
"TRIGGERS",
"SIMPLE_TRIGGERS",
"CRON_TRIGGERS",
"SIMPROP_TRIGGERS",
"BLOB_TRIGGERS",
"CALENDARS",
"PAUSED_TRIGGER_GRPS",
"FIRED_TRIGGERS",
"SCHEDULER_STATE",
"LOCKS"
);
private static final Map<String, Set<String>> REQUIRED_COLUMNS = Map.ofEntries(
Map.entry("JOB_DETAILS", Set.of(
"SCHED_NAME", "JOB_NAME", "JOB_GROUP", "DESCRIPTION", "JOB_CLASS_NAME",
"IS_DURABLE", "IS_NONCONCURRENT", "IS_UPDATE_DATA", "REQUESTS_RECOVERY",
"JOB_DATA"
)),
Map.entry("TRIGGERS", Set.of(
"SCHED_NAME", "TRIGGER_NAME", "TRIGGER_GROUP", "JOB_NAME", "JOB_GROUP",
"DESCRIPTION", "NEXT_FIRE_TIME", "PREV_FIRE_TIME", "PRIORITY", "TRIGGER_STATE",
"TRIGGER_TYPE", "START_TIME", "END_TIME", "CALENDAR_NAME", "MISFIRE_INSTR",
"JOB_DATA"
)),
Map.entry("SIMPLE_TRIGGERS", Set.of(
"SCHED_NAME", "TRIGGER_NAME", "TRIGGER_GROUP", "REPEAT_COUNT",
"REPEAT_INTERVAL", "TIMES_TRIGGERED"
)),
Map.entry("CRON_TRIGGERS", Set.of(
"SCHED_NAME", "TRIGGER_NAME", "TRIGGER_GROUP", "CRON_EXPRESSION", "TIME_ZONE_ID"
)),
Map.entry("SIMPROP_TRIGGERS", Set.of(
"SCHED_NAME", "TRIGGER_NAME", "TRIGGER_GROUP", "STR_PROP_1", "STR_PROP_2",
"STR_PROP_3", "INT_PROP_1", "INT_PROP_2", "LONG_PROP_1", "LONG_PROP_2",
"DEC_PROP_1", "DEC_PROP_2", "BOOL_PROP_1", "BOOL_PROP_2"
)),
Map.entry("BLOB_TRIGGERS", Set.of(
"SCHED_NAME", "TRIGGER_NAME", "TRIGGER_GROUP", "BLOB_DATA"
)),
Map.entry("CALENDARS", Set.of("SCHED_NAME", "CALENDAR_NAME", "CALENDAR")),
Map.entry("PAUSED_TRIGGER_GRPS", Set.of("SCHED_NAME", "TRIGGER_GROUP")),
Map.entry("FIRED_TRIGGERS", Set.of(
"SCHED_NAME", "ENTRY_ID", "TRIGGER_NAME", "TRIGGER_GROUP", "INSTANCE_NAME",
"FIRED_TIME", "SCHED_TIME", "PRIORITY", "STATE", "JOB_NAME", "JOB_GROUP",
"IS_NONCONCURRENT", "REQUESTS_RECOVERY"
)),
Map.entry("SCHEDULER_STATE", Set.of(
"SCHED_NAME", "INSTANCE_NAME", "LAST_CHECKIN_TIME", "CHECKIN_INTERVAL"
)),
Map.entry("LOCKS", Set.of("SCHED_NAME", "LOCK_NAME"))
);
private QuartzSchemaValidator() {
}
/**
* 校验指定 DataSource 中的 Quartz 表和关键列。
*
* @param dataSource 调用方 DataSource
* @param tablePrefix Quartz 表前缀,可包含 Schema
* @throws ScheduleException 表或关键列缺失、元数据读取失败时抛出
*/
public static void validate(DataSource dataSource, String tablePrefix) {
Objects.requireNonNull(dataSource, "dataSource must not be null");
PrefixParts prefixParts = PrefixParts.parse(tablePrefix);
try (Connection connection = dataSource.getConnection()) {
DatabaseMetaData metadata = connection.getMetaData();
String databaseProduct = metadata.getDatabaseProductName();
MetadataScope metadataScope = resolveMetadataScope(
connection,
metadata,
prefixParts.schema()
);
Map<String, String> actualTables = loadTables(
metadata,
metadataScope
);
List<String> missingTables = new ArrayList<>();
for (String suffix : REQUIRED_SUFFIXES) {
String expected = normalize(prefixParts.tablePrefix() + suffix);
if (!actualTables.containsKey(expected)) {
missingTables.add(prefixParts.displayPrefix() + suffix);
}
}
if (!missingTables.isEmpty()) {
throw invalid(
"missing Quartz tables in " + databaseProduct + ": "
+ String.join(", ", missingTables)
);
}
validateColumns(
metadata,
metadataScope,
prefixParts,
actualTables,
databaseProduct
);
} catch (SQLException exception) {
throw new ScheduleException(
ScheduleErrorCode.SCHEMA_INVALID,
"failed to inspect Quartz schema for prefix " + tablePrefix,
exception
);
}
}
/**
* 解析 JDBC 元数据查询使用的 Catalog 与 Schema。
*
* @param connection 当前数据库连接
* @param metadata JDBC 元数据
* @param declaredSchema 表前缀中显式声明的 Schema可为空
* @return 仅覆盖 Quartz 实际查询范围的元数据作用域
* @throws SQLException 读取连接或数据库能力失败时抛出
*/
private static MetadataScope resolveMetadataScope(
Connection connection,
DatabaseMetaData metadata,
String declaredSchema
) throws SQLException {
if (
!metadata.supportsSchemasInTableDefinitions()
&& metadata.supportsCatalogsInTableDefinitions()
) {
String catalog = declaredSchema == null ? connection.getCatalog() : declaredSchema;
return new MetadataScope(catalog, null);
}
String schema = declaredSchema == null ? connection.getSchema() : declaredSchema;
return new MetadataScope(connection.getCatalog(), schema);
}
/**
* 加载指定作用域内的数据库表。
*
* @param metadata JDBC 元数据
* @param metadataScope Catalog 与 Schema 作用域
* @return 规范化表名到实际表名的映射
* @throws SQLException 元数据读取失败时抛出
*/
private static Map<String, String> loadTables(
DatabaseMetaData metadata,
MetadataScope metadataScope
) throws SQLException {
Map<String, String> tables = new HashMap<>();
try (ResultSet resultSet = metadata.getTables(
metadataScope.catalog(),
metadataScope.schema(),
null,
new String[]{"TABLE"}
)) {
while (resultSet.next()) {
String tableName = resultSet.getString("TABLE_NAME");
if (tableName != null) {
tables.put(normalize(tableName), tableName);
}
}
}
return tables;
}
/**
* 校验指定作用域内 Quartz 表的必需列。
*
* @param metadata JDBC 元数据
* @param metadataScope Catalog 与 Schema 作用域
* @param prefixParts Quartz 表前缀
* @param actualTables 已发现表
* @param databaseProduct 数据库产品名
* @throws SQLException 元数据读取失败时抛出
*/
private static void validateColumns(
DatabaseMetaData metadata,
MetadataScope metadataScope,
PrefixParts prefixParts,
Map<String, String> actualTables,
String databaseProduct
) throws SQLException {
for (Map.Entry<String, Set<String>> requirement : REQUIRED_COLUMNS.entrySet()) {
String normalizedName = normalize(prefixParts.tablePrefix() + requirement.getKey());
String actualName = actualTables.get(normalizedName);
Set<String> actualColumns = new HashSet<>();
try (ResultSet resultSet = metadata.getColumns(
metadataScope.catalog(),
metadataScope.schema(),
actualName,
null
)) {
while (resultSet.next()) {
actualColumns.add(normalize(resultSet.getString("COLUMN_NAME")));
}
}
Set<String> missingColumns = new HashSet<>(requirement.getValue());
missingColumns.removeAll(actualColumns);
if (!missingColumns.isEmpty()) {
throw invalid(
"missing columns in " + databaseProduct + " table "
+ prefixParts.displayPrefix() + requirement.getKey() + ": "
+ String.join(", ", missingColumns)
);
}
}
}
private static ScheduleException invalid(String message) {
return new ScheduleException(ScheduleErrorCode.SCHEMA_INVALID, message);
}
private static String normalize(String value) {
return value == null ? "" : value.toUpperCase(Locale.ROOT);
}
/** JDBC 元数据的 Catalog 与 Schema 查询范围。 */
private record MetadataScope(String catalog, String schema) {
}
private record PrefixParts(String schema, String tablePrefix, String displayPrefix) {
private static PrefixParts parse(String value) {
if (value == null || value.isBlank()) {
throw new IllegalArgumentException("tablePrefix must not be blank");
}
String normalized = value.trim();
int separator = normalized.lastIndexOf('.');
if (separator < 0) {
return new PrefixParts(null, normalized, normalized);
}
String schema = normalized.substring(0, separator);
String prefix = normalized.substring(separator + 1);
if (schema.isBlank() || prefix.isBlank()) {
throw new IllegalArgumentException("invalid tablePrefix: " + value);
}
return new PrefixParts(schema, prefix, normalized);
}
}
}

View File

@@ -0,0 +1,249 @@
-- Quartz 2.5.2 JDBC JobStore 建表脚本。
-- 上游来源https://raw.githubusercontent.com/quartz-scheduler/quartz/v2.5.2/quartz/src/main/resources/org/quartz/impl/jdbcjobstore/tables_h2.sql
-- 本脚本仅创建结构,不包含 DROP请由调用方纳入自己的数据库迁移流程。
-- Thanks to Amir Kibbar and Peter Rietzler for contributing the schema for H2 database,
-- and verifying that it works with Quartz's StdJDBCDelegate
--
-- H2 2.x 已移除 MVCC URL 参数;默认 MVStore 提供行级锁。
--
--
-- In your Quartz properties file, you'll need to set
-- org.quartz.jobStore.driverDelegateClass = org.quartz.impl.jdbcjobstore.StdJDBCDelegate
CREATE TABLE QRTZ_CALENDARS (
SCHED_NAME VARCHAR(120) NOT NULL,
CALENDAR_NAME VARCHAR (200) NOT NULL ,
CALENDAR BLOB NOT NULL
);
CREATE TABLE QRTZ_CRON_TRIGGERS (
SCHED_NAME VARCHAR(120) NOT NULL,
TRIGGER_NAME VARCHAR (200) NOT NULL ,
TRIGGER_GROUP VARCHAR (200) NOT NULL ,
CRON_EXPRESSION VARCHAR (120) NOT NULL ,
TIME_ZONE_ID VARCHAR (80)
);
CREATE TABLE QRTZ_FIRED_TRIGGERS (
SCHED_NAME VARCHAR(120) NOT NULL,
ENTRY_ID VARCHAR (95) NOT NULL ,
TRIGGER_NAME VARCHAR (200) NOT NULL ,
TRIGGER_GROUP VARCHAR (200) NOT NULL ,
INSTANCE_NAME VARCHAR (200) NOT NULL ,
FIRED_TIME BIGINT NOT NULL ,
SCHED_TIME BIGINT NOT NULL ,
PRIORITY INTEGER NOT NULL ,
STATE VARCHAR (16) NOT NULL,
JOB_NAME VARCHAR (200) NULL ,
JOB_GROUP VARCHAR (200) NULL ,
IS_NONCONCURRENT BOOLEAN NULL ,
REQUESTS_RECOVERY BOOLEAN NULL
);
CREATE TABLE QRTZ_PAUSED_TRIGGER_GRPS (
SCHED_NAME VARCHAR(120) NOT NULL,
TRIGGER_GROUP VARCHAR (200) NOT NULL
);
CREATE TABLE QRTZ_SCHEDULER_STATE (
SCHED_NAME VARCHAR(120) NOT NULL,
INSTANCE_NAME VARCHAR (200) NOT NULL ,
LAST_CHECKIN_TIME BIGINT NOT NULL ,
CHECKIN_INTERVAL BIGINT NOT NULL
);
CREATE TABLE QRTZ_LOCKS (
SCHED_NAME VARCHAR(120) NOT NULL,
LOCK_NAME VARCHAR (40) NOT NULL
);
CREATE TABLE QRTZ_JOB_DETAILS (
SCHED_NAME VARCHAR(120) NOT NULL,
JOB_NAME VARCHAR (200) NOT NULL ,
JOB_GROUP VARCHAR (200) NOT NULL ,
DESCRIPTION VARCHAR (250) NULL ,
JOB_CLASS_NAME VARCHAR (250) NOT NULL ,
IS_DURABLE BOOLEAN NOT NULL ,
IS_NONCONCURRENT BOOLEAN NOT NULL ,
IS_UPDATE_DATA BOOLEAN NOT NULL ,
REQUESTS_RECOVERY BOOLEAN NOT NULL ,
JOB_DATA BLOB NULL
);
CREATE TABLE QRTZ_SIMPLE_TRIGGERS (
SCHED_NAME VARCHAR(120) NOT NULL,
TRIGGER_NAME VARCHAR (200) NOT NULL ,
TRIGGER_GROUP VARCHAR (200) NOT NULL ,
REPEAT_COUNT BIGINT NOT NULL ,
REPEAT_INTERVAL BIGINT NOT NULL ,
TIMES_TRIGGERED BIGINT NOT NULL
);
CREATE TABLE QRTZ_SIMPROP_TRIGGERS (
SCHED_NAME VARCHAR(120) NOT NULL,
TRIGGER_NAME VARCHAR(200) NOT NULL,
TRIGGER_GROUP VARCHAR(200) NOT NULL,
STR_PROP_1 VARCHAR(512) NULL,
STR_PROP_2 VARCHAR(512) NULL,
STR_PROP_3 VARCHAR(512) NULL,
INT_PROP_1 INTEGER NULL,
INT_PROP_2 INTEGER NULL,
LONG_PROP_1 BIGINT NULL,
LONG_PROP_2 BIGINT NULL,
DEC_PROP_1 NUMERIC(13,4) NULL,
DEC_PROP_2 NUMERIC(13,4) NULL,
BOOL_PROP_1 BOOLEAN NULL,
BOOL_PROP_2 BOOLEAN NULL
);
CREATE TABLE QRTZ_BLOB_TRIGGERS (
SCHED_NAME VARCHAR(120) NOT NULL,
TRIGGER_NAME VARCHAR (200) NOT NULL ,
TRIGGER_GROUP VARCHAR (200) NOT NULL ,
BLOB_DATA BLOB NULL
);
CREATE TABLE QRTZ_TRIGGERS (
SCHED_NAME VARCHAR(120) NOT NULL,
TRIGGER_NAME VARCHAR (200) NOT NULL ,
TRIGGER_GROUP VARCHAR (200) NOT NULL ,
JOB_NAME VARCHAR (200) NOT NULL ,
JOB_GROUP VARCHAR (200) NOT NULL ,
DESCRIPTION VARCHAR (250) NULL ,
NEXT_FIRE_TIME BIGINT NULL ,
PREV_FIRE_TIME BIGINT NULL ,
PRIORITY INTEGER NULL ,
TRIGGER_STATE VARCHAR (16) NOT NULL ,
TRIGGER_TYPE VARCHAR (8) NOT NULL ,
START_TIME BIGINT NOT NULL ,
END_TIME BIGINT NULL ,
CALENDAR_NAME VARCHAR (200) NULL ,
MISFIRE_INSTR SMALLINT NULL ,
JOB_DATA BLOB NULL
);
ALTER TABLE QRTZ_CALENDARS ADD
CONSTRAINT PK_QRTZ_CALENDARS PRIMARY KEY
(
SCHED_NAME,
CALENDAR_NAME
);
ALTER TABLE QRTZ_CRON_TRIGGERS ADD
CONSTRAINT PK_QRTZ_CRON_TRIGGERS PRIMARY KEY
(
SCHED_NAME,
TRIGGER_NAME,
TRIGGER_GROUP
);
ALTER TABLE QRTZ_FIRED_TRIGGERS ADD
CONSTRAINT PK_QRTZ_FIRED_TRIGGERS PRIMARY KEY
(
SCHED_NAME,
ENTRY_ID
);
ALTER TABLE QRTZ_PAUSED_TRIGGER_GRPS ADD
CONSTRAINT PK_QRTZ_PAUSED_TRIGGER_GRPS PRIMARY KEY
(
SCHED_NAME,
TRIGGER_GROUP
);
ALTER TABLE QRTZ_SCHEDULER_STATE ADD
CONSTRAINT PK_QRTZ_SCHEDULER_STATE PRIMARY KEY
(
SCHED_NAME,
INSTANCE_NAME
);
ALTER TABLE QRTZ_LOCKS ADD
CONSTRAINT PK_QRTZ_LOCKS PRIMARY KEY
(
SCHED_NAME,
LOCK_NAME
);
ALTER TABLE QRTZ_JOB_DETAILS ADD
CONSTRAINT PK_QRTZ_JOB_DETAILS PRIMARY KEY
(
SCHED_NAME,
JOB_NAME,
JOB_GROUP
);
ALTER TABLE QRTZ_SIMPLE_TRIGGERS ADD
CONSTRAINT PK_QRTZ_SIMPLE_TRIGGERS PRIMARY KEY
(
SCHED_NAME,
TRIGGER_NAME,
TRIGGER_GROUP
);
ALTER TABLE QRTZ_SIMPROP_TRIGGERS ADD
CONSTRAINT PK_QRTZ_SIMPROP_TRIGGERS PRIMARY KEY
(
SCHED_NAME,
TRIGGER_NAME,
TRIGGER_GROUP
);
ALTER TABLE QRTZ_TRIGGERS ADD
CONSTRAINT PK_QRTZ_TRIGGERS PRIMARY KEY
(
SCHED_NAME,
TRIGGER_NAME,
TRIGGER_GROUP
);
ALTER TABLE QRTZ_CRON_TRIGGERS ADD
CONSTRAINT FK_QRTZ_CRON_TRIGGERS_QRTZ_TRIGGERS FOREIGN KEY
(
SCHED_NAME,
TRIGGER_NAME,
TRIGGER_GROUP
) REFERENCES QRTZ_TRIGGERS (
SCHED_NAME,
TRIGGER_NAME,
TRIGGER_GROUP
) ON DELETE CASCADE;
ALTER TABLE QRTZ_SIMPLE_TRIGGERS ADD
CONSTRAINT FK_QRTZ_SIMPLE_TRIGGERS_QRTZ_TRIGGERS FOREIGN KEY
(
SCHED_NAME,
TRIGGER_NAME,
TRIGGER_GROUP
) REFERENCES QRTZ_TRIGGERS (
SCHED_NAME,
TRIGGER_NAME,
TRIGGER_GROUP
) ON DELETE CASCADE;
ALTER TABLE QRTZ_SIMPROP_TRIGGERS ADD
CONSTRAINT FK_QRTZ_SIMPROP_TRIGGERS_QRTZ_TRIGGERS FOREIGN KEY
(
SCHED_NAME,
TRIGGER_NAME,
TRIGGER_GROUP
) REFERENCES QRTZ_TRIGGERS (
SCHED_NAME,
TRIGGER_NAME,
TRIGGER_GROUP
) ON DELETE CASCADE;
ALTER TABLE QRTZ_TRIGGERS ADD
CONSTRAINT FK_QRTZ_TRIGGERS_QRTZ_JOB_DETAILS FOREIGN KEY
(
SCHED_NAME,
JOB_NAME,
JOB_GROUP
) REFERENCES QRTZ_JOB_DETAILS (
SCHED_NAME,
JOB_NAME,
JOB_GROUP
);

View File

@@ -0,0 +1,170 @@
-- Quartz 2.5.2 JDBC JobStore 建表脚本。
-- 上游来源https://raw.githubusercontent.com/quartz-scheduler/quartz/v2.5.2/quartz/src/main/resources/org/quartz/impl/jdbcjobstore/tables_mysql_innodb.sql
-- 本脚本仅创建结构,不包含 DROP请由调用方纳入自己的数据库迁移流程。
--
-- In your Quartz properties file, you'll need to set
-- org.quartz.jobStore.driverDelegateClass = org.quartz.impl.jdbcjobstore.StdJDBCDelegate
--
--
-- By: Ron Cordell - roncordell
-- I didn't see this anywhere, so I thought I'd post it here. This is the script from Quartz to create the tables in a MySQL database, modified to use INNODB instead of MYISAM.
CREATE TABLE QRTZ_JOB_DETAILS(
SCHED_NAME VARCHAR(120) NOT NULL,
JOB_NAME VARCHAR(190) NOT NULL,
JOB_GROUP VARCHAR(190) NOT NULL,
DESCRIPTION VARCHAR(250) NULL,
JOB_CLASS_NAME VARCHAR(250) NOT NULL,
IS_DURABLE VARCHAR(1) NOT NULL,
IS_NONCONCURRENT VARCHAR(1) NOT NULL,
IS_UPDATE_DATA VARCHAR(1) NOT NULL,
REQUESTS_RECOVERY VARCHAR(1) NOT NULL,
JOB_DATA BLOB NULL,
PRIMARY KEY (SCHED_NAME,JOB_NAME,JOB_GROUP))
ENGINE=InnoDB;
CREATE TABLE QRTZ_TRIGGERS (
SCHED_NAME VARCHAR(120) NOT NULL,
TRIGGER_NAME VARCHAR(190) NOT NULL,
TRIGGER_GROUP VARCHAR(190) NOT NULL,
JOB_NAME VARCHAR(190) NOT NULL,
JOB_GROUP VARCHAR(190) NOT NULL,
DESCRIPTION VARCHAR(250) NULL,
NEXT_FIRE_TIME BIGINT(13) NULL,
PREV_FIRE_TIME BIGINT(13) NULL,
PRIORITY INTEGER NULL,
TRIGGER_STATE VARCHAR(16) NOT NULL,
TRIGGER_TYPE VARCHAR(8) NOT NULL,
START_TIME BIGINT(13) NOT NULL,
END_TIME BIGINT(13) NULL,
CALENDAR_NAME VARCHAR(190) NULL,
MISFIRE_INSTR SMALLINT(2) NULL,
JOB_DATA BLOB NULL,
PRIMARY KEY (SCHED_NAME,TRIGGER_NAME,TRIGGER_GROUP),
FOREIGN KEY (SCHED_NAME,JOB_NAME,JOB_GROUP)
REFERENCES QRTZ_JOB_DETAILS(SCHED_NAME,JOB_NAME,JOB_GROUP))
ENGINE=InnoDB;
CREATE TABLE QRTZ_SIMPLE_TRIGGERS (
SCHED_NAME VARCHAR(120) NOT NULL,
TRIGGER_NAME VARCHAR(190) NOT NULL,
TRIGGER_GROUP VARCHAR(190) NOT NULL,
REPEAT_COUNT BIGINT(7) NOT NULL,
REPEAT_INTERVAL BIGINT(12) NOT NULL,
TIMES_TRIGGERED BIGINT(10) NOT NULL,
PRIMARY KEY (SCHED_NAME,TRIGGER_NAME,TRIGGER_GROUP),
FOREIGN KEY (SCHED_NAME,TRIGGER_NAME,TRIGGER_GROUP)
REFERENCES QRTZ_TRIGGERS(SCHED_NAME,TRIGGER_NAME,TRIGGER_GROUP))
ENGINE=InnoDB;
CREATE TABLE QRTZ_CRON_TRIGGERS (
SCHED_NAME VARCHAR(120) NOT NULL,
TRIGGER_NAME VARCHAR(190) NOT NULL,
TRIGGER_GROUP VARCHAR(190) NOT NULL,
CRON_EXPRESSION VARCHAR(120) NOT NULL,
TIME_ZONE_ID VARCHAR(80),
PRIMARY KEY (SCHED_NAME,TRIGGER_NAME,TRIGGER_GROUP),
FOREIGN KEY (SCHED_NAME,TRIGGER_NAME,TRIGGER_GROUP)
REFERENCES QRTZ_TRIGGERS(SCHED_NAME,TRIGGER_NAME,TRIGGER_GROUP))
ENGINE=InnoDB;
CREATE TABLE QRTZ_SIMPROP_TRIGGERS
(
SCHED_NAME VARCHAR(120) NOT NULL,
TRIGGER_NAME VARCHAR(190) NOT NULL,
TRIGGER_GROUP VARCHAR(190) NOT NULL,
STR_PROP_1 VARCHAR(512) NULL,
STR_PROP_2 VARCHAR(512) NULL,
STR_PROP_3 VARCHAR(512) NULL,
INT_PROP_1 INT NULL,
INT_PROP_2 INT NULL,
LONG_PROP_1 BIGINT NULL,
LONG_PROP_2 BIGINT NULL,
DEC_PROP_1 NUMERIC(13,4) NULL,
DEC_PROP_2 NUMERIC(13,4) NULL,
BOOL_PROP_1 VARCHAR(1) NULL,
BOOL_PROP_2 VARCHAR(1) NULL,
PRIMARY KEY (SCHED_NAME,TRIGGER_NAME,TRIGGER_GROUP),
FOREIGN KEY (SCHED_NAME,TRIGGER_NAME,TRIGGER_GROUP)
REFERENCES QRTZ_TRIGGERS(SCHED_NAME,TRIGGER_NAME,TRIGGER_GROUP))
ENGINE=InnoDB;
CREATE TABLE QRTZ_BLOB_TRIGGERS (
SCHED_NAME VARCHAR(120) NOT NULL,
TRIGGER_NAME VARCHAR(190) NOT NULL,
TRIGGER_GROUP VARCHAR(190) NOT NULL,
BLOB_DATA BLOB NULL,
PRIMARY KEY (SCHED_NAME,TRIGGER_NAME,TRIGGER_GROUP),
INDEX (SCHED_NAME,TRIGGER_NAME, TRIGGER_GROUP),
FOREIGN KEY (SCHED_NAME,TRIGGER_NAME,TRIGGER_GROUP)
REFERENCES QRTZ_TRIGGERS(SCHED_NAME,TRIGGER_NAME,TRIGGER_GROUP))
ENGINE=InnoDB;
CREATE TABLE QRTZ_CALENDARS (
SCHED_NAME VARCHAR(120) NOT NULL,
CALENDAR_NAME VARCHAR(190) NOT NULL,
CALENDAR BLOB NOT NULL,
PRIMARY KEY (SCHED_NAME,CALENDAR_NAME))
ENGINE=InnoDB;
CREATE TABLE QRTZ_PAUSED_TRIGGER_GRPS (
SCHED_NAME VARCHAR(120) NOT NULL,
TRIGGER_GROUP VARCHAR(190) NOT NULL,
PRIMARY KEY (SCHED_NAME,TRIGGER_GROUP))
ENGINE=InnoDB;
CREATE TABLE QRTZ_FIRED_TRIGGERS (
SCHED_NAME VARCHAR(120) NOT NULL,
ENTRY_ID VARCHAR(95) NOT NULL,
TRIGGER_NAME VARCHAR(190) NOT NULL,
TRIGGER_GROUP VARCHAR(190) NOT NULL,
INSTANCE_NAME VARCHAR(190) NOT NULL,
FIRED_TIME BIGINT(13) NOT NULL,
SCHED_TIME BIGINT(13) NOT NULL,
PRIORITY INTEGER NOT NULL,
STATE VARCHAR(16) NOT NULL,
JOB_NAME VARCHAR(190) NULL,
JOB_GROUP VARCHAR(190) NULL,
IS_NONCONCURRENT VARCHAR(1) NULL,
REQUESTS_RECOVERY VARCHAR(1) NULL,
PRIMARY KEY (SCHED_NAME,ENTRY_ID))
ENGINE=InnoDB;
CREATE TABLE QRTZ_SCHEDULER_STATE (
SCHED_NAME VARCHAR(120) NOT NULL,
INSTANCE_NAME VARCHAR(190) NOT NULL,
LAST_CHECKIN_TIME BIGINT(13) NOT NULL,
CHECKIN_INTERVAL BIGINT(13) NOT NULL,
PRIMARY KEY (SCHED_NAME,INSTANCE_NAME))
ENGINE=InnoDB;
CREATE TABLE QRTZ_LOCKS (
SCHED_NAME VARCHAR(120) NOT NULL,
LOCK_NAME VARCHAR(40) NOT NULL,
PRIMARY KEY (SCHED_NAME,LOCK_NAME))
ENGINE=InnoDB;
CREATE INDEX IDX_QRTZ_J_REQ_RECOVERY ON QRTZ_JOB_DETAILS(SCHED_NAME,REQUESTS_RECOVERY);
CREATE INDEX IDX_QRTZ_J_GRP ON QRTZ_JOB_DETAILS(SCHED_NAME,JOB_GROUP);
CREATE INDEX IDX_QRTZ_T_J ON QRTZ_TRIGGERS(SCHED_NAME,JOB_NAME,JOB_GROUP);
CREATE INDEX IDX_QRTZ_T_JG ON QRTZ_TRIGGERS(SCHED_NAME,JOB_GROUP);
CREATE INDEX IDX_QRTZ_T_C ON QRTZ_TRIGGERS(SCHED_NAME,CALENDAR_NAME);
CREATE INDEX IDX_QRTZ_T_G ON QRTZ_TRIGGERS(SCHED_NAME,TRIGGER_GROUP);
CREATE INDEX IDX_QRTZ_T_STATE ON QRTZ_TRIGGERS(SCHED_NAME,TRIGGER_STATE);
CREATE INDEX IDX_QRTZ_T_N_STATE ON QRTZ_TRIGGERS(SCHED_NAME,TRIGGER_NAME,TRIGGER_GROUP,TRIGGER_STATE);
CREATE INDEX IDX_QRTZ_T_N_G_STATE ON QRTZ_TRIGGERS(SCHED_NAME,TRIGGER_GROUP,TRIGGER_STATE);
CREATE INDEX IDX_QRTZ_T_NEXT_FIRE_TIME ON QRTZ_TRIGGERS(SCHED_NAME,NEXT_FIRE_TIME);
CREATE INDEX IDX_QRTZ_T_NFT_ST ON QRTZ_TRIGGERS(SCHED_NAME,TRIGGER_STATE,NEXT_FIRE_TIME);
CREATE INDEX IDX_QRTZ_T_NFT_MISFIRE ON QRTZ_TRIGGERS(SCHED_NAME,MISFIRE_INSTR,NEXT_FIRE_TIME);
CREATE INDEX IDX_QRTZ_T_NFT_ST_MISFIRE ON QRTZ_TRIGGERS(SCHED_NAME,MISFIRE_INSTR,NEXT_FIRE_TIME,TRIGGER_STATE);
CREATE INDEX IDX_QRTZ_T_NFT_ST_MISFIRE_GRP ON QRTZ_TRIGGERS(SCHED_NAME,MISFIRE_INSTR,NEXT_FIRE_TIME,TRIGGER_GROUP,TRIGGER_STATE);
CREATE INDEX IDX_QRTZ_FT_TRIG_INST_NAME ON QRTZ_FIRED_TRIGGERS(SCHED_NAME,INSTANCE_NAME);
CREATE INDEX IDX_QRTZ_FT_INST_JOB_REQ_RCVRY ON QRTZ_FIRED_TRIGGERS(SCHED_NAME,INSTANCE_NAME,REQUESTS_RECOVERY);
CREATE INDEX IDX_QRTZ_FT_J_G ON QRTZ_FIRED_TRIGGERS(SCHED_NAME,JOB_NAME,JOB_GROUP);
CREATE INDEX IDX_QRTZ_FT_JG ON QRTZ_FIRED_TRIGGERS(SCHED_NAME,JOB_GROUP);
CREATE INDEX IDX_QRTZ_FT_T_G ON QRTZ_FIRED_TRIGGERS(SCHED_NAME,TRIGGER_NAME,TRIGGER_GROUP);
CREATE INDEX IDX_QRTZ_FT_TG ON QRTZ_FIRED_TRIGGERS(SCHED_NAME,TRIGGER_GROUP);

View File

@@ -0,0 +1,197 @@
-- Quartz 2.5.2 JDBC JobStore 建表脚本。
-- 上游来源https://raw.githubusercontent.com/quartz-scheduler/quartz/v2.5.2/quartz/src/main/resources/org/quartz/impl/jdbcjobstore/tables_postgres.sql
-- 本脚本仅创建结构,不包含 DROP请由调用方纳入自己的数据库迁移流程。
-- Thanks to Patrick Lightbody for submitting this...
--
-- In your Quartz properties file, you'll need to set
-- org.quartz.jobStore.driverDelegateClass = org.quartz.impl.jdbcjobstore.PostgreSQLDelegate
CREATE TABLE QRTZ_JOB_DETAILS
(
SCHED_NAME VARCHAR(120) NOT NULL,
JOB_NAME VARCHAR(200) NOT NULL,
JOB_GROUP VARCHAR(200) NOT NULL,
DESCRIPTION VARCHAR(250) NULL,
JOB_CLASS_NAME VARCHAR(250) NOT NULL,
IS_DURABLE BOOL NOT NULL,
IS_NONCONCURRENT BOOL NOT NULL,
IS_UPDATE_DATA BOOL NOT NULL,
REQUESTS_RECOVERY BOOL NOT NULL,
JOB_DATA BYTEA NULL,
PRIMARY KEY (SCHED_NAME, JOB_NAME, JOB_GROUP)
);
CREATE TABLE QRTZ_TRIGGERS
(
SCHED_NAME VARCHAR(120) NOT NULL,
TRIGGER_NAME VARCHAR(200) NOT NULL,
TRIGGER_GROUP VARCHAR(200) NOT NULL,
JOB_NAME VARCHAR(200) NOT NULL,
JOB_GROUP VARCHAR(200) NOT NULL,
DESCRIPTION VARCHAR(250) NULL,
NEXT_FIRE_TIME BIGINT NULL,
PREV_FIRE_TIME BIGINT NULL,
PRIORITY INTEGER NULL,
TRIGGER_STATE VARCHAR(16) NOT NULL,
TRIGGER_TYPE VARCHAR(8) NOT NULL,
START_TIME BIGINT NOT NULL,
END_TIME BIGINT NULL,
CALENDAR_NAME VARCHAR(200) NULL,
MISFIRE_INSTR SMALLINT NULL,
JOB_DATA BYTEA NULL,
PRIMARY KEY (SCHED_NAME, TRIGGER_NAME, TRIGGER_GROUP),
FOREIGN KEY (SCHED_NAME, JOB_NAME, JOB_GROUP)
REFERENCES QRTZ_JOB_DETAILS (SCHED_NAME, JOB_NAME, JOB_GROUP)
);
CREATE TABLE QRTZ_SIMPLE_TRIGGERS
(
SCHED_NAME VARCHAR(120) NOT NULL,
TRIGGER_NAME VARCHAR(200) NOT NULL,
TRIGGER_GROUP VARCHAR(200) NOT NULL,
REPEAT_COUNT BIGINT NOT NULL,
REPEAT_INTERVAL BIGINT NOT NULL,
TIMES_TRIGGERED BIGINT NOT NULL,
PRIMARY KEY (SCHED_NAME, TRIGGER_NAME, TRIGGER_GROUP),
FOREIGN KEY (SCHED_NAME, TRIGGER_NAME, TRIGGER_GROUP)
REFERENCES QRTZ_TRIGGERS (SCHED_NAME, TRIGGER_NAME, TRIGGER_GROUP)
);
CREATE TABLE QRTZ_CRON_TRIGGERS
(
SCHED_NAME VARCHAR(120) NOT NULL,
TRIGGER_NAME VARCHAR(200) NOT NULL,
TRIGGER_GROUP VARCHAR(200) NOT NULL,
CRON_EXPRESSION VARCHAR(120) NOT NULL,
TIME_ZONE_ID VARCHAR(80),
PRIMARY KEY (SCHED_NAME, TRIGGER_NAME, TRIGGER_GROUP),
FOREIGN KEY (SCHED_NAME, TRIGGER_NAME, TRIGGER_GROUP)
REFERENCES QRTZ_TRIGGERS (SCHED_NAME, TRIGGER_NAME, TRIGGER_GROUP)
);
CREATE TABLE QRTZ_SIMPROP_TRIGGERS
(
SCHED_NAME VARCHAR(120) NOT NULL,
TRIGGER_NAME VARCHAR(200) NOT NULL,
TRIGGER_GROUP VARCHAR(200) NOT NULL,
STR_PROP_1 VARCHAR(512) NULL,
STR_PROP_2 VARCHAR(512) NULL,
STR_PROP_3 VARCHAR(512) NULL,
INT_PROP_1 INT NULL,
INT_PROP_2 INT NULL,
LONG_PROP_1 BIGINT NULL,
LONG_PROP_2 BIGINT NULL,
DEC_PROP_1 NUMERIC(13, 4) NULL,
DEC_PROP_2 NUMERIC(13, 4) NULL,
BOOL_PROP_1 BOOL NULL,
BOOL_PROP_2 BOOL NULL,
PRIMARY KEY (SCHED_NAME, TRIGGER_NAME, TRIGGER_GROUP),
FOREIGN KEY (SCHED_NAME, TRIGGER_NAME, TRIGGER_GROUP)
REFERENCES QRTZ_TRIGGERS (SCHED_NAME, TRIGGER_NAME, TRIGGER_GROUP)
);
CREATE TABLE QRTZ_BLOB_TRIGGERS
(
SCHED_NAME VARCHAR(120) NOT NULL,
TRIGGER_NAME VARCHAR(200) NOT NULL,
TRIGGER_GROUP VARCHAR(200) NOT NULL,
BLOB_DATA BYTEA NULL,
PRIMARY KEY (SCHED_NAME, TRIGGER_NAME, TRIGGER_GROUP),
FOREIGN KEY (SCHED_NAME, TRIGGER_NAME, TRIGGER_GROUP)
REFERENCES QRTZ_TRIGGERS (SCHED_NAME, TRIGGER_NAME, TRIGGER_GROUP)
);
CREATE TABLE QRTZ_CALENDARS
(
SCHED_NAME VARCHAR(120) NOT NULL,
CALENDAR_NAME VARCHAR(200) NOT NULL,
CALENDAR BYTEA NOT NULL,
PRIMARY KEY (SCHED_NAME, CALENDAR_NAME)
);
CREATE TABLE QRTZ_PAUSED_TRIGGER_GRPS
(
SCHED_NAME VARCHAR(120) NOT NULL,
TRIGGER_GROUP VARCHAR(200) NOT NULL,
PRIMARY KEY (SCHED_NAME, TRIGGER_GROUP)
);
CREATE TABLE QRTZ_FIRED_TRIGGERS
(
SCHED_NAME VARCHAR(120) NOT NULL,
ENTRY_ID VARCHAR(95) NOT NULL,
TRIGGER_NAME VARCHAR(200) NOT NULL,
TRIGGER_GROUP VARCHAR(200) NOT NULL,
INSTANCE_NAME VARCHAR(200) NOT NULL,
FIRED_TIME BIGINT NOT NULL,
SCHED_TIME BIGINT NOT NULL,
PRIORITY INTEGER NOT NULL,
STATE VARCHAR(16) NOT NULL,
JOB_NAME VARCHAR(200) NULL,
JOB_GROUP VARCHAR(200) NULL,
IS_NONCONCURRENT BOOL NULL,
REQUESTS_RECOVERY BOOL NULL,
PRIMARY KEY (SCHED_NAME, ENTRY_ID)
);
CREATE TABLE QRTZ_SCHEDULER_STATE
(
SCHED_NAME VARCHAR(120) NOT NULL,
INSTANCE_NAME VARCHAR(200) NOT NULL,
LAST_CHECKIN_TIME BIGINT NOT NULL,
CHECKIN_INTERVAL BIGINT NOT NULL,
PRIMARY KEY (SCHED_NAME, INSTANCE_NAME)
);
CREATE TABLE QRTZ_LOCKS
(
SCHED_NAME VARCHAR(120) NOT NULL,
LOCK_NAME VARCHAR(40) NOT NULL,
PRIMARY KEY (SCHED_NAME, LOCK_NAME)
);
CREATE INDEX IDX_QRTZ_J_REQ_RECOVERY
ON QRTZ_JOB_DETAILS (SCHED_NAME, REQUESTS_RECOVERY);
CREATE INDEX IDX_QRTZ_J_GRP
ON QRTZ_JOB_DETAILS (SCHED_NAME, JOB_GROUP);
CREATE INDEX IDX_QRTZ_T_J
ON QRTZ_TRIGGERS (SCHED_NAME, JOB_NAME, JOB_GROUP);
CREATE INDEX IDX_QRTZ_T_JG
ON QRTZ_TRIGGERS (SCHED_NAME, JOB_GROUP);
CREATE INDEX IDX_QRTZ_T_C
ON QRTZ_TRIGGERS (SCHED_NAME, CALENDAR_NAME);
CREATE INDEX IDX_QRTZ_T_G
ON QRTZ_TRIGGERS (SCHED_NAME, TRIGGER_GROUP);
CREATE INDEX IDX_QRTZ_T_STATE
ON QRTZ_TRIGGERS (SCHED_NAME, TRIGGER_STATE);
CREATE INDEX IDX_QRTZ_T_N_STATE
ON QRTZ_TRIGGERS (SCHED_NAME, TRIGGER_NAME, TRIGGER_GROUP, TRIGGER_STATE);
CREATE INDEX IDX_QRTZ_T_N_G_STATE
ON QRTZ_TRIGGERS (SCHED_NAME, TRIGGER_GROUP, TRIGGER_STATE);
CREATE INDEX IDX_QRTZ_T_NEXT_FIRE_TIME
ON QRTZ_TRIGGERS (SCHED_NAME, NEXT_FIRE_TIME);
CREATE INDEX IDX_QRTZ_T_NFT_ST
ON QRTZ_TRIGGERS (SCHED_NAME, TRIGGER_STATE, NEXT_FIRE_TIME);
CREATE INDEX IDX_QRTZ_T_NFT_MISFIRE
ON QRTZ_TRIGGERS (SCHED_NAME, MISFIRE_INSTR, NEXT_FIRE_TIME);
CREATE INDEX IDX_QRTZ_T_NFT_ST_MISFIRE
ON QRTZ_TRIGGERS (SCHED_NAME, MISFIRE_INSTR, NEXT_FIRE_TIME, TRIGGER_STATE);
CREATE INDEX IDX_QRTZ_T_NFT_ST_MISFIRE_GRP
ON QRTZ_TRIGGERS (SCHED_NAME, MISFIRE_INSTR, NEXT_FIRE_TIME, TRIGGER_GROUP, TRIGGER_STATE);
CREATE INDEX IDX_QRTZ_FT_TRIG_INST_NAME
ON QRTZ_FIRED_TRIGGERS (SCHED_NAME, INSTANCE_NAME);
CREATE INDEX IDX_QRTZ_FT_INST_JOB_REQ_RCVRY
ON QRTZ_FIRED_TRIGGERS (SCHED_NAME, INSTANCE_NAME, REQUESTS_RECOVERY);
CREATE INDEX IDX_QRTZ_FT_J_G
ON QRTZ_FIRED_TRIGGERS (SCHED_NAME, JOB_NAME, JOB_GROUP);
CREATE INDEX IDX_QRTZ_FT_JG
ON QRTZ_FIRED_TRIGGERS (SCHED_NAME, JOB_GROUP);
CREATE INDEX IDX_QRTZ_FT_T_G
ON QRTZ_FIRED_TRIGGERS (SCHED_NAME, TRIGGER_NAME, TRIGGER_GROUP);
CREATE INDEX IDX_QRTZ_FT_TG
ON QRTZ_FIRED_TRIGGERS (SCHED_NAME, TRIGGER_GROUP);

View File

@@ -0,0 +1,283 @@
package com.easyagents.scheduler.quartz;
import com.easyagents.scheduler.ConcurrencyPolicy;
import com.easyagents.scheduler.MisfirePolicy;
import com.easyagents.scheduler.OnceSchedulePlan;
import com.easyagents.scheduler.ScheduleDefinition;
import com.easyagents.scheduler.ScheduleErrorCode;
import com.easyagents.scheduler.ScheduleException;
import com.easyagents.scheduler.ScheduleFireContext;
import com.easyagents.scheduler.ScheduleHandler;
import com.easyagents.scheduler.ScheduleId;
import org.h2.jdbcx.JdbcDataSource;
import org.h2.tools.RunScript;
import org.junit.Test;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.nio.charset.StandardCharsets;
import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.Statement;
import java.time.Instant;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
/**
* Quartz JDBC JobStore、建表资源和外部 DataSource 所有权的集成测试。
*/
public class QuartzJdbcIntegrationTest {
/**
* 验证 H2 建表脚本、结构校验和 JDBC JobStore 实际触发链路。
*/
@Test
public void shouldRunWithCallerManagedDataSource() throws Exception {
JdbcDataSource dataSource = dataSource();
executeSchema(dataSource);
QuartzSchemaValidator.validate(dataSource, "QRTZ_");
CountDownLatch latch = new CountDownLatch(1);
QuartzSchedulerConfig config = jdbcConfig();
QuartzScheduleService service = QuartzSchedulerFactory.createJdbc(
dataSource,
config,
List.of(handler(latch)),
List.of()
);
try {
service.start();
service.create(new ScheduleDefinition(
new ScheduleId("jdbc", "once"),
"jdbc-handler",
new OnceSchedulePlan(Instant.now().plusMillis(500)),
MisfirePolicy.FIRE_ONCE_NOW,
ConcurrencyPolicy.DISALLOW,
true,
Map.of("source", "h2"),
"jdbc integration"
));
assertTrue("JDBC handler did not execute", latch.await(8, TimeUnit.SECONDS));
} finally {
service.close();
}
// 同一 JVM 使用相同 Scheduler Name 重建时覆盖已释放的内部 Provider 引用。
CountDownLatch restartedLatch = new CountDownLatch(1);
QuartzScheduleService restarted = QuartzSchedulerFactory.createJdbc(
dataSource,
config,
List.of(handler(restartedLatch)),
List.of()
);
try {
restarted.start();
assertTrue(restarted.get(new ScheduleId("jdbc", "once")).isPresent());
restarted.triggerNow(
new ScheduleId("jdbc", "once"),
"restart-trigger",
Map.of()
);
assertTrue(
"rebuilt scheduler did not trigger persisted job",
restartedLatch.await(5, TimeUnit.SECONDS)
);
} finally {
restarted.close();
}
// Provider 只释放内部强引用,调用方 DataSource 仍可继续使用。
try (
Connection connection = dataSource.getConnection();
Statement statement = connection.createStatement();
ResultSet resultSet = statement.executeQuery("SELECT COUNT(*) FROM QRTZ_JOB_DETAILS")
) {
assertTrue(resultSet.next());
assertEquals(1, resultSet.getInt(1));
}
}
/**
* 验证缺少 Quartz 表时结构校验拒绝启动。
*/
@Test
public void shouldRejectMissingSchema() {
try {
QuartzSchemaValidator.validate(dataSource(), "QRTZ_");
fail("expected schema validation failure");
} catch (ScheduleException exception) {
assertEquals(ScheduleErrorCode.SCHEMA_INVALID, exception.errorCode());
}
}
/**
* 验证关键列缺失时结构校验不会把不兼容结构误判为可用。
*/
@Test
public void shouldRejectMissingCriticalColumn() throws Exception {
JdbcDataSource dataSource = dataSource();
executeSchema(dataSource);
try (
Connection connection = dataSource.getConnection();
Statement statement = connection.createStatement()
) {
statement.execute("ALTER TABLE QRTZ_CRON_TRIGGERS DROP COLUMN CRON_EXPRESSION");
}
try {
QuartzSchemaValidator.validate(dataSource, "QRTZ_");
fail("expected critical column validation failure");
} catch (ScheduleException exception) {
assertEquals(ScheduleErrorCode.SCHEMA_INVALID, exception.errorCode());
assertTrue(exception.getMessage().contains("CRON_EXPRESSION"));
}
}
/**
* 验证未限定前缀只校验当前 Schema不会接受其他 Schema 的同名表。
*/
@Test
public void shouldRejectQuartzTablesOutsideCurrentSchema() throws Exception {
JdbcDataSource dataSource = dataSource();
try (
Connection connection = dataSource.getConnection();
Statement statement = connection.createStatement()
) {
statement.execute("CREATE SCHEMA OTHER");
statement.execute("SET SCHEMA OTHER");
executeSchema(connection);
}
try {
QuartzSchemaValidator.validate(dataSource, "QRTZ_");
fail("expected current schema validation failure");
} catch (ScheduleException exception) {
assertEquals(ScheduleErrorCode.SCHEMA_INVALID, exception.errorCode());
}
QuartzSchemaValidator.validate(dataSource, "OTHER.QRTZ_");
}
/**
* 验证 JDBC JobStore 能原子删除仍在执行立即触发的任务。
*/
@Test
public void shouldDeleteRunningScheduleAtomicallyInJdbcStore() throws Exception {
JdbcDataSource dataSource = dataSource();
executeSchema(dataSource);
CountDownLatch started = new CountDownLatch(1);
CountDownLatch release = new CountDownLatch(1);
CountDownLatch completed = new CountDownLatch(1);
ScheduleHandler blockingHandler = new ScheduleHandler() {
@Override
public String code() {
return "delete-handler";
}
@Override
public void execute(ScheduleFireContext context) throws InterruptedException {
started.countDown();
release.await();
completed.countDown();
}
};
QuartzScheduleService service = QuartzSchedulerFactory.createJdbc(
dataSource,
jdbcConfig(),
List.of(blockingHandler),
List.of()
);
ScheduleId scheduleId = new ScheduleId("jdbc", "delete-running");
try {
service.start();
service.create(new ScheduleDefinition(
scheduleId,
blockingHandler.code(),
new OnceSchedulePlan(Instant.parse("2099-01-01T00:00:00Z")),
MisfirePolicy.FIRE_ONCE_NOW,
ConcurrencyPolicy.DISALLOW,
true,
Map.of(),
"delete running"
));
service.triggerNow(scheduleId, "delete-running-1", Map.of());
assertTrue("JDBC handler did not start", started.await(5, TimeUnit.SECONDS));
assertTrue(service.delete(scheduleId));
assertTrue(service.get(scheduleId).isEmpty());
} finally {
release.countDown();
completed.await(5, TimeUnit.SECONDS);
service.close();
}
}
private static JdbcDataSource dataSource() {
JdbcDataSource dataSource = new JdbcDataSource();
dataSource.setURL("jdbc:h2:mem:scheduler-" + UUID.randomUUID() + ";DB_CLOSE_DELAY=-1");
dataSource.setUser("sa");
dataSource.setPassword("");
return dataSource;
}
private static void executeSchema(JdbcDataSource dataSource) throws Exception {
try (Connection connection = dataSource.getConnection()) {
executeSchema(connection);
}
}
/**
* 在指定连接的当前 Schema 中执行 H2 Quartz 建表资源。
*
* @param connection 目标数据库连接
* @throws Exception 资源读取或脚本执行失败时抛出
*/
private static void executeSchema(Connection connection) throws Exception {
InputStream stream = QuartzJdbcIntegrationTest.class
.getClassLoader()
.getResourceAsStream("quartz-schema/h2-2.5.2.sql");
if (stream == null) {
throw new IllegalStateException("H2 Quartz schema resource not found");
}
try (InputStreamReader reader = new InputStreamReader(stream, StandardCharsets.UTF_8)) {
RunScript.execute(connection, reader);
}
}
private static QuartzSchedulerConfig jdbcConfig() {
return new QuartzSchedulerConfig(
"jdbc-scheduler-" + UUID.randomUUID(),
"NON_CLUSTERED",
"QRTZ_",
QuartzSchedulerConfig.STANDARD_JDBC_DELEGATE,
false,
2,
Thread.NORM_PRIORITY,
15_000L,
1_000L,
true,
30_000L,
true
);
}
private static ScheduleHandler handler(CountDownLatch latch) {
return new ScheduleHandler() {
@Override
public String code() {
return "jdbc-handler";
}
@Override
public void execute(ScheduleFireContext context) {
latch.countDown();
}
};
}
}

View File

@@ -0,0 +1,94 @@
package com.easyagents.scheduler.quartz;
import com.easyagents.scheduler.ConcurrencyPolicy;
import com.easyagents.scheduler.MisfirePolicy;
import com.easyagents.scheduler.OnceSchedulePlan;
import com.easyagents.scheduler.ScheduleDefinition;
import com.easyagents.scheduler.ScheduleErrorCode;
import com.easyagents.scheduler.ScheduleException;
import com.easyagents.scheduler.ScheduleId;
import org.junit.Test;
import org.quartz.JobDataMap;
import java.nio.charset.StandardCharsets;
import java.time.Instant;
import java.util.Base64;
import java.util.Map;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertThrows;
/**
* {@link QuartzScheduleMapper} 持久字段严格性测试。
*/
public class QuartzScheduleMapperTest {
/**
* 验证立即触发数据保留 Misfire 监听所需的调度标识。
*/
@Test
public void shouldCarryScheduleIdentityForImmediateTrigger() {
ScheduleId scheduleId = new ScheduleId("mapper", "immediate");
JobDataMap data = QuartzScheduleMapper.immediateData(
scheduleId,
"invocation",
Map.of("key", "value")
);
assertEquals(scheduleId, QuartzScheduleMapper.toScheduleId(data));
}
/**
* 验证损坏的恢复布尔值不会静默解释为 false。
*/
@Test
public void shouldRejectCorruptedBooleanValue() {
JobDataMap data = QuartzScheduleMapper.toJobDetail(definition()).getJobDataMap();
data.put(QuartzScheduleMapper.KEY_RECOVER, "corrupted");
ScheduleException exception = assertThrows(
ScheduleException.class,
() -> QuartzScheduleMapper.toDefinition(data)
);
assertEquals(ScheduleErrorCode.PROVIDER_FAILURE, exception.errorCode());
}
/**
* 验证持久参数损坏会归类为 Provider 数据错误。
*/
@Test
public void shouldClassifyCorruptedPersistedParametersAsProviderFailure() {
JobDataMap data = QuartzScheduleMapper.toJobDetail(definition()).getJobDataMap();
String encodedKey = Base64.getUrlEncoder().withoutPadding().encodeToString(
"payload".getBytes(StandardCharsets.UTF_8)
);
data.put(
QuartzScheduleMapper.PARAMETER_PREFIX + encodedKey,
"x".repeat(ScheduleDefinition.MAX_PARAMETER_BYTES)
);
ScheduleException exception = assertThrows(
ScheduleException.class,
() -> QuartzScheduleMapper.toDefinition(data)
);
assertEquals(ScheduleErrorCode.PROVIDER_FAILURE, exception.errorCode());
}
/**
* 创建映射测试定义。
*
* @return 测试定义
*/
private static ScheduleDefinition definition() {
return new ScheduleDefinition(
new ScheduleId("mapper", "strict"),
"handler",
new OnceSchedulePlan(Instant.parse("2099-01-01T00:00:00Z")),
MisfirePolicy.FIRE_ONCE_NOW,
ConcurrencyPolicy.DISALLOW,
true,
Map.of(),
"strict mapper"
);
}
}

View File

@@ -0,0 +1,471 @@
package com.easyagents.scheduler.quartz;
import com.easyagents.scheduler.ConcurrencyPolicy;
import com.easyagents.scheduler.CronSchedulePlan;
import com.easyagents.scheduler.MisfirePolicy;
import com.easyagents.scheduler.OnceSchedulePlan;
import com.easyagents.scheduler.ScheduleDefinition;
import com.easyagents.scheduler.ScheduleErrorCode;
import com.easyagents.scheduler.ScheduleException;
import com.easyagents.scheduler.ScheduleFireContext;
import com.easyagents.scheduler.ScheduleHandler;
import com.easyagents.scheduler.ScheduleId;
import com.easyagents.scheduler.ScheduleStatus;
import org.junit.After;
import org.junit.Test;
import org.quartz.JobDetail;
import org.quartz.Scheduler;
import org.quartz.impl.StdSchedulerFactory;
import java.nio.charset.StandardCharsets;
import java.time.Instant;
import java.time.ZoneId;
import java.util.Base64;
import java.util.Map;
import java.util.Properties;
import java.util.UUID;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Consumer;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
/**
* {@link QuartzScheduleService} 的无数据库行为测试。
*/
public class QuartzScheduleServiceTest {
private QuartzScheduleService service;
/**
* 关闭每个测试创建的 Scheduler。
*/
@After
public void tearDown() {
if (service != null) {
service.close();
}
}
/**
* 验证创建幂等、冲突检测、暂停恢复、原子替换和删除。
*/
@Test
public void shouldManageScheduleLifecycle() throws Exception {
service = newRamService(context -> { });
ScheduleDefinition original = cronDefinition("handler", "0 0 0 1 1 ? 2099");
assertEquals(original, service.create(original).definition());
assertEquals(original, service.create(original).definition());
ScheduleDefinition conflicting = cronDefinition("handler", "0 0 0 2 1 ? 2099");
try {
service.create(conflicting);
fail("expected schedule conflict");
} catch (ScheduleException exception) {
assertEquals(ScheduleErrorCode.SCHEDULE_CONFLICT, exception.errorCode());
}
assertEquals(ScheduleStatus.PAUSED, service.pause(original.id()).status());
assertEquals(ScheduleStatus.SCHEDULED, service.resume(original.id()).status());
assertEquals(conflicting, service.replace(conflicting).definition());
assertTrue(service.delete(original.id()));
assertFalse(service.delete(original.id()));
}
/**
* 验证立即触发会覆盖同名基础参数并透传调用标识。
*/
@Test
public void shouldDispatchImmediateTriggerWithOverrideParameters() throws Exception {
CountDownLatch latch = new CountDownLatch(1);
AtomicReference<ScheduleFireContext> captured = new AtomicReference<>();
service = newRamService(context -> {
captured.set(context);
latch.countDown();
});
ScheduleDefinition definition = new ScheduleDefinition(
new ScheduleId("test", "immediate"),
"handler",
new OnceSchedulePlan(Instant.parse("2099-01-01T00:00:00Z")),
MisfirePolicy.FIRE_ONCE_NOW,
ConcurrencyPolicy.DISALLOW,
true,
Map.of("same", "base", "stable", "value"),
"immediate test"
);
service.create(definition);
service.triggerNow(definition.id(), "invoke-1", Map.of("same", "override"));
assertTrue("handler did not execute", latch.await(5, TimeUnit.SECONDS));
assertNotNull(captured.get());
assertEquals("invoke-1", captured.get().invocationId());
assertEquals("override", captured.get().parameters().get("same"));
assertEquals("value", captured.get().parameters().get("stable"));
}
/**
* 验证立即触发会在返回回执前校验基础参数与覆盖参数的合并结果。
*/
@Test
public void shouldRejectOversizedMergedImmediateParameters() throws Exception {
AtomicInteger executions = new AtomicInteger();
service = newRamService(context -> executions.incrementAndGet());
ScheduleDefinition definition = new ScheduleDefinition(
new ScheduleId("test", "oversized-immediate"),
"handler",
new OnceSchedulePlan(Instant.parse("2099-01-01T00:00:00Z")),
MisfirePolicy.FIRE_ONCE_NOW,
ConcurrencyPolicy.DISALLOW,
true,
Map.of("base", "x".repeat(9_000)),
"oversized immediate test"
);
service.create(definition);
try {
service.triggerNow(
definition.id(),
"oversized-invocation",
Map.of("override", "y".repeat(9_000))
);
fail("expected oversized merged parameters");
} catch (ScheduleException exception) {
assertEquals(ScheduleErrorCode.INVALID_DEFINITION, exception.errorCode());
}
assertEquals(0, executions.get());
}
/**
* 验证立即触发读取到损坏持久参数时返回 Provider 错误。
*/
@Test
public void shouldClassifyCorruptedPersistedParametersBeforeImmediateTrigger()
throws Exception {
service = newRamService(context -> { });
ScheduleDefinition definition = onceDefinition(
"corrupted-immediate",
MisfirePolicy.FIRE_ONCE_NOW,
ConcurrencyPolicy.DISALLOW,
Instant.parse("2099-01-01T00:00:00Z")
);
service.create(definition);
JobDetail job = service.quartzScheduler().getJobDetail(
QuartzScheduleMapper.jobKey(definition.id())
);
String encodedKey = Base64.getUrlEncoder().withoutPadding().encodeToString(
"payload".getBytes(StandardCharsets.UTF_8)
);
job.getJobDataMap().put(
QuartzScheduleMapper.PARAMETER_PREFIX + encodedKey,
"x".repeat(ScheduleDefinition.MAX_PARAMETER_BYTES)
);
service.quartzScheduler().addJob(job, true);
try {
service.triggerNow(definition.id(), "corrupted-invocation", Map.of());
fail("expected corrupted persisted parameters");
} catch (ScheduleException exception) {
assertEquals(ScheduleErrorCode.PROVIDER_FAILURE, exception.errorCode());
}
}
/**
* 验证预览不会落库且非法 Cron 返回稳定错误码。
*/
@Test
public void shouldPreviewFireTimesWithoutPersisting() throws Exception {
service = newRamService(context -> { });
assertEquals(
3,
service.nextFireTimes(
new CronSchedulePlan("0 0/5 * * * ?", ZoneId.of("Asia/Shanghai")),
3
).size()
);
try {
service.nextFireTimes(new CronSchedulePlan("bad cron", ZoneId.of("UTC")), 1);
fail("expected invalid cron");
} catch (ScheduleException exception) {
assertEquals(ScheduleErrorCode.INVALID_DEFINITION, exception.errorCode());
}
}
/**
* 验证 DISALLOW 策略会串行执行同一 Schedule 的立即触发。
*/
@Test
public void shouldDisallowConcurrentExecutionForSameSchedule() throws Exception {
CountDownLatch firstStarted = new CountDownLatch(1);
CountDownLatch release = new CountDownLatch(1);
CountDownLatch completed = new CountDownLatch(2);
AtomicInteger active = new AtomicInteger();
AtomicInteger maximum = new AtomicInteger();
service = newRamService(context -> {
int current = active.incrementAndGet();
maximum.accumulateAndGet(current, Math::max);
firstStarted.countDown();
await(release);
active.decrementAndGet();
completed.countDown();
});
ScheduleDefinition definition = onceDefinition(
"serial",
MisfirePolicy.FIRE_ONCE_NOW,
ConcurrencyPolicy.DISALLOW,
Instant.parse("2099-01-01T00:00:00Z")
);
service.create(definition);
service.triggerNow(definition.id(), "serial-1", Map.of());
assertTrue(firstStarted.await(5, TimeUnit.SECONDS));
service.triggerNow(definition.id(), "serial-2", Map.of());
Thread.sleep(250L);
assertEquals(1, maximum.get());
release.countDown();
assertTrue(completed.await(5, TimeUnit.SECONDS));
assertEquals(1, maximum.get());
}
/**
* 验证 ALLOW 策略允许同一 Schedule 按 Worker 容量并发执行。
*/
@Test
public void shouldAllowConcurrentExecutionForSameSchedule() throws Exception {
CountDownLatch bothStarted = new CountDownLatch(2);
CountDownLatch release = new CountDownLatch(1);
CountDownLatch completed = new CountDownLatch(2);
AtomicInteger active = new AtomicInteger();
AtomicInteger maximum = new AtomicInteger();
service = newRamService(context -> {
int current = active.incrementAndGet();
maximum.accumulateAndGet(current, Math::max);
bothStarted.countDown();
await(release);
active.decrementAndGet();
completed.countDown();
});
ScheduleDefinition definition = onceDefinition(
"parallel",
MisfirePolicy.FIRE_ONCE_NOW,
ConcurrencyPolicy.ALLOW,
Instant.parse("2099-01-01T00:00:00Z")
);
service.create(definition);
service.triggerNow(definition.id(), "parallel-1", Map.of());
service.triggerNow(definition.id(), "parallel-2", Map.of());
assertTrue("both handlers did not start", bothStarted.await(5, TimeUnit.SECONDS));
assertEquals(2, maximum.get());
release.countDown();
assertTrue(completed.await(5, TimeUnit.SECONDS));
}
/**
* 验证删除运行中任务不会与 Quartz 临时 Trigger 的完成清理发生竞态。
*/
@Test
public void shouldDeleteScheduleWhileImmediateTriggerIsRunning() throws Exception {
CountDownLatch started = new CountDownLatch(1);
CountDownLatch release = new CountDownLatch(1);
CountDownLatch completed = new CountDownLatch(1);
service = newRamService(context -> {
started.countDown();
await(release);
completed.countDown();
});
ScheduleDefinition definition = onceDefinition(
"delete-running",
MisfirePolicy.FIRE_ONCE_NOW,
ConcurrencyPolicy.DISALLOW,
Instant.parse("2099-01-01T00:00:00Z")
);
service.create(definition);
service.triggerNow(definition.id(), "delete-running-1", Map.of());
assertTrue(started.await(5, TimeUnit.SECONDS));
try {
assertTrue(service.delete(definition.id()));
assertTrue(service.get(definition.id()).isEmpty());
} finally {
release.countDown();
}
assertTrue(completed.await(5, TimeUnit.SECONDS));
}
/**
* 验证一次性任务的 SKIP 与 FIRE_ONCE_NOW Misfire 映射。
*/
@Test
public void shouldApplyOnceMisfirePolicies() throws Exception {
CountDownLatch fired = new CountDownLatch(1);
AtomicInteger executions = new AtomicInteger();
service = newRamService(context -> {
executions.incrementAndGet();
if (context.scheduleId().name().equals("fire-misfire")) {
fired.countDown();
}
});
Instant missedAt = Instant.now().minusSeconds(2);
service.create(onceDefinition(
"skip-misfire",
MisfirePolicy.SKIP,
ConcurrencyPolicy.DISALLOW,
missedAt
));
service.create(onceDefinition(
"fire-misfire",
MisfirePolicy.FIRE_ONCE_NOW,
ConcurrencyPolicy.DISALLOW,
missedAt
));
assertTrue("FIRE_ONCE_NOW did not execute", fired.await(5, TimeUnit.SECONDS));
Thread.sleep(250L);
assertEquals(1, executions.get());
assertEquals(
ScheduleStatus.COMPLETE,
service.get(new ScheduleId("test", "skip-misfire")).orElseThrow().status()
);
}
/**
* 验证关闭等待超过上限后会有界返回并中断内部 Handler 线程。
*/
@Test
public void shouldBoundShutdownWaitAndInterruptHandler() throws Exception {
CountDownLatch started = new CountDownLatch(1);
CountDownLatch interrupted = new CountDownLatch(1);
service = newRamService(context -> {
started.countDown();
try {
Thread.sleep(60_000L);
} catch (InterruptedException exception) {
interrupted.countDown();
Thread.currentThread().interrupt();
}
}, 200L);
ScheduleDefinition definition = onceDefinition(
"bounded-close",
MisfirePolicy.FIRE_ONCE_NOW,
ConcurrencyPolicy.DISALLOW,
Instant.parse("2099-01-01T00:00:00Z")
);
service.create(definition);
service.triggerNow(definition.id(), "bounded-close-1", Map.of());
assertTrue(started.await(5, TimeUnit.SECONDS));
long startedAt = System.nanoTime();
try {
service.close();
fail("expected bounded shutdown timeout");
} catch (ScheduleException exception) {
assertEquals(ScheduleErrorCode.PROVIDER_FAILURE, exception.errorCode());
}
long elapsedMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startedAt);
assertTrue("shutdown exceeded bound: " + elapsedMillis, elapsedMillis < 2_000L);
assertTrue("handler thread was not interrupted", interrupted.await(2, TimeUnit.SECONDS));
}
private QuartzScheduleService newRamService(Consumer<ScheduleFireContext> handler)
throws Exception {
return newRamService(handler, 30_000L);
}
private QuartzScheduleService newRamService(
Consumer<ScheduleFireContext> handler,
long shutdownWaitTimeoutMillis
) throws Exception {
Properties properties = new Properties();
properties.setProperty(
"org.quartz.scheduler.instanceName",
"ram-scheduler-" + UUID.randomUUID()
);
properties.setProperty("org.quartz.scheduler.instanceId", "NON_CLUSTERED");
properties.setProperty("org.quartz.scheduler.interruptJobsOnShutdown", "true");
properties.setProperty("org.quartz.threadPool.class", "org.quartz.simpl.SimpleThreadPool");
properties.setProperty("org.quartz.threadPool.threadCount", "2");
properties.setProperty("org.quartz.jobStore.class", "org.quartz.simpl.RAMJobStore");
properties.setProperty("org.quartz.jobStore.misfireThreshold", "100");
Scheduler scheduler = new StdSchedulerFactory(properties).getScheduler();
QuartzScheduleService result = QuartzSchedulerFactory.attach(
scheduler,
true,
shutdownWaitTimeoutMillis,
java.util.List.of(new ScheduleHandler() {
@Override
public String code() {
return "handler";
}
@Override
public void execute(ScheduleFireContext context) {
handler.accept(context);
}
}),
java.util.List.of()
);
result.start();
assertSame(scheduler, result.quartzScheduler());
return result;
}
private static ScheduleDefinition cronDefinition(String handlerCode, String expression) {
return new ScheduleDefinition(
new ScheduleId("test", "lifecycle"),
handlerCode,
new CronSchedulePlan(expression, ZoneId.of("UTC")),
MisfirePolicy.SKIP,
ConcurrencyPolicy.DISALLOW,
true,
Map.of("key", "value"),
"lifecycle test"
);
}
private static ScheduleDefinition onceDefinition(
String name,
MisfirePolicy misfirePolicy,
ConcurrencyPolicy concurrencyPolicy,
Instant fireAt
) {
return new ScheduleDefinition(
new ScheduleId("test", name),
"handler",
new OnceSchedulePlan(fireAt),
misfirePolicy,
concurrencyPolicy,
true,
Map.of(),
name
);
}
private static void await(CountDownLatch latch) {
boolean interrupted = false;
try {
while (true) {
try {
latch.await();
return;
} catch (InterruptedException exception) {
interrupted = true;
}
}
} finally {
if (interrupted) {
Thread.currentThread().interrupt();
}
}
}
}