diff --git a/easy-agents-scheduler/easy-agents-scheduler-core/src/main/java/com/easyagents/scheduler/ScheduleRefireException.java b/easy-agents-scheduler/easy-agents-scheduler-core/src/main/java/com/easyagents/scheduler/ScheduleRefireException.java
new file mode 100644
index 0000000..01ef029
--- /dev/null
+++ b/easy-agents-scheduler/easy-agents-scheduler-core/src/main/java/com/easyagents/scheduler/ScheduleRefireException.java
@@ -0,0 +1,53 @@
+package com.easyagents.scheduler;
+
+import java.time.Duration;
+
+/**
+ * 请求调度提供方持久化重试同一次逻辑触发的异常。
+ *
+ * 仅用于尚未产生业务副作用、且调用方必须保证至少一次登记的短 Handler。
+ * {@code maxRefires=0} 表示不限制持久化重试次数；Provider 正常情况下应释放当前执行线程，
+ * 并通过持久化的延迟触发保留原 fire 上下文。仅当 JobStore 无法写入持久重试时，
+ * Provider 可按相同退避暂时占用当前线程并原地重试，以避免正常完成造成 fire 丢失。
+ */
+public class ScheduleRefireException extends RuntimeException {
+
+ private static final int DEFAULT_MAX_REFIRES = 0;
+ private static final Duration DEFAULT_BASE_DELAY = Duration.ofMillis(250);
+ private final int maxRefires;
+ private final Duration baseDelay;
+
+ public ScheduleRefireException(String message, Throwable cause) {
+ this(message, cause, DEFAULT_MAX_REFIRES, DEFAULT_BASE_DELAY);
+ }
+
+ public ScheduleRefireException(String message, Throwable cause,
+ int maxRefires, Duration baseDelay) {
+ super(requireMessage(message), cause);
+ if (maxRefires < 0) throw new IllegalArgumentException("maxRefires must not be negative");
+ if (baseDelay == null || baseDelay.isZero() || baseDelay.isNegative()) {
+ throw new IllegalArgumentException("baseDelay must be positive");
+ }
+ this.maxRefires = maxRefires;
+ this.baseDelay = baseDelay;
+ }
+
+ /**
+ * @return 最大重新执行次数，0 表示不限制
+ */
+ public int maxRefires() {
+ return maxRefires;
+ }
+
+ public Duration delayFor(int refireNumber) {
+ int shift = Math.min(4, Math.max(0, refireNumber - 1));
+ return baseDelay.multipliedBy(1L << shift);
+ }
+
+ private static String requireMessage(String message) {
+ if (message == null || message.isBlank()) {
+ throw new IllegalArgumentException("message must not be blank");
+ }
+ return message;
+ }
+}
diff --git a/easy-agents-scheduler/easy-agents-scheduler-quartz/src/main/java/com/easyagents/scheduler/quartz/AbstractDispatchJob.java b/easy-agents-scheduler/easy-agents-scheduler-quartz/src/main/java/com/easyagents/scheduler/quartz/AbstractDispatchJob.java
index 2f73903..22e8c00 100644
--- a/easy-agents-scheduler/easy-agents-scheduler-quartz/src/main/java/com/easyagents/scheduler/quartz/AbstractDispatchJob.java
+++ b/easy-agents-scheduler/easy-agents-scheduler-quartz/src/main/java/com/easyagents/scheduler/quartz/AbstractDispatchJob.java
@@ -29,6 +29,9 @@ abstract class AbstractDispatchJob implements InterruptableJob {
throw new JobExecutionException("easy-agents scheduler runtime is not available");
}
quartzRuntime.execute(context);
+ } catch (JobExecutionException exception) {
+ // JobExecutionException 继承 SchedulerException，必须保留 refire 等控制语义。
+ throw exception;
} catch (SchedulerException exception) {
throw new JobExecutionException("failed to access scheduler runtime", exception, false);
} finally {
diff --git a/easy-agents-scheduler/easy-agents-scheduler-quartz/src/main/java/com/easyagents/scheduler/quartz/QuartzRuntime.java b/easy-agents-scheduler/easy-agents-scheduler-quartz/src/main/java/com/easyagents/scheduler/quartz/QuartzRuntime.java
index 2d5b30d..c7e3081 100644
--- a/easy-agents-scheduler/easy-agents-scheduler-quartz/src/main/java/com/easyagents/scheduler/quartz/QuartzRuntime.java
+++ b/easy-agents-scheduler/easy-agents-scheduler-quartz/src/main/java/com/easyagents/scheduler/quartz/QuartzRuntime.java
@@ -7,9 +7,16 @@ import com.easyagents.scheduler.ScheduleExecutionListener;
import com.easyagents.scheduler.ScheduleFireContext;
import com.easyagents.scheduler.ScheduleHandler;
import com.easyagents.scheduler.ScheduleId;
+import com.easyagents.scheduler.ScheduleRefireException;
import org.quartz.JobDataMap;
import org.quartz.JobExecutionContext;
import org.quartz.JobExecutionException;
+import org.quartz.ObjectAlreadyExistsException;
+import org.quartz.SchedulerException;
+import org.quartz.SimpleScheduleBuilder;
+import org.quartz.Trigger;
+import org.quartz.TriggerBuilder;
+import org.quartz.TriggerKey;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -19,6 +26,10 @@ import java.util.Collection;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
+import java.nio.charset.StandardCharsets;
+import java.util.Date;
+import java.util.UUID;
+import java.util.concurrent.locks.LockSupport;
/**
* 当前 Scheduler 节点的 Handler 和监听器运行时。
@@ -55,15 +66,27 @@ final class QuartzRuntime {
void execute(JobExecutionContext quartzContext) throws JobExecutionException {
JobDataMap data = quartzContext.getMergedJobDataMap();
ScheduleDefinition definition = QuartzScheduleMapper.toDefinition(data);
- Instant actualFireTime = toInstant(quartzContext.getFireTime(), Instant.now());
+ Instant observedActualFireTime = toInstant(quartzContext.getFireTime(), Instant.now());
+ Instant scheduledFireTime = instantValue(
+ data,
+ QuartzScheduleMapper.KEY_RETRY_SCHEDULED_FIRE_TIME,
+ toInstant(quartzContext.getScheduledFireTime(), observedActualFireTime)
+ );
+ Instant actualFireTime = instantValue(
+ data,
+ QuartzScheduleMapper.KEY_RETRY_ACTUAL_FIRE_TIME,
+ observedActualFireTime
+ );
ScheduleFireContext context = new ScheduleFireContext(
definition.id(),
definition.handlerCode(),
- toInstant(quartzContext.getScheduledFireTime(), actualFireTime),
+ scheduledFireTime,
actualFireTime,
- quartzContext.getFireInstanceId(),
+ stringValue(data, QuartzScheduleMapper.KEY_RETRY_FIRE_INSTANCE_ID,
+ quartzContext.getFireInstanceId()),
stringValue(data, QuartzScheduleMapper.KEY_INVOCATION),
- quartzContext.isRecovering(),
+ booleanValue(data, QuartzScheduleMapper.KEY_RETRY_RECOVERING,
+ quartzContext.isRecovering()),
definition.parameters()
);
ScheduleHandler handler = handlers.get(definition.handlerCode());
@@ -82,10 +105,80 @@ final class QuartzRuntime {
notifySucceeded(context, elapsed(startedAt));
} catch (Exception exception) {
notifyFailed(context, elapsed(startedAt), exception);
+ if (exception instanceof ScheduleRefireException retry) {
+ int attempt = Math.max(
+ intValue(data, QuartzScheduleMapper.KEY_RETRY_ATTEMPT, 0),
+ quartzContext.getRefireCount()
+ );
+ if (retry.maxRefires() == 0 || attempt < retry.maxRefires()) {
+ try {
+ persistRetry(quartzContext, context, retry, attempt + 1);
+ return;
+ } catch (SchedulerException retryFailure) {
+ exception.addSuppressed(retryFailure);
+ log.error("Failed to persist schedule handler retry: scheduleId={}, attempt={}",
+ context.scheduleId(), attempt + 1, retryFailure);
+ if (canRefireInPlace(quartzContext, retry, attempt + 1)) {
+ // JobStore 暂时不可写时，正常完成原 fire 会造成登记丢失。
+ // 仅在这条降级路径短时占用当前 worker；一旦持久重试落库即释放。
+ throw new JobExecutionException(exception, true);
+ }
+ }
+ } else {
+ log.error("Schedule handler retry limit exhausted: scheduleId={}, retries={}",
+ context.scheduleId(), retry.maxRefires(), exception);
+ }
+ }
+ // 不在 Quartz worker 内无限 refire。持久化重试失败时保留原异常，
+ // 由 requestRecovery 和集群故障恢复处理未完成的 fired trigger。
throw new JobExecutionException(exception, false);
}
}
+ private static boolean canRefireInPlace(
+ JobExecutionContext context,
+ ScheduleRefireException retry,
+ int attempt
+ ) {
+ LockSupport.parkNanos(retry.delayFor(attempt).toNanos());
+ if (Thread.currentThread().isInterrupted()) return false;
+ try {
+ return !context.getScheduler().isShutdown();
+ } catch (SchedulerException exception) {
+ return false;
+ }
+ }
+
+ private static void persistRetry(
+ JobExecutionContext quartzContext,
+ ScheduleFireContext context,
+ ScheduleRefireException retry,
+ int attempt
+ ) throws SchedulerException {
+ String source = context.scheduleId() + "|" + context.fireInstanceId() + "|"
+ + String.valueOf(context.invocationId()) + "|" + attempt;
+ String retryName = "retry-" + UUID.nameUUIDFromBytes(
+ source.getBytes(StandardCharsets.UTF_8));
+ TriggerKey retryKey = new TriggerKey(
+ retryName,
+ QuartzScheduleMapper.GROUP_PREFIX + "retry." + context.scheduleId().namespace()
+ );
+ Trigger trigger = TriggerBuilder.newTrigger()
+ .withIdentity(retryKey)
+ .forJob(quartzContext.getJobDetail().getKey())
+ .usingJobData(QuartzScheduleMapper.retryData(context, attempt))
+ .startAt(Date.from(Instant.now().plus(retry.delayFor(attempt))))
+ .withSchedule(SimpleScheduleBuilder.simpleSchedule()
+ .withRepeatCount(0)
+ .withMisfireHandlingInstructionFireNow())
+ .build();
+ try {
+ quartzContext.getScheduler().scheduleJob(trigger);
+ } catch (ObjectAlreadyExistsException ignored) {
+ // 同一原始 fire/attempt 的确定性 key 已落库，即视为持久化成功。
+ }
+ }
+
/**
* 向监听器发布 Misfire 事件。
*
@@ -175,4 +268,24 @@ final class QuartzRuntime {
Object value = data.get(key);
return value == null ? null : value.toString();
}
+
+ private static String stringValue(JobDataMap data, String key, String fallback) {
+ String value = stringValue(data, key);
+ return value == null ? fallback : value;
+ }
+
+ private static Instant instantValue(JobDataMap data, String key, Instant fallback) {
+ String value = stringValue(data, key);
+ return value == null ? fallback : Instant.ofEpochMilli(Long.parseLong(value));
+ }
+
+ private static boolean booleanValue(JobDataMap data, String key, boolean fallback) {
+ String value = stringValue(data, key);
+ return value == null ? fallback : Boolean.parseBoolean(value);
+ }
+
+ private static int intValue(JobDataMap data, String key, int fallback) {
+ String value = stringValue(data, key);
+ return value == null ? fallback : Integer.parseInt(value);
+ }
}
diff --git a/easy-agents-scheduler/easy-agents-scheduler-quartz/src/main/java/com/easyagents/scheduler/quartz/QuartzScheduleMapper.java b/easy-agents-scheduler/easy-agents-scheduler-quartz/src/main/java/com/easyagents/scheduler/quartz/QuartzScheduleMapper.java
index 9a59b71..f6666ae 100644
--- a/easy-agents-scheduler/easy-agents-scheduler-quartz/src/main/java/com/easyagents/scheduler/quartz/QuartzScheduleMapper.java
+++ b/easy-agents-scheduler/easy-agents-scheduler-quartz/src/main/java/com/easyagents/scheduler/quartz/QuartzScheduleMapper.java
@@ -7,6 +7,7 @@ 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.ScheduleId;
import com.easyagents.scheduler.SchedulePlan;
import org.quartz.CronScheduleBuilder;
@@ -47,6 +48,11 @@ final class QuartzScheduleMapper {
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 KEY_RETRY_ATTEMPT = "ea.retry.attempt";
+ static final String KEY_RETRY_SCHEDULED_FIRE_TIME = "ea.retry.scheduledFireTime";
+ static final String KEY_RETRY_ACTUAL_FIRE_TIME = "ea.retry.actualFireTime";
+ static final String KEY_RETRY_FIRE_INSTANCE_ID = "ea.retry.fireInstanceId";
+ static final String KEY_RETRY_RECOVERING = "ea.retry.recovering";
static final String PARAMETER_PREFIX = "ea.parameter.";
static final String IMMEDIATE_PARAMETER_PREFIX = "ea.immediateParameter.";
static final String GROUP_PREFIX = "ea.scheduler.";
@@ -153,6 +159,30 @@ final class QuartzScheduleMapper {
return data;
}
+ /**
+ * 固化持久化重试所需的原始 fire 上下文。
+ */
+ static JobDataMap retryData(ScheduleFireContext context, int attempt) {
+ JobDataMap data = identityData(context.scheduleId());
+ // Trigger 数据覆盖 JobDetail；任务定义被 replace 后，既有 fire 的重试仍须
+ // 派发给首次触发时的 Handler，而不是意外切换到新 Handler。
+ data.put(KEY_HANDLER, context.handlerCode());
+ data.put(KEY_RETRY_ATTEMPT, Integer.toString(attempt));
+ data.put(KEY_RETRY_SCHEDULED_FIRE_TIME,
+ Long.toString(context.scheduledFireTime().toEpochMilli()));
+ data.put(KEY_RETRY_ACTUAL_FIRE_TIME,
+ Long.toString(context.actualFireTime().toEpochMilli()));
+ data.put(KEY_RETRY_FIRE_INSTANCE_ID, context.fireInstanceId());
+ data.put(KEY_RETRY_RECOVERING, Boolean.toString(context.recovering()));
+ if (context.invocationId() != null) {
+ data.put(KEY_INVOCATION, context.invocationId());
+ }
+ // 重试必须沿用首次 fire 的参数快照，不能读取期间被替换的新定义参数。
+ data.put(KEY_IMMEDIATE_PARAMETER_SNAPSHOT, Boolean.TRUE.toString());
+ putParameters(data, context.parameters(), IMMEDIATE_PARAMETER_PREFIX);
+ return data;
+ }
+
/**
* 从持久 JobData 恢复公共定义。
*
diff --git a/easy-agents-scheduler/easy-agents-scheduler-quartz/src/main/java/com/easyagents/scheduler/quartz/QuartzScheduleService.java b/easy-agents-scheduler/easy-agents-scheduler-quartz/src/main/java/com/easyagents/scheduler/quartz/QuartzScheduleService.java
index a47f42c..078dd37 100644
--- a/easy-agents-scheduler/easy-agents-scheduler-quartz/src/main/java/com/easyagents/scheduler/quartz/QuartzScheduleService.java
+++ b/easy-agents-scheduler/easy-agents-scheduler-quartz/src/main/java/com/easyagents/scheduler/quartz/QuartzScheduleService.java
@@ -335,7 +335,8 @@ public final class QuartzScheduleService implements ScheduleService, AutoCloseab
}
try {
if (!scheduler.isShutdown()) {
- // false 保证 close 本身有界；Quartz 会向内部 InterruptableJob 发送中断。
+ // Factory 启用 interruptJobsOnShutdown，Quartz 会先中断内部
+ // InterruptableJob；false 只表示不再无界等待忽略中断的 Handler。
scheduler.shutdown(false);
}
} catch (SchedulerException exception) {
diff --git a/easy-agents-scheduler/easy-agents-scheduler-quartz/src/main/java/com/easyagents/scheduler/quartz/QuartzSchedulerConfig.java b/easy-agents-scheduler/easy-agents-scheduler-quartz/src/main/java/com/easyagents/scheduler/quartz/QuartzSchedulerConfig.java
index 070f38f..35c9f48 100644
--- a/easy-agents-scheduler/easy-agents-scheduler-quartz/src/main/java/com/easyagents/scheduler/quartz/QuartzSchedulerConfig.java
+++ b/easy-agents-scheduler/easy-agents-scheduler-quartz/src/main/java/com/easyagents/scheduler/quartz/QuartzSchedulerConfig.java
@@ -10,6 +10,8 @@ package com.easyagents.scheduler.quartz;
* @param clustered 是否启用 JDBC 集群
* @param threadCount Quartz Worker 线程数
* @param threadPriority Quartz Worker 线程优先级
+ * @param batchTriggerAcquisitionMaxCount 单次批量获取 Trigger 的最大数量
+ * @param batchTriggerAcquisitionFireAheadTimeWindowMillis 可提前纳入批量的时间窗口，单位毫秒
* @param clusterCheckinIntervalMillis 集群心跳间隔，单位毫秒
* @param misfireThresholdMillis Misfire 判定阈值，单位毫秒
* @param waitForJobsToCompleteOnShutdown 关闭时是否等待运行中任务完成
@@ -24,6 +26,8 @@ public record QuartzSchedulerConfig(
boolean clustered,
int threadCount,
int threadPriority,
+ int batchTriggerAcquisitionMaxCount,
+ long batchTriggerAcquisitionFireAheadTimeWindowMillis,
long clusterCheckinIntervalMillis,
long misfireThresholdMillis,
boolean waitForJobsToCompleteOnShutdown,
@@ -55,6 +59,18 @@ public record QuartzSchedulerConfig(
if (threadPriority < Thread.MIN_PRIORITY || threadPriority > Thread.MAX_PRIORITY) {
throw new IllegalArgumentException("threadPriority must be between 1 and 10");
}
+ if (batchTriggerAcquisitionMaxCount < 1
+ || batchTriggerAcquisitionMaxCount > threadCount) {
+ throw new IllegalArgumentException(
+ "batchTriggerAcquisitionMaxCount must be between 1 and threadCount"
+ );
+ }
+ if (batchTriggerAcquisitionFireAheadTimeWindowMillis < 0
+ || batchTriggerAcquisitionFireAheadTimeWindowMillis > 60_000L) {
+ throw new IllegalArgumentException(
+ "batchTriggerAcquisitionFireAheadTimeWindowMillis must be between 0 and 60000"
+ );
+ }
if (clusterCheckinIntervalMillis < 1000) {
throw new IllegalArgumentException(
"clusterCheckinIntervalMillis must be at least 1000"
@@ -85,6 +101,8 @@ public record QuartzSchedulerConfig(
true,
8,
Thread.NORM_PRIORITY,
+ 1,
+ 0L,
15_000L,
60_000L,
true,
diff --git a/easy-agents-scheduler/easy-agents-scheduler-quartz/src/main/java/com/easyagents/scheduler/quartz/QuartzSchedulerFactory.java b/easy-agents-scheduler/easy-agents-scheduler-quartz/src/main/java/com/easyagents/scheduler/quartz/QuartzSchedulerFactory.java
index 57c206f..7460ebd 100644
--- a/easy-agents-scheduler/easy-agents-scheduler-quartz/src/main/java/com/easyagents/scheduler/quartz/QuartzSchedulerFactory.java
+++ b/easy-agents-scheduler/easy-agents-scheduler-quartz/src/main/java/com/easyagents/scheduler/quartz/QuartzSchedulerFactory.java
@@ -154,7 +154,7 @@ public final class QuartzSchedulerFactory {
* @param dataSourceName Quartz 内部 DataSource 名称
* @return Quartz 属性
*/
- private static Properties properties(
+ static Properties properties(
QuartzSchedulerConfig config,
String dataSourceName
) {
@@ -173,6 +173,14 @@ public final class QuartzSchedulerFactory {
"org.quartz.threadPool.threadPriority",
Integer.toString(config.threadPriority())
);
+ properties.setProperty(
+ "org.quartz.scheduler.batchTriggerAcquisitionMaxCount",
+ Integer.toString(config.batchTriggerAcquisitionMaxCount())
+ );
+ properties.setProperty(
+ "org.quartz.scheduler.batchTriggerAcquisitionFireAheadTimeWindow",
+ Long.toString(config.batchTriggerAcquisitionFireAheadTimeWindowMillis())
+ );
properties.setProperty(
"org.quartz.jobStore.class",
"org.quartz.impl.jdbcjobstore.JobStoreTX"
@@ -182,6 +190,10 @@ public final class QuartzSchedulerFactory {
config.driverDelegateClass()
);
properties.setProperty("org.quartz.jobStore.useProperties", "true");
+ properties.setProperty(
+ "org.quartz.jobStore.acquireTriggersWithinLock",
+ Boolean.toString(config.batchTriggerAcquisitionMaxCount() > 1)
+ );
properties.setProperty("org.quartz.jobStore.dataSource", dataSourceName);
properties.setProperty("org.quartz.jobStore.tablePrefix", config.tablePrefix());
properties.setProperty(
diff --git a/easy-agents-scheduler/easy-agents-scheduler-quartz/src/test/java/com/easyagents/scheduler/quartz/QuartzJdbcIntegrationTest.java b/easy-agents-scheduler/easy-agents-scheduler-quartz/src/test/java/com/easyagents/scheduler/quartz/QuartzJdbcIntegrationTest.java
index c7d9fac..6b90107 100644
--- a/easy-agents-scheduler/easy-agents-scheduler-quartz/src/test/java/com/easyagents/scheduler/quartz/QuartzJdbcIntegrationTest.java
+++ b/easy-agents-scheduler/easy-agents-scheduler-quartz/src/test/java/com/easyagents/scheduler/quartz/QuartzJdbcIntegrationTest.java
@@ -9,6 +9,7 @@ import com.easyagents.scheduler.ScheduleException;
import com.easyagents.scheduler.ScheduleFireContext;
import com.easyagents.scheduler.ScheduleHandler;
import com.easyagents.scheduler.ScheduleId;
+import com.easyagents.scheduler.ScheduleRefireException;
import org.h2.jdbcx.JdbcDataSource;
import org.h2.tools.RunScript;
import org.junit.Test;
@@ -20,9 +21,11 @@ import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.Statement;
import java.time.Instant;
+import java.time.Duration;
import java.util.List;
import java.util.Map;
import java.util.UUID;
+import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
@@ -218,6 +221,83 @@ public class QuartzJdbcIntegrationTest {
}
}
+ /**
+ * 验证延迟重试先落入 JDBC JobStore，关闭并重建节点后仍使用原 fire 上下文执行。
+ */
+ @Test
+ public void shouldRecoverPersistedRetryAfterSchedulerRestart() throws Exception {
+ JdbcDataSource dataSource = dataSource();
+ executeSchema(dataSource);
+ QuartzSchedulerConfig config = jdbcConfig();
+ CountDownLatch firstFailed = new CountDownLatch(1);
+ CountDownLatch retryCompleted = new CountDownLatch(1);
+ CopyOnWriteArrayList contexts = new CopyOnWriteArrayList<>();
+ ScheduleHandler retryingHandler = new ScheduleHandler() {
+ @Override
+ public String code() {
+ return "persistent-retry-handler";
+ }
+
+ @Override
+ public void execute(ScheduleFireContext context) {
+ contexts.add(context);
+ if (contexts.size() == 1) {
+ firstFailed.countDown();
+ throw new ScheduleRefireException("temporary database failure",
+ new IllegalStateException("unavailable"), 2,
+ Duration.ofSeconds(2));
+ }
+ retryCompleted.countDown();
+ }
+ };
+ ScheduleId scheduleId = new ScheduleId("jdbc", "persistent-retry");
+ QuartzScheduleService first = QuartzSchedulerFactory.createJdbc(
+ dataSource, config, List.of(retryingHandler), List.of());
+ first.start();
+ first.create(new ScheduleDefinition(
+ scheduleId,
+ retryingHandler.code(),
+ new OnceSchedulePlan(Instant.parse("2099-01-01T00:00:00Z")),
+ MisfirePolicy.FIRE_ONCE_NOW,
+ ConcurrencyPolicy.DISALLOW,
+ true,
+ Map.of("snapshot", "original"),
+ "persistent retry"
+ ));
+ first.triggerNow(scheduleId, "persistent-invocation", Map.of());
+ assertTrue(firstFailed.await(5, TimeUnit.SECONDS));
+ awaitRetryTrigger(dataSource);
+ first.close();
+
+ QuartzScheduleService restarted = QuartzSchedulerFactory.createJdbc(
+ dataSource, config, List.of(retryingHandler), List.of());
+ try {
+ restarted.start();
+ assertTrue("persisted retry did not execute after restart",
+ retryCompleted.await(8, TimeUnit.SECONDS));
+ assertEquals(contexts.get(0).scheduledFireTime(), contexts.get(1).scheduledFireTime());
+ assertEquals(contexts.get(0).actualFireTime(), contexts.get(1).actualFireTime());
+ assertEquals(contexts.get(0).fireInstanceId(), contexts.get(1).fireInstanceId());
+ assertEquals("persistent-invocation", contexts.get(1).invocationId());
+ } finally {
+ restarted.close();
+ }
+ }
+
+ private static void awaitRetryTrigger(JdbcDataSource dataSource) throws Exception {
+ long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5);
+ do {
+ try (Connection connection = dataSource.getConnection();
+ Statement statement = connection.createStatement();
+ ResultSet result = statement.executeQuery(
+ "SELECT COUNT(*) FROM QRTZ_TRIGGERS WHERE TRIGGER_GROUP LIKE 'ea.scheduler.retry.%'")) {
+ if (result.next() && result.getInt(1) > 0) return;
+ }
+ Thread.sleep(25L);
+ } while (System.nanoTime() < deadline);
+ fail("persistent retry trigger was not stored");
+ }
+
private static JdbcDataSource dataSource() {
JdbcDataSource dataSource = new JdbcDataSource();
dataSource.setURL("jdbc:h2:mem:scheduler-" + UUID.randomUUID() + ";DB_CLOSE_DELAY=-1");
@@ -259,6 +339,8 @@ public class QuartzJdbcIntegrationTest {
false,
2,
Thread.NORM_PRIORITY,
+ 1,
+ 0L,
15_000L,
1_000L,
true,
diff --git a/easy-agents-scheduler/easy-agents-scheduler-quartz/src/test/java/com/easyagents/scheduler/quartz/QuartzRuntimeTest.java b/easy-agents-scheduler/easy-agents-scheduler-quartz/src/test/java/com/easyagents/scheduler/quartz/QuartzRuntimeTest.java
new file mode 100644
index 0000000..486980e
--- /dev/null
+++ b/easy-agents-scheduler/easy-agents-scheduler-quartz/src/test/java/com/easyagents/scheduler/quartz/QuartzRuntimeTest.java
@@ -0,0 +1,105 @@
+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.ScheduleHandler;
+import com.easyagents.scheduler.ScheduleId;
+import com.easyagents.scheduler.ScheduleRefireException;
+import org.junit.Test;
+import org.quartz.JobDetail;
+import org.quartz.JobExecutionContext;
+import org.quartz.JobExecutionException;
+import org.quartz.Scheduler;
+import org.quartz.SchedulerException;
+import org.quartz.Trigger;
+
+import java.lang.reflect.Proxy;
+import java.time.Duration;
+import java.time.Instant;
+import java.util.Date;
+import java.util.List;
+import java.util.Map;
+
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+/** {@link QuartzRuntime} 失败控制语义测试。 */
+public class QuartzRuntimeTest {
+
+ /** JobStore 无法保存延迟触发时不得把当前 fire 当作正常完成。 */
+ @Test
+ public void shouldRefireInPlaceWhenPersistentRetryCannotBeStored() throws Exception {
+ ScheduleDefinition definition = new ScheduleDefinition(
+ new ScheduleId("test", "retry-store-failure"),
+ "handler",
+ new OnceSchedulePlan(Instant.parse("2099-01-01T00:00:00Z")),
+ MisfirePolicy.FIRE_ONCE_NOW,
+ ConcurrencyPolicy.DISALLOW,
+ true,
+ Map.of(),
+ "retry store failure"
+ );
+ JobDetail job = QuartzScheduleMapper.toJobDetail(definition);
+ Scheduler scheduler = proxy(Scheduler.class, (proxy, method, arguments) -> {
+ if ("scheduleJob".equals(method.getName())
+ && arguments != null && arguments.length == 1
+ && arguments[0] instanceof Trigger) {
+ throw new SchedulerException("job store unavailable");
+ }
+ if ("isShutdown".equals(method.getName())) return false;
+ return defaultValue(method.getReturnType());
+ });
+ Date fireTime = new Date();
+ JobExecutionContext context = proxy(JobExecutionContext.class,
+ (proxy, method, arguments) -> switch (method.getName()) {
+ case "getMergedJobDataMap" -> job.getJobDataMap();
+ case "getJobDetail" -> job;
+ case "getScheduler" -> scheduler;
+ case "getFireTime", "getScheduledFireTime" -> fireTime;
+ case "getFireInstanceId" -> "fire-1";
+ case "isRecovering" -> false;
+ case "getRefireCount" -> 0;
+ default -> defaultValue(method.getReturnType());
+ });
+ QuartzRuntime runtime = new QuartzRuntime(List.of(new ScheduleHandler() {
+ @Override
+ public String code() {
+ return "handler";
+ }
+
+ @Override
+ public void execute(com.easyagents.scheduler.ScheduleFireContext context) {
+ throw new ScheduleRefireException("registration unavailable",
+ new IllegalStateException("database unavailable"), 0,
+ Duration.ofMillis(1));
+ }
+ }), List.of());
+
+ try {
+ runtime.execute(context);
+ fail("expected refire request");
+ } catch (JobExecutionException exception) {
+ assertTrue(exception.refireImmediately());
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ private static T proxy(Class type, java.lang.reflect.InvocationHandler handler) {
+ return (T) Proxy.newProxyInstance(type.getClassLoader(), new Class[]{type}, handler);
+ }
+
+ private static Object defaultValue(Class type) {
+ if (!type.isPrimitive()) return null;
+ if (type == boolean.class) return false;
+ if (type == byte.class) return (byte) 0;
+ if (type == short.class) return (short) 0;
+ if (type == int.class) return 0;
+ if (type == long.class) return 0L;
+ if (type == float.class) return 0F;
+ if (type == double.class) return 0D;
+ if (type == char.class) return '\0';
+ return null;
+ }
+}
diff --git a/easy-agents-scheduler/easy-agents-scheduler-quartz/src/test/java/com/easyagents/scheduler/quartz/QuartzScheduleServiceTest.java b/easy-agents-scheduler/easy-agents-scheduler-quartz/src/test/java/com/easyagents/scheduler/quartz/QuartzScheduleServiceTest.java
index 238c0ed..8559542 100644
--- a/easy-agents-scheduler/easy-agents-scheduler-quartz/src/test/java/com/easyagents/scheduler/quartz/QuartzScheduleServiceTest.java
+++ b/easy-agents-scheduler/easy-agents-scheduler-quartz/src/test/java/com/easyagents/scheduler/quartz/QuartzScheduleServiceTest.java
@@ -10,20 +10,25 @@ import com.easyagents.scheduler.ScheduleException;
import com.easyagents.scheduler.ScheduleFireContext;
import com.easyagents.scheduler.ScheduleHandler;
import com.easyagents.scheduler.ScheduleId;
+import com.easyagents.scheduler.ScheduleRefireException;
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 org.quartz.impl.matchers.GroupMatcher;
import java.nio.charset.StandardCharsets;
+import java.time.Duration;
import java.time.Instant;
import java.time.ZoneId;
import java.util.Base64;
+import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.UUID;
+import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
@@ -112,6 +117,108 @@ public class QuartzScheduleServiceTest {
assertEquals("value", captured.get().parameters().get("stable"));
}
+ /** 验证可恢复失败会持久化延迟重试，并保留原始 fire 上下文。 */
+ @Test
+ public void shouldRefireRetryableHandlerFailure() throws Exception {
+ AtomicInteger attempts = new AtomicInteger();
+ CountDownLatch completed = new CountDownLatch(1);
+ CopyOnWriteArrayList contexts = new CopyOnWriteArrayList<>();
+ service = newRamService(context -> {
+ contexts.add(context);
+ if (attempts.incrementAndGet() == 1) {
+ throw new ScheduleRefireException("temporary registration failure",
+ new IllegalStateException("database unavailable"), 2,
+ Duration.ofMillis(100));
+ }
+ completed.countDown();
+ });
+ ScheduleDefinition definition = onceDefinition(
+ "retryable-immediate",
+ MisfirePolicy.FIRE_ONCE_NOW,
+ ConcurrencyPolicy.DISALLOW,
+ Instant.parse("2099-01-01T00:00:00Z")
+ );
+ service.create(definition);
+
+ service.triggerNow(definition.id(), "retryable-invocation", Map.of());
+
+ assertTrue("retryable handler was not refired", completed.await(5, TimeUnit.SECONDS));
+ assertEquals(2, attempts.get());
+ assertEquals(contexts.get(0).scheduledFireTime(), contexts.get(1).scheduledFireTime());
+ assertEquals(contexts.get(0).actualFireTime(), contexts.get(1).actualFireTime());
+ assertEquals(contexts.get(0).fireInstanceId(), contexts.get(1).fireInstanceId());
+ assertEquals("retryable-invocation", contexts.get(1).invocationId());
+ }
+
+ /** 持久化延迟重试不得占用当前 Quartz worker。 */
+ @Test
+ public void shouldReleaseWorkerWhilePersistentRetryIsDelayed() throws Exception {
+ CountDownLatch retryScheduled = new CountDownLatch(1);
+ CountDownLatch healthyCompleted = new CountDownLatch(1);
+ CountDownLatch retryCompleted = new CountDownLatch(1);
+ AtomicInteger retryAttempts = new AtomicInteger();
+ service = newRamService(context -> {
+ if (context.scheduleId().name().equals("delayed-retry")) {
+ if (retryAttempts.incrementAndGet() == 1) {
+ retryScheduled.countDown();
+ throw new ScheduleRefireException("temporary registration failure",
+ new IllegalStateException("database unavailable"), 1,
+ Duration.ofSeconds(1));
+ }
+ retryCompleted.countDown();
+ } else {
+ healthyCompleted.countDown();
+ }
+ }, 30_000L, 1);
+ ScheduleDefinition retry = onceDefinition(
+ "delayed-retry", MisfirePolicy.FIRE_ONCE_NOW,
+ ConcurrencyPolicy.DISALLOW, Instant.parse("2099-01-01T00:00:00Z"));
+ ScheduleDefinition healthy = onceDefinition(
+ "healthy-during-retry", MisfirePolicy.FIRE_ONCE_NOW,
+ ConcurrencyPolicy.DISALLOW, Instant.parse("2099-01-01T00:00:00Z"));
+ service.create(retry);
+ service.create(healthy);
+
+ service.triggerNow(retry.id(), "retry-1", Map.of());
+ assertTrue(retryScheduled.await(5, TimeUnit.SECONDS));
+ service.triggerNow(healthy.id(), "healthy-1", Map.of());
+
+ assertTrue("single Quartz worker remained occupied by delayed retry",
+ healthyCompleted.await(750, TimeUnit.MILLISECONDS));
+ assertTrue(retryCompleted.await(5, TimeUnit.SECONDS));
+ }
+
+ /** 已持久化 fire 的重试不得因 replace 而切换到新 Handler。 */
+ @Test
+ public void shouldKeepOriginalHandlerWhenScheduleIsReplacedDuringRetry() throws Exception {
+ AtomicInteger oldAttempts = new AtomicInteger();
+ AtomicInteger newAttempts = new AtomicInteger();
+ CountDownLatch oldRetryCompleted = new CountDownLatch(1);
+ service = newRamServiceWithHandlers(List.of(
+ handler("old-handler", context -> {
+ if (oldAttempts.incrementAndGet() == 1) {
+ throw new ScheduleRefireException("temporary registration failure",
+ new IllegalStateException("database unavailable"), 1,
+ Duration.ofMillis(400));
+ }
+ oldRetryCompleted.countDown();
+ }),
+ handler("new-handler", context -> newAttempts.incrementAndGet())
+ ), 30_000L, 2);
+ ScheduleId id = new ScheduleId("test", "replace-during-retry");
+ ScheduleDefinition original = definition(id, "old-handler");
+ service.create(original);
+ service.triggerNow(id, "replace-retry-1", Map.of());
+ awaitPersistentRetryTrigger();
+
+ service.replace(definition(id, "new-handler"));
+
+ assertTrue("old handler retry did not complete",
+ oldRetryCompleted.await(5, TimeUnit.SECONDS));
+ assertEquals(2, oldAttempts.get());
+ assertEquals(0, newAttempts.get());
+ }
+
/**
* 验证立即触发会在返回回执前校验基础参数与覆盖参数的合并结果。
*/
@@ -385,6 +492,23 @@ public class QuartzScheduleServiceTest {
private QuartzScheduleService newRamService(
Consumer handler,
long shutdownWaitTimeoutMillis
+ ) throws Exception {
+ return newRamService(handler, shutdownWaitTimeoutMillis, 2);
+ }
+
+ private QuartzScheduleService newRamService(
+ Consumer handler,
+ long shutdownWaitTimeoutMillis,
+ int threadCount
+ ) throws Exception {
+ return newRamServiceWithHandlers(List.of(handler("handler", handler)),
+ shutdownWaitTimeoutMillis, threadCount);
+ }
+
+ private QuartzScheduleService newRamServiceWithHandlers(
+ List handlers,
+ long shutdownWaitTimeoutMillis,
+ int threadCount
) throws Exception {
Properties properties = new Properties();
properties.setProperty(
@@ -394,7 +518,7 @@ public class QuartzScheduleServiceTest {
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.threadPool.threadCount", Integer.toString(threadCount));
properties.setProperty("org.quartz.jobStore.class", "org.quartz.simpl.RAMJobStore");
properties.setProperty("org.quartz.jobStore.misfireThreshold", "100");
Scheduler scheduler = new StdSchedulerFactory(properties).getScheduler();
@@ -402,17 +526,7 @@ public class QuartzScheduleServiceTest {
scheduler,
true,
shutdownWaitTimeoutMillis,
- java.util.List.of(new ScheduleHandler() {
- @Override
- public String code() {
- return "handler";
- }
-
- @Override
- public void execute(ScheduleFireContext context) {
- handler.accept(context);
- }
- }),
+ handlers,
java.util.List.of()
);
result.start();
@@ -420,6 +534,50 @@ public class QuartzScheduleServiceTest {
return result;
}
+ private void awaitPersistentRetryTrigger() throws Exception {
+ long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5);
+ while (System.nanoTime() < deadline) {
+ boolean found = service.quartzScheduler()
+ .getTriggerKeys(GroupMatcher.anyTriggerGroup())
+ .stream()
+ .anyMatch(key -> key.getGroup().startsWith(
+ QuartzScheduleMapper.GROUP_PREFIX + "retry."));
+ if (found) return;
+ Thread.sleep(10L);
+ }
+ fail("persistent retry trigger was not created");
+ }
+
+ private static ScheduleHandler handler(
+ String code,
+ Consumer consumer
+ ) {
+ return new ScheduleHandler() {
+ @Override
+ public String code() {
+ return code;
+ }
+
+ @Override
+ public void execute(ScheduleFireContext context) {
+ consumer.accept(context);
+ }
+ };
+ }
+
+ private static ScheduleDefinition definition(ScheduleId id, String handlerCode) {
+ return new ScheduleDefinition(
+ id,
+ handlerCode,
+ new OnceSchedulePlan(Instant.parse("2099-01-01T00:00:00Z")),
+ MisfirePolicy.FIRE_ONCE_NOW,
+ ConcurrencyPolicy.DISALLOW,
+ true,
+ Map.of(),
+ id.name()
+ );
+ }
+
private static ScheduleDefinition cronDefinition(String handlerCode, String expression) {
return new ScheduleDefinition(
new ScheduleId("test", "lifecycle"),
diff --git a/easy-agents-scheduler/easy-agents-scheduler-quartz/src/test/java/com/easyagents/scheduler/quartz/QuartzSchedulerFactoryTest.java b/easy-agents-scheduler/easy-agents-scheduler-quartz/src/test/java/com/easyagents/scheduler/quartz/QuartzSchedulerFactoryTest.java
new file mode 100644
index 0000000..664bb46
--- /dev/null
+++ b/easy-agents-scheduler/easy-agents-scheduler-quartz/src/test/java/com/easyagents/scheduler/quartz/QuartzSchedulerFactoryTest.java
@@ -0,0 +1,40 @@
+package com.easyagents.scheduler.quartz;
+
+import org.junit.Test;
+
+import java.util.Properties;
+
+import static org.junit.Assert.assertEquals;
+
+/** {@link QuartzSchedulerFactory} 原生属性映射测试。 */
+public class QuartzSchedulerFactoryTest {
+
+ @Test
+ public void batchAcquisitionMustRunWithinJobStoreLock() {
+ QuartzSchedulerConfig config = new QuartzSchedulerConfig(
+ "batch-scheduler",
+ "NON_CLUSTERED",
+ "QRTZ_",
+ QuartzSchedulerConfig.STANDARD_JDBC_DELEGATE,
+ false,
+ 8,
+ Thread.NORM_PRIORITY,
+ 8,
+ 1_000L,
+ 15_000L,
+ 60_000L,
+ true,
+ 30_000L,
+ false
+ );
+
+ Properties properties = QuartzSchedulerFactory.properties(config, "testDs");
+
+ assertEquals("8", properties.getProperty(
+ "org.quartz.scheduler.batchTriggerAcquisitionMaxCount"));
+ assertEquals("1000", properties.getProperty(
+ "org.quartz.scheduler.batchTriggerAcquisitionFireAheadTimeWindow"));
+ assertEquals("true", properties.getProperty(
+ "org.quartz.jobStore.acquireTriggersWithinLock"));
+ }
+}
diff --git a/easy-agents-scheduler/easy-agents-scheduler-spring-boot-starter/pom.xml b/easy-agents-scheduler/easy-agents-scheduler-spring-boot-starter/pom.xml
index 2054a77..6c72a5e 100644
--- a/easy-agents-scheduler/easy-agents-scheduler-spring-boot-starter/pom.xml
+++ b/easy-agents-scheduler/easy-agents-scheduler-spring-boot-starter/pom.xml
@@ -53,6 +53,11 @@
h2
test
+
+ org.springframework
+ spring-jdbc
+ test
+
junit
junit
diff --git a/easy-agents-scheduler/easy-agents-scheduler-spring-boot-starter/src/main/java/com/easyagents/scheduler/spring/boot/EasyAgentsSchedulerAutoConfiguration.java b/easy-agents-scheduler/easy-agents-scheduler-spring-boot-starter/src/main/java/com/easyagents/scheduler/spring/boot/EasyAgentsSchedulerAutoConfiguration.java
index c12eb80..f6c9cf4 100644
--- a/easy-agents-scheduler/easy-agents-scheduler-spring-boot-starter/src/main/java/com/easyagents/scheduler/spring/boot/EasyAgentsSchedulerAutoConfiguration.java
+++ b/easy-agents-scheduler/easy-agents-scheduler-spring-boot-starter/src/main/java/com/easyagents/scheduler/spring/boot/EasyAgentsSchedulerAutoConfiguration.java
@@ -18,6 +18,7 @@ import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.autoconfigure.jdbc.DataSourceAutoConfiguration;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.boot.sql.init.dependency.DependsOnDatabaseInitialization;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@@ -61,6 +62,7 @@ public class EasyAgentsSchedulerAutoConfiguration {
*/
@Bean(name = SCHEDULER_BEAN_NAME, initMethod = "start", destroyMethod = "close")
@ConditionalOnMissingBean(ScheduleService.class)
+ @DependsOnDatabaseInitialization
public QuartzScheduleService easyAgentsQuartzScheduleService(
EasyAgentsSchedulerProperties properties,
ListableBeanFactory beanFactory,
@@ -83,6 +85,8 @@ public class EasyAgentsSchedulerAutoConfiguration {
quartz.isClustered(),
quartz.getThreadCount(),
quartz.getThreadPriority(),
+ quartz.getBatchTriggerAcquisitionMaxCount(),
+ quartz.getBatchTriggerAcquisitionFireAheadTimeWindowMillis(),
quartz.getClusterCheckinIntervalMillis(),
quartz.getMisfireThresholdMillis(),
quartz.isWaitForJobsToCompleteOnShutdown(),
diff --git a/easy-agents-scheduler/easy-agents-scheduler-spring-boot-starter/src/main/java/com/easyagents/scheduler/spring/boot/EasyAgentsSchedulerProperties.java b/easy-agents-scheduler/easy-agents-scheduler-spring-boot-starter/src/main/java/com/easyagents/scheduler/spring/boot/EasyAgentsSchedulerProperties.java
index 0d1d055..c527309 100644
--- a/easy-agents-scheduler/easy-agents-scheduler-spring-boot-starter/src/main/java/com/easyagents/scheduler/spring/boot/EasyAgentsSchedulerProperties.java
+++ b/easy-agents-scheduler/easy-agents-scheduler-spring-boot-starter/src/main/java/com/easyagents/scheduler/spring/boot/EasyAgentsSchedulerProperties.java
@@ -104,6 +104,12 @@ public class EasyAgentsSchedulerProperties {
/** Quartz Worker 线程优先级。 */
private int threadPriority = Thread.NORM_PRIORITY;
+ /** 单次批量获取 Trigger 的最大数量。 */
+ private int batchTriggerAcquisitionMaxCount = 1;
+
+ /** 可提前纳入批量的时间窗口，单位毫秒。 */
+ private long batchTriggerAcquisitionFireAheadTimeWindowMillis;
+
/** 集群心跳间隔，单位毫秒。 */
private long clusterCheckinIntervalMillis = 15_000L;
@@ -251,6 +257,45 @@ public class EasyAgentsSchedulerProperties {
this.threadPriority = threadPriority;
}
+ /**
+ * 返回单次批量获取 Trigger 的最大数量。
+ *
+ * @return 批量上限
+ */
+ public int getBatchTriggerAcquisitionMaxCount() {
+ return batchTriggerAcquisitionMaxCount;
+ }
+
+ /**
+ * 设置单次批量获取 Trigger 的最大数量。
+ *
+ * @param batchTriggerAcquisitionMaxCount 批量上限
+ */
+ public void setBatchTriggerAcquisitionMaxCount(
+ int batchTriggerAcquisitionMaxCount) {
+ this.batchTriggerAcquisitionMaxCount = batchTriggerAcquisitionMaxCount;
+ }
+
+ /**
+ * 返回可提前纳入批量的时间窗口。
+ *
+ * @return 毫秒窗口
+ */
+ public long getBatchTriggerAcquisitionFireAheadTimeWindowMillis() {
+ return batchTriggerAcquisitionFireAheadTimeWindowMillis;
+ }
+
+ /**
+ * 设置可提前纳入批量的时间窗口。
+ *
+ * @param batchTriggerAcquisitionFireAheadTimeWindowMillis 毫秒窗口
+ */
+ public void setBatchTriggerAcquisitionFireAheadTimeWindowMillis(
+ long batchTriggerAcquisitionFireAheadTimeWindowMillis) {
+ this.batchTriggerAcquisitionFireAheadTimeWindowMillis =
+ batchTriggerAcquisitionFireAheadTimeWindowMillis;
+ }
+
/**
* 返回集群心跳间隔。
*
diff --git a/easy-agents-scheduler/easy-agents-scheduler-spring-boot-starter/src/test/java/com/easyagents/scheduler/spring/boot/EasyAgentsSchedulerAutoConfigurationTest.java b/easy-agents-scheduler/easy-agents-scheduler-spring-boot-starter/src/test/java/com/easyagents/scheduler/spring/boot/EasyAgentsSchedulerAutoConfigurationTest.java
index debcfef..4e52025 100644
--- a/easy-agents-scheduler/easy-agents-scheduler-spring-boot-starter/src/test/java/com/easyagents/scheduler/spring/boot/EasyAgentsSchedulerAutoConfigurationTest.java
+++ b/easy-agents-scheduler/easy-agents-scheduler-spring-boot-starter/src/test/java/com/easyagents/scheduler/spring/boot/EasyAgentsSchedulerAutoConfigurationTest.java
@@ -114,6 +114,47 @@ public class EasyAgentsSchedulerAutoConfigurationTest {
}
}
+ /**
+ * 验证调度器等待 Spring Boot 数据库脚本初始化完成后再启动。
+ */
+ @Test
+ public void shouldWaitForDatabaseInitializationBeforeStartingScheduler() throws Exception {
+ JdbcDataSource dataSource = dataSource();
+ Map properties = enabledProperties();
+ properties.put("easy-agents.scheduler.data-source-bean-name", "schedulerDataSource");
+ properties.put("spring.sql.init.mode", "always");
+ properties.put(
+ "spring.sql.init.schema-locations",
+ "classpath:quartz-schema/h2-2.5.2.sql"
+ );
+
+ SpringApplication application = new SpringApplication(AutoDiscoveryApplication.class);
+ application.setWebApplicationType(WebApplicationType.NONE);
+ application.setDefaultProperties(properties);
+ application.addInitializers(applicationContext -> {
+ GenericApplicationContext genericContext =
+ (GenericApplicationContext) applicationContext;
+ genericContext.registerBean(
+ "schedulerDataSource",
+ DataSource.class,
+ () -> dataSource
+ );
+ });
+
+ try (ConfigurableApplicationContext context = application.run()) {
+ assertNotNull(context.getBean(ScheduleService.class));
+ try (
+ Connection connection = dataSource.getConnection();
+ Statement statement = connection.createStatement();
+ ResultSet resultSet = statement.executeQuery(
+ "SELECT COUNT(*) FROM QRTZ_SCHEDULER_STATE"
+ )
+ ) {
+ assertTrue(resultSet.next());
+ }
+ }
+ }
+
/**
* 验证多个 DataSource 未明确选择时启动失败并提供可操作信息。
*/
@@ -162,6 +203,13 @@ public class EasyAgentsSchedulerAutoConfigurationTest {
properties.put("easy-agents.scheduler.quartz.instance-id", "NON_CLUSTERED");
properties.put("easy-agents.scheduler.quartz.clustered", "false");
properties.put("easy-agents.scheduler.quartz.thread-count", "2");
+ properties.put(
+ "easy-agents.scheduler.quartz.batch-trigger-acquisition-max-count", "2"
+ );
+ properties.put(
+ "easy-agents.scheduler.quartz.batch-trigger-acquisition-fire-ahead-time-window-millis",
+ "500"
+ );
properties.put("easy-agents.scheduler.quartz.misfire-threshold-millis", "1000");
return properties;
}