持久任务：Spring Batch 6 的运行模型与 Quartz 2 的持久调度

@Scheduled 只回答“进程内存里的一个定时器到点了”。真实业务还要回答另外两组问题：一批数据怎么分块、每块失败后从哪继续、重复执行会不会重复写业务表（这是批处理的运行模型）；任务定义能不能跨进程、多实例会不会同时触发、实例崩了谁接管、错过了若干个点补不补（这是持久调度器）。Spring 生态把这两组问题交给两个不同的项目：Spring Batch 提供 job/step/chunk 与可持久化的执行元数据，但自己不调度；Quartz 提供持久化的触发与集群调度，但不懂“一批数据”的语义。本页把两者分别放进真实 MySQL 8.4.7 跑一组有界实验，逐项观察“调度成功”与“业务成功”之间的缝隙。

本页是 U46（#83）的权威正文，覆盖 PRD P06 的 Batch/Quartz 分段与 F15 的官方集成登记。单机 @Scheduled 的语义（fixedRate 追赶、异常吞噬、关闭两开关）仍归调度执行：@Scheduled 语义、异常吞噬、关闭与 JMS 集成边界；独立调度平台 xxl-job 的四段链路归数据方向的 DAG 调度正文；JdbcTemplate.batchUpdate 的驱动级批处理归Spring 事务：JDBC、异常转换、批处理与生成键；消息通道与错误流归 U47，AMQP/Kafka/Pulsar 直连归 U48。

问题、术语与三个项目

先把口语里的“定时任务”拆成四层，每层有各自的持久对象：

层	谁负责	持久对象	一句话
进程内定时	`@Scheduled` + `TaskScheduler`	无（内存）	到点调用一个方法；进程退出就没了
批处理运行模型	Spring Batch	`BATCH_JOB_INSTANCE` / `BATCH_JOB_EXECUTION` / `BATCH_STEP_EXECUTION` / 执行上下文	一次运行由哪些 step、哪些 chunk、事务边界与 checkpoint 组成
持久调度	Quartz	`QRTZ_JOB_DETAILS` / `QRTZ_TRIGGERS` / `QRTZ_FIRED_TRIGGERS` / `QRTZ_SCHEDULER_STATE` / `QRTZ_LOCKS`	触发器何时该响、谁抢到了、多实例谁活着
平台化调度	xxl-job 等	自有任务表与日志表	调度、执行器、日志、告警与运维界面（U85 已讲）

术语在本页固定为：

术语	本页定义	容易混淆的说法
JobInstance	业务上的一次“作业单”，身份 = `JOB_NAME + JOB_KEY`	不是 JobExecution 的同义词
JobExecution	对同一 JobInstance 的一次尝试；失败重启产生新 JobExecution	不是“重跑就改原行”
StepExecution	一个 step 在某个 JobExecution 里的一次执行，带读/写/提交/回滚计数	不是 chunk
chunk	处理 + 写入 + 提交的原子单位；`chunkSize` 是提交间隔	不是“一批读进来就一事务”
checkpoint	在 chunk 提交时写入执行上下文的读位置	不是“每读一条就存一次”
misfire	触发器到点时调度器没来得及处理	不是“业务重试”
集群接管	存活实例发现死实例并把它的触发/心跳清理掉	不是“主从选举”

三个项目各管一段：Spring Batch 的 JobLauncher 只负责“创建执行并运行”，它不会自己到点；Quartz 的 Job 只知道“到点调用 execute”，它不提供 chunk、事务边界或失败跳过；@Scheduled 两样都没有。把它们串起来（Quartz 触发 Batch 启动、Batch 写 Quartz 的任务表）是应用层的集成，不是框架承诺——本页在“容量与设计取舍”一节给出组合方式和责任边界。

版本、工程与观察入口

固定组合：Spring Boot 4.0.9 BOM、Spring Batch 6.0.6（tag v6.0.6，提交 cc8c2f31e58e4bbeca3fd9d04a4c17d92c8f46d9）、Quartz 2.5.2（tag v2.5.2，提交 ed52d39452d156874afb2188c461ff0e7c90cae8）、Spring Framework 7.0.10、mysql-connector-j 9.7.0、MySQL 8.4.7（本机 DBngin）。运行环境：BellSoft Liberica JDK 21.0.11、Maven 3.9.9、macOS 27.2 arm64。

公开入口：

实验代码sh

python3 labs/batch-quartz/run.py

脚本每次运行创建独有的 MySQL 库 study_u46_<runid>_<uuid>（每个测试方法一个），用官方 schema-mysql.sql 与官方 tables_mysql_innodb.sql 建表；结束时测试 fixture 关闭连接池并 DROP DATABASE，runner 再核对没有 study_u46% 残留。完整命令、退出码、19 个测试名、观察行与清理结果见运行证据，工程边界见实验说明。

6.0 的 API 迁移必须先说清，否则照 5.x 的文章写会编译不过：item 接口从 org.springframework.batch.item 迁到 org.springframework.batch.infrastructure.item；ChunkOrientedTasklet、FaultTolerantStepBuilder、SimpleStepFactoryBean 全部标了 @Deprecated(since="6.0", forRemoval=true)，新的 chunk 模型是 ChunkOrientedStep + ChunkOrientedStepBuilder；StepBuilder.chunk(int, PlatformTransactionManager) 弃用，新的 chunk(int) 不再接受事务管理器参数，默认值是 ResourcelessTransactionManager——本实验第一版就因此“整块失败但业务表一行都没留”，因为根本没有业务事务。JobOperator.stop/abandon/restart 的 long id 重载弃用，6.0 新增 recover(JobExecution)。

Boot 4 把 Batch 自动配置拆成两个模块：spring-boot-batch（内存 store，无 DataSource 时的默认）与 spring-boot-batch-jdbc（JDBC store），starter 是 spring-boot-starter-batch-jdbc；Quartz 仍在 spring-boot-quartz，spring.quartz.job-store-type 默认 MEMORY。

源码依据为固定 tag 的连续片段，元数据含整文件 SHA-256 与行范围，由 scripts/check_source_excerpts.py 联网回源比对。本页只引用一次完整的成功运行（exit 0，19 个测试、18 条观察行）；开发过程中失败的多轮运行与修复记录留在仓库外缓存 ~/.cache/software-engineering-study/java-depth-impl/u46/runtime/ 下（证据 JSON 的 logs 字段指向本次运行目录），不以失败运行的数字冒充结论。

身份：JOB_NAME 与 identifying 参数

Batch 的“一次作业”不是由启动时间决定，而是由作业名 + 识别参数决定。DefaultJobKeyGenerator 只拼接 identifying 参数（按名字排序后取 MD5），非 identifying 参数完全不参与：

原码java

	@Override
	public String generateKey(JobParameters source) {

		Assert.notNull(source, "source must not be null");
		Set<JobParameter<?>> parameters = source.parameters();
		StringBuilder stringBuffer = new StringBuilder();
		List<String> keys = parameters.stream().map(JobParameter::name).sorted().toList();
		for (String key : keys) {
			JobParameter<?> jobParameter = source.getParameter(key);
			if (jobParameter != null && jobParameter.identifying()) {
				stringBuffer.append(jobParameter);
			}
		}
		return DigestUtils.md5DigestAsHex(stringBuffer.toString().getBytes(StandardCharsets.UTF_8));
	}

JdbcJobInstanceDao.createJobInstance 把 JOB_NAME + JOB_KEY 写进 BATCH_JOB_INSTANCE，并依赖表上的 UNIQUE (JOB_NAME, JOB_KEY) 兜底；查找实例时用同一个生成器：

原码java

public JobInstance createJobInstance(String jobName, JobParameters jobParameters) {

	Assert.notNull(jobName, "Job name must not be null.");
	Assert.notNull(jobParameters, "JobParameters must not be null.");

	Assert.state(getJobInstance(jobName, jobParameters) == null, "JobInstance must not already exist");

	long jobInstanceId = jobInstanceIncrementer.nextLongValue();

	JobInstance jobInstance = new JobInstance(jobInstanceId, jobName);
	jobInstance.incrementVersion();

	Object[] parameters = new Object[] { jobInstanceId, jobName, jobKeyGenerator.generateKey(jobParameters),
			jobInstance.getVersion() };
	getJdbcTemplate().update(getQuery(CREATE_JOB_INSTANCE), parameters,
			new int[] { Types.BIGINT, Types.VARCHAR, Types.VARCHAR, Types.INTEGER });

	return jobInstance;
}

实验 batchIdentity：同一 job 用 runDate=2026-10-31 跑两次，第二次抛 JobInstanceAlreadyCompleteException；换 runDate=2026-10-30 得到新实例；同名 job 加上不同的 traceId（非 identifying）仍然命中同一实例、同样被拒。观察行 instances=2 executions=2 jobKeyDiffers=true nonIdentifyingIgnored=true 与 BATCH_JOB_INSTANCE 的 2 行、JOB_KEY 不同一致。

实践含义：

“重跑同一天的数据”天然被 Batch 的实例身份挡住——要么换 identifying 参数（例如加 run.id 时间戳）开新实例，要么用 JobOperator.startNextInstance（要求 job 配了 JobParametersIncrementer），要么先 abandon 旧实例。
把“业务日期”设成 identifying 参数、把“重试次数/操作人”设成非 identifying 参数，就能得到“每天一个实例、可以在同一天内多次重启”的语义；反过来配置会让每次重试都变成新实例。
JOB_KEY 是 MD5 摘要，不要把它当业务幂等键对外暴露；业务幂等仍要自己的键（见重复消费、幂等与顺序消息）。

运行状态机与“创建执行”的校验链

BatchStatus 的枚举顺序有语义：越往后越“严重”，聚合取最大值，COMPLETED 在第一位，所以只有全部子项都 COMPLETED 才可能聚合出 COMPLETED：

原码java

 * The order of the status values is significant because it can be used to aggregate a
 * set of status values. The result should be the maximum value. Since {@code
 * COMPLETED} is first in the order, only if all elements of an execution are {@code
 * COMPLETED} can the aggregate status be COMPLETED. A running execution is expected
 * to move from {@code STARTING} to {@code STARTED} to {@code COMPLETED} (through the
 * order defined by {@link #upgradeTo(BatchStatus)}). Higher values than {@code
 * STARTED} signify more serious failures. {@code ABANDONED} is used for steps that
 * have finished processing but were not successful and where they should be skipped
 * on a restart (so {@code FAILED} is the wrong status).
 */

/**
 * The batch job has successfully completed its execution.
 */
COMPLETED,
/**
 * Status of a batch job prior to its execution.
 */
STARTING,
/**
 * Status of a batch job that is running.
 */
STARTED,
/**
 * Status of batch job waiting for a step to complete before stopping the batch job.
 */
STOPPING,
/**
 * Status of a batch job that has been stopped by request.
 */
STOPPED,
/**
 * Status of a batch job that has failed during its execution.
 */
FAILED,
/**
 * Status of a batch job that did not stop properly and can not be restarted.
 */
ABANDONED,
/**
 * Status of a batch job that is in an uncertain state.
 */
UNKNOWN;

isRunning 只认 STARTING / STARTED / STOPPING 三态：

原码java

/**
 * Convenience method to decide if a status indicates that work is in progress.
 * @return true if the status is STARTING, STARTED, STOPPING
 */
public boolean isRunning() {
	return this == STARTING || this == STARTED || this == STOPPING;
}

UNKNOWN 是单独的语义：事务提交结果无法确定时（提交或回滚本身失败），框架把 step 升级为 UNKNOWN 并终止流程，重启会被拒绝，必须人工介入或先 recover：

原码java

private void markUnknownIfOutcomeIsUnknown(StepExecution stepExecution) {
	registerSynchronization(new TransactionSynchronization() {
		@Override
		public void afterCompletion(int status) {
			if (status == STATUS_UNKNOWN) {
				logger.error("Chunk transaction ended with an unknown outcome, the step execution is marked as "
						+ BatchStatus.UNKNOWN);
				stepExecution.upgradeStatus(BatchStatus.UNKNOWN);
				stepExecution.setTerminateOnly();
			}
		}
	});

真正决定“这次启动是全新、重启还是拒绝”的是 launcher 创建执行前的检查链：

原码java

private JobExecution createJobExecution(Job job, JobParameters jobParameters)
		throws JobExecutionAlreadyRunningException, JobRestartException, JobInstanceAlreadyCompleteException,
		InvalidJobParametersException {
	JobInstance jobInstance = jobRepository.getJobInstance(job.getName(), jobParameters);
	ExecutionContext executionContext;
	if (jobInstance == null) { // fresh start
		logger.debug(
				"Creating a new job instance for job = " + job.getName() + " with parameters = " + jobParameters);
		jobInstance = jobRepository.createJobInstance(job.getName(), jobParameters);
		executionContext = new ExecutionContext();
	}
	else { // restart
		logger.debug(
				"Found existing job instance for job = " + job.getName() + " with parameters = " + jobParameters);
		JobExecution lastJobExecution = jobRepository.getLastJobExecution(jobInstance);
		if (lastJobExecution == null) {
			throw new IllegalStateException("Cannot find any job execution for job instance: " + jobInstance);
		}
		// A new execution is only created once these checks pass on the previous one,
		// so only the last execution can be running or complete.
		if (lastJobExecution.isRunning()) {
			throw new JobExecutionAlreadyRunningException(
					"A job execution for this job is already running: " + lastJobExecution);
		}
		BatchStatus status = lastJobExecution.getStatus();
		if (status == BatchStatus.UNKNOWN) {
			throw new JobRestartException("Cannot restart job from UNKNOWN status. "
					+ "The last execution ended with a failure that could not be rolled back, "
					+ "so it may be dangerous to proceed. Manual intervention is probably necessary.");
		}
		if (status == BatchStatus.COMPLETED || status == BatchStatus.ABANDONED) {
			JobParameters identifyingJobParameters = new JobParameters(
					lastJobExecution.getJobParameters().getIdentifyingParameters());
			throw new JobInstanceAlreadyCompleteException(
					"A job instance already exists and is complete for identifying parameters="
							+ identifyingJobParameters + ".  If you want to run this job again, "
							+ "change the parameters.");
		}
		// check if the job is restartable
		if (!job.isRestartable()) {
			throw new JobRestartException("JobInstance already exists and is not restartable");
		}
		/*
		 * validate here if it has stepExecutions that are UNKNOWN, STARTING, STARTED
		 * and STOPPING retrieve the previous execution and check
		 */
		for (StepExecution execution : lastJobExecution.getStepExecutions()) {
			BatchStatus stepStatus = execution.getStatus();
			if (stepStatus.isRunning()) {
				throw new JobExecutionAlreadyRunningException(
						"A job execution for this job is already running: " + lastJobExecution);
			}
			else if (stepStatus == BatchStatus.UNKNOWN) {
				throw new JobRestartException(
						"Cannot restart step [" + execution.getStepName() + "] from UNKNOWN status. "
								+ "The last execution ended with a failure that could not be rolled back, "
								+ "so it may be dangerous to proceed. Manual intervention is probably necessary.");
			}
		}

		executionContext = lastJobExecution.getExecutionContext();
	}

	// Check the validity of the parameters before creating anything
	// in the repository...
	job.getJobParametersValidator().validate(jobParameters);

逐条读：

先按 (jobName, parameters) 查实例；没有就是 fresh start，直接建实例。
有实例则取最后一次执行（SQL 是 MAX(JOB_EXECUTION_ID)，不是按时间排序）：running → JobExecutionAlreadyRunningException；UNKNOWN → JobRestartException；COMPLETED 或 ABANDONED → JobInstanceAlreadyCompleteException；!job.isRestartable() → JobRestartException。
再看上一次执行的每个 step：running → AlreadyRunning；UNKNOWN → Restart。这两步是“step 卡死但 job 没更新”的兜底。
校验参数（JobParametersValidator）在创建任何记录之前；默认 validator 对未知参数只告警不拒绝，需要严格模式要自己声明 required/optional 键。
最后才 jobRepository.createJobExecution(...) 写新执行行。

run() 本体很短：创建执行，然后交给 taskExecutor 执行。默认 SyncTaskExecutor，所以 start() 在 job 跑完之前不会返回；实验 batchConcurrentLaunch 先把执行器换成异步线程池，才观察到“start() 返回时状态是 STARTING、同参数第二次启动被 AlreadyRunning 拒绝、最终只有 1 条执行记录”。

原码java

public JobExecution run(final Job job, final JobParameters jobParameters)
		throws JobExecutionAlreadyRunningException, JobRestartException, JobInstanceAlreadyCompleteException,
		InvalidJobParametersException {
	Assert.notNull(job, "The Job must not be null.");
	Assert.notNull(jobParameters, "The JobParameters must not be null.");
	JobExecution jobExecution = createJobExecution(job, jobParameters);
	launchJobExecution(job, jobExecution);
	return jobExecution;
}

这里有一个容易踩的并发窗口：检查（查 last execution）与插入（createExecution）不是同一把锁。Batch 的兜底是 BATCH_JOB_INSTANCE 上的唯一约束与 create* 的 SERIALIZABLE 事务（见“仓储事务与并发启动”）。多进程同时启动同一实例时，输的一方会拿到唯一键冲突或 JobExecutionAlreadyRunningException，而不是跑出两条并行执行——但这依赖数据库支持串行化，见下文隔离级别说明。

step 执行：跳过与重启

SimpleJob 的顺序执行语义只有一条：任一 step 不是 COMPLETED 就停下，job 状态取最后一个 step 的状态：

原码java

protected void doExecute(JobExecution execution)
		throws JobInterruptedException, JobRestartException, StartLimitExceededException {

	StepExecution stepExecution = null;
	for (Step step : steps) {
		stepExecution = handleStep(step, execution);
		if (stepExecution.getStatus() != BatchStatus.COMPLETED) {
			//
			// Terminate the job if a step fails
			//
			break;
		}
	}

	//
	// Update the job status to be the same as the last step
	//
	if (stepExecution != null) {
		if (logger.isDebugEnabled()) {
			logger.debug("Upgrading JobExecution status: " + stepExecution);
		}
		execution.upgradeStatus(stepExecution.getStatus());
		execution.setExitStatus(stepExecution.getExitStatus());
	}
}

重启时“已完成的 step 不重跑”由 SimpleStepHandler.shouldStart 决定：COMPLETED && !allowStartIfComplete 或 ABANDONED 都返回 false；startLimit 默认 Integer.MAX_VALUE，只在你显式收紧时才拦截：

原码java

protected boolean shouldStart(StepExecution lastStepExecution, JobExecution jobExecution, Step step)
		throws JobRestartException, StartLimitExceededException {

	BatchStatus stepStatus;
	if (lastStepExecution == null) {
		stepStatus = BatchStatus.STARTING;
	}
	else {
		stepStatus = lastStepExecution.getStatus();
	}

	if (stepStatus == BatchStatus.UNKNOWN) {
		throw new JobRestartException("Cannot restart step from UNKNOWN status. "
				+ "The last execution ended with a failure that could not be rolled back, "
				+ "so it may be dangerous to proceed. Manual intervention is probably necessary.");
	}

	if ((stepStatus == BatchStatus.COMPLETED && !step.isAllowStartIfComplete())
			|| stepStatus == BatchStatus.ABANDONED) {
		// step is complete, false should be returned, indicating that the
		// step should not be started
		if (logger.isInfoEnabled()) {
			logger.info("Step already complete or not restartable, so no action to execute: " + lastStepExecution);
		}
		return false;
	}

	JobInstance jobInstance = jobExecution.getJobInstance();
	long stepExecutionCount = 0;
	try {
		stepExecutionCount = jobRepository.getStepExecutionCount(jobInstance, step.getName());
	}
	catch (NoSuchStepException e) {
		throw new JobRestartException("Unable to count step executions for job instance " + jobInstance.getId(), e);
	}
	if (stepExecutionCount < step.getStartLimit()) {
		// step start count is less than start max, return true
		return true;
	}
	else {
		// start max has been exceeded, throw an exception.
		throw new StartLimitExceededException(
				"Maximum start limit exceeded for step: " + step.getName() + "StartMax: " + step.getStartLimit());
	}

step 真的重启时，新 StepExecution 继承上一次的 ExecutionContext（并去掉 batch.executed 标记），这正是 reader 能接着上次位置读的通道：

原码java

if (shouldStart(lastStepExecution, execution, step)) {

	currentStepExecution = jobRepository.createStepExecution(step.getName(), execution);

	boolean isRestart = (lastStepExecution != null
			&& !lastStepExecution.getStatus().equals(BatchStatus.COMPLETED));

	if (isRestart) {
		currentStepExecution.setExecutionContext(lastStepExecution.getExecutionContext());

		if (lastStepExecution.getExecutionContext().containsKey("batch.executed")) {
			currentStepExecution.getExecutionContext().remove("batch.executed");
		}
	}
	else {
		currentStepExecution.setExecutionContext(new ExecutionContext(executionContext));
	}

实验 batchStepSkip：两步 job，第一步计数、第二步第一次失败；重启后 step1Runs=1 step2Runs=2 step1Executions=1——已完成 step 既没有新执行行，也没有再跑。

chunk 事务与 checkpoint

chunk 是 Batch 的提交单位：每个 chunk 一个业务事务，读、处理、写、以及“把 reader 的位置写进执行上下文”在同一个事务里提交。核心循环：

原码java

protected void doExecute(StepExecution stepExecution) throws Exception {
	stepExecution.getExecutionContext().put(STEP_TYPE_KEY, this.getClass().getName());
	while (this.chunkTracker.get().moreItems() && !interrupted(stepExecution)) {
		// process next chunk in its own transaction
		this.transactionTemplate.executeWithoutResult(transactionStatus -> {
			ChunkTransactionEvent chunkTransactionEvent = new ChunkTransactionEvent(stepExecution.getStepName(),
					stepExecution.getId());
			chunkTransactionEvent.begin();
			markUnknownIfOutcomeIsUnknown(stepExecution);
			StepExecutionState stateBeforeTransaction = StepExecutionState.of(stepExecution);
			StepContribution contribution = stepExecution.createStepContribution();
			ChunkTracker<I, O> tracker = this.chunkTracker.get();
			boolean scanning = tracker.isScanMode();
			processNextChunk(transactionStatus, contribution, stepExecution);

			// Skip update during rollback to avoid OptimisticLockingFailureException
			if (transactionStatus.isRollbackOnly()) {
				// Explicitly mark as locally rollback-only to prevent
				// UnexpectedRollbackException when the transaction manager
				// (eg JpaTransactionManager) has marked it as globally rollback-only
				// (eg after a JPA flush failure) but not locally rollback-only.
				transactionStatus.setRollbackOnly();
				chunkTransactionEvent.transactionStatus = BatchMetrics.STATUS_ROLLED_BACK;
				chunkTransactionEvent.commit();
				return;
			}

			saveState(stepExecution, tracker, scanning, stateBeforeTransaction);
			chunkTransactionEvent.transactionStatus = BatchMetrics.STATUS_COMMITTED;
			chunkTransactionEvent.commit();
		});
		// a scanned item skipped on write is rolled back with its transaction, so the
		// skip is committed in a new transaction, as if the item had been written:
		// otherwise, a restart after the step ended (or the JVM crashed) before the
		// next commit would process the skipped item again
		ChunkTracker<I, O> tracker = this.chunkTracker.get();
		if (tracker.consumeScannedItemSkip()) {
			this.transactionTemplate.executeWithoutResult(transactionStatus -> {
				markUnknownIfOutcomeIsUnknown(stepExecution);
				saveState(stepExecution, tracker, true, StepExecutionState.of(stepExecution));
			});
		}
	}
}

三个关键点：

先读满一块再进事务：readChunk 在事务内读 chunkSize 条（本实验为 2），读到的位置由 reader 自己计数。
saveState 在提交前写 checkpoint：compositeItemStream.update(executionContext) 把读位置写进上下文，再 jobRepository.update(stepExecution) 更新计数。位置的权威来源是 BATCH_STEP_EXECUTION_CONTEXT（SHORT_CONTEXT 2500 字符以内直接存，超出走 SERIALIZED_CONTEXT 大字段）。
回滚时 checkpoint 不前进：事务被标记回滚则直接 return，不写状态；所以失败块整体回滚、下一轮从这一块的起点重读。

reader 侧的对称逻辑（AbstractItemCountingItemStreamItemReader）是 open 时从上下文恢复计数并 jumpToItem，update 时把当前计数写回：

原码java

public void open(ExecutionContext executionContext) throws ItemStreamException {
	super.open(executionContext);
	try {
		doOpen();
	}
	catch (Exception e) {
		throw new ItemStreamException("Failed to initialize the reader", e);
	}
	if (!isSaveState()) {
		return;
	}

	if (executionContext.containsKey(getExecutionContextKey(READ_COUNT_MAX))) {
		maxItemCount = executionContext.getInt(getExecutionContextKey(READ_COUNT_MAX));
	}

	int itemCount = 0;
	if (executionContext.containsKey(getExecutionContextKey(READ_COUNT))) {
		itemCount = executionContext.getInt(getExecutionContextKey(READ_COUNT));
	}
	else if (currentItemCount > 0) {
		itemCount = currentItemCount;
	}

	if (itemCount > 0 && itemCount < maxItemCount) {
		try {
			jumpToItem(itemCount);
		}
		catch (Exception e) {
			throw new ItemStreamException("Could not move to stored position on restart", e);
		}
	}

	currentItemCount = itemCount;

}

@Override
public void update(ExecutionContext executionContext) throws ItemStreamException {
	super.update(executionContext);
	if (saveState) {
		Assert.notNull(executionContext, "ExecutionContext must not be null");
		executionContext.putInt(getExecutionContextKey(READ_COUNT), currentItemCount);
		if (maxItemCount < Integer.MAX_VALUE) {
			executionContext.putInt(getExecutionContextKey(READ_COUNT_MAX), maxItemCount);
		}
	}

}

实验 batchRestart 的数字把这条链说全了：6 条数据、chunk=2、第 3 条写入时失败——失败后业务表只有前一块的 2 行（rowsAfterFailure=2），step 状态 FAILED、ROLLBACK_COUNT=1；重启后从第 3 条继续（reads=10 而不是 6，因为失败块被重读了一遍），最终 finalRows=6、第二次执行的 COMMIT_COUNT=2。

“至少一次”落在业务表上：失败块里已经写入但未提交的行会随回滚消失，重启后重新写；而已经提交的块不会重写。所以重复只发生在“最后一块未提交的边界附近”，业务侧仍应使用幂等键（唯一约束、ON DUPLICATE KEY、Inbox），不能指望 Batch 的 checkpoint 提供“恰好一次”。

容错默认值：不开 faultTolerant 会怎样

Batch 6 的 ChunkOrientedStepBuilder 默认值写得很直白：faultTolerant=false、retryLimit=0、skipLimit=10；构建时若不满足“显式配了可跳过异常或 skipLimit>0”，策略退化为 NeverSkipItemSkipPolicy：

原码java

if (this.retryPolicy == null) {
	if (!this.retryableExceptions.isEmpty() || this.retryLimit > 0) {
		// Default to Exception.class when retryLimit is set without explicit
		// retry() configuration. This prevents retrying fatal JVM errors like
		// OutOfMemoryError and StackOverflowError.
		Set<Class<? extends Throwable>> exceptions = this.retryableExceptions.isEmpty()
				? Set.of(Exception.class) : this.retryableExceptions;
		this.retryPolicy = RetryPolicy.builder().maxRetries(this.retryLimit).includes(exceptions).build();
	}
	else {
		this.retryPolicy = throwable -> false;
	}
}
chunkOrientedStep.setRetryPolicy(this.retryPolicy);
if (this.skipPolicy == null) {
	if (!this.skippableExceptions.isEmpty() || this.skipLimit > 0) {
		this.skipPolicy = new LimitCheckingExceptionHierarchySkipPolicy(this.skippableExceptions,
				this.skipLimit);
	}
	else {
		this.skipPolicy = new NeverSkipItemSkipPolicy();
	}
}
chunkOrientedStep.setSkipPolicy(this.skipPolicy);
chunkOrientedStep.setFaultTolerant(this.faultTolerant);

faultTolerant() 只是把开关置 true，是否真的重试/跳过仍看策略：

原码java

/**
 * Set whether the step is fault-tolerant or not. A fault-tolerant step can handle
 * failures and continue processing without failing the entire step. This is useful
 * for scenarios where individual items may fail and be skipped, but the overall step
 * should still complete successfully. Defaults to false.
 * @return this for fluent chaining
 */
public ChunkOrientedStepBuilder<I, O> faultTolerant() {
	this.faultTolerant = true;
	return self();
}

写失败的容错路径分两段：先在事务内重试整块；重试耗尽后进入 scan，在新事务里逐条重写以定位坏条目。可跳过（no-rollback）的写异常会被强制回滚整块，再靠 scan 把好条目补写、坏条目交给 skip 策略：

原码java

RetryCallback<Object, Exception> retryCallback = context -> {
	contextHolder.set(context);

	if (!data.scanning()) {
		chunkMonitor.setChunkSize(inputs.size());
		Timer.Sample sample = MicrometerMetrics.createTimerSample(meterRegistry);
		String status = BatchMetrics.STATUS_SUCCESS;
		try {
			doWrite(outputs);
		}
		catch (Exception e) {
			status = BatchMetrics.STATUS_FAILURE;
			if (rollbackClassifier.classify(e)) {
				throw e;
			}
			/*
			 * If the exception is marked as no-rollback, we need to override
			 * that, otherwise there's no way to write the rest of the chunk or to
			 * honour the skip listener contract.
			 */
			throw new ForceRollbackForWriteSkipException(
					"Force rollback on skippable exception so that skipped item can be located.", e);
		}
		finally {
			stopTimer(sample, contribution.getStepExecution(), "chunk.write", status, "Chunk writing");
		}
		contribution.incrementWriteCount(outputs.size());
		contribution.incrementWriteSkipCount(outputs.getSkipsSize());
	}
	else {
		scan(contribution, inputs, outputs, chunkMonitor, false);
	}
	return null;

};

原码java

Chunk<O> items = Chunk.of(outputIterator.next());
inputIterator.next();
try {
	writeItems(items);
	// If successful we are going to return and allow
	// the driver to commit...
	doAfterWrite(items);
	contribution.incrementWriteCount(1);
	inputIterator.remove();
	outputIterator.remove();
}
catch (Exception e) {
	try {
		doOnWriteError(e, items);
	}
	finally {
		Throwable cause = e;
		if (e instanceof StepListenerFailedException) {
			cause = e.getCause();
		}

		if (!shouldSkip(itemWriteSkipPolicy, cause, -1) && !rollbackClassifier.classify(cause)) {
			inputIterator.remove();
			outputIterator.remove();
		}
		else {
			checkSkipPolicy(inputIterator, outputIterator, cause, contribution, recovery);
		}
		if (rollbackClassifier.classify(cause)) {
			throw (Exception) cause;
		}
	}
}

实验两组对照：

batchSkipWithoutFt：只写 .skipLimit(10)，不调 .faultTolerant()——step FAILED，WRITE_SKIP_COUNT=0。设了 skipLimit 不等于能跳过。
batchFaultTolerant：.faultTolerant().skip(IllegalStateException.class).skipLimit(5)，第 2 条永久失败——step COMPLETED，业务表 3 行（1/3/4），WRITE_SKIP_COUNT=1。

代价要说清：scan 阶段把整块拆成逐条写入，每条一个新事务；块越大、坏条目越少，放大越明显。skip 是“允许丢数据”的业务决策，配合 SkipListener 落审计表才完整；skipLimit 用尽会抛 SkipLimitExceededException 让 step 失败，不会静默兜底。

重试与跳过如何分工：retry 只给暂态错误（连接复位、死锁），skip 给确定性错误（解析失败、脏数据）。6.0 的构建注释明确写了“不要用 retry 处理确定性失败”；把确定性问题放进 retry 只会让失败慢三倍再进入同一个结局。

仓储事务与并发启动

JobRepository 的 create* 方法被事务拦截器包成 REQUIRES_NEW + ISOLATION_SERIALIZABLE，其余方法 REQUIRED；默认还开着“事务态校验”，在已有事务里调用 create* 会直接抛异常：

原码java

private static class DefaultJobRepositoryTransactionAttributeSource extends NameMatchTransactionAttributeSource {

	public DefaultJobRepositoryTransactionAttributeSource(String isolationLevelForCreate) {
		Properties transactionAttributes = new Properties();
		transactionAttributes.setProperty("create*",
				TRANSACTION_PROPAGATION_PREFIX + Propagation.REQUIRES_NEW + "," + isolationLevelForCreate);
		transactionAttributes.setProperty("getLastJobExecution*",
				TRANSACTION_PROPAGATION_PREFIX + Propagation.REQUIRES_NEW + "," + isolationLevelForCreate);
		transactionAttributes.setProperty("*", "PROPAGATION_REQUIRED");
		this.setProperties(transactionAttributes);
	}

原码java

public JobRepository getObject() throws Exception {
	TransactionInterceptor advice = new TransactionInterceptor((TransactionManager) this.transactionManager,
			this.transactionAttributeSource);
	if (this.validateTransactionState) {
		DefaultPointcutAdvisor advisor = new DefaultPointcutAdvisor((MethodInterceptor) invocation -> {
			if (TransactionSynchronizationManager.isActualTransactionActive()) {
				throw new IllegalStateException("Existing transaction detected in JobRepository. "
						+ "Please fix this and try again (e.g. remove @Transactional annotations from client).");
			}
			return invocation.proceed();
		});
		NameMatchMethodPointcut pointcut = new NameMatchMethodPointcut();
		pointcut.addMethodName("create*");
		advisor.setPointcut(pointcut);
		this.proxyFactory.addAdvisor(advisor);
	}

这两条合起来说明：

“两个进程同时启动同一实例只会有一个成功”依赖数据库的串行化隔离，而不是应用层锁；MySQL InnoDB 在 SERIALIZABLE 下把普通 SELECT 升级为共享锁，配合唯一约束实现。官方文档也承认这是“相当激进”的默认，并允许降到 READ_COMMITTED——代价是并发冲突时由唯一约束兜底，异常类型可能变成 DuplicateKeyException。
不要在 @Transactional 方法里调 jobOperator.start(...)：业务事务里调用 create* 触发 validateTransactionState 的兜底异常。把启动放到事务外（或消息驱动、调度器回调的顶层）是正确用法。
批量写业务表时，让 writer 用事务绑定连接（DataSourceUtils.getConnection / JdbcTemplate 自动如此）。实验第一版 writer 自己 dataSource.getConnection()，块回滚后行还在，观察结果直接误导了结论——这不是 Batch 的坑，而是自己绕过了事务。

abandon、recover 与 stop

三个“非正常结束”操作语义完全不同：

stop(execution)：给运行中的执行发停止信号，状态置 STOPPING，step 在 chunk 边界检查中断；能否停下来取决于 step 是否配合，返回值只表示“信号已发出”。
abandon(execution)：把执行升级为 ABANDONED，只能对已停止/失败的执行用；ABANDONED 与 COMPLETED 一样阻止同实例再次启动：

原码java

public JobExecution abandon(JobExecution jobExecution) throws JobExecutionAlreadyRunningException {
	Assert.notNull(jobExecution, "JobExecution must not be null");
	if (jobExecution.getStatus().isLessThan(BatchStatus.STOPPING)) {
		throw new JobExecutionAlreadyRunningException(
				"JobExecution is running or complete and therefore cannot be aborted");
	}
	if (logger.isInfoEnabled()) {
		logger.info("Aborting job execution: " + jobExecution);
	}
	jobExecution.upgradeStatus(BatchStatus.ABANDONED);
	jobExecution.setEndTime(LocalDateTime.now());
	jobRepository.update(jobExecution);

	return jobExecution;
}

recover(execution)（6.0 新增）：把“进程突然死亡导致卡在 STARTED”的执行标为 FAILED，并在上下文写 batch.recovered，使其重新可重启；它只处理 running 状态的 step，不碰已完成/UNKNOWN：

原码java

	for (StepExecution stepExecution : jobExecution.getStepExecutions()) {
		BatchStatus stepStatus = stepExecution.getStatus();
		if (stepStatus.isRunning()) {
			stepExecution.setStatus(BatchStatus.FAILED);
			stepExecution.setEndTime(LocalDateTime.now());
			stepExecution.getExecutionContext().put("batch.recovered", true);
			jobRepository.update(stepExecution);
		}
	}

	jobExecution.setStatus(BatchStatus.FAILED);
	jobExecution.setEndTime(LocalDateTime.now());
	jobExecution.getExecutionContext().put("batch.recovered", true);
	jobRepository.update(jobExecution);

	return jobExecution;
}

实验 batchAbandonRecover 验证了三段：abandon 后同参数启动被拒（AlreadyComplete）；把执行人为改回 STARTED 后启动被拒（AlreadyRunning）；recover 之后 restart 被允许并产生新的 JobExecution（step 仍失败，所以新执行也 FAILED，但“重启本身被允许”）。

运维口诀：先看状态再选操作——FAILED/STOPPED 用 restart，STARTED/STARTING 卡死用 recover，确定不再重试用 abandon；UNKNOWN 只能人工查库后决定（recover 会拒绝）。

Quartz：持久调度循环

Quartz 的存储模型只有 11 张表，本页关心 5 张：QRTZ_JOB_DETAILS（作业定义）、QRTZ_TRIGGERS（触发器与下次触发时间、状态）、QRTZ_FIRED_TRIGGERS（本次触发的占用记录）、QRTZ_SCHEDULER_STATE（实例心跳）、QRTZ_LOCKS（行锁）。

调度线程的主循环每轮做四件事：取“30 秒窗口内到点的触发器”→ 等到点 → 置为执行中并推进下次触发时间 → 交给线程池跑 JobRunShell：

原码java

long now = System.currentTimeMillis();

clearSignaledSchedulingChange();
try {
    triggers = qsRsrcs.getJobStore().acquireNextTriggers(
            now + idleWaitTime, Math.min(availThreadCount, qsRsrcs.getMaxBatchSize()), qsRsrcs.getBatchTimeWindow());
    acquiresFailed = 0;

抢占是条件 UPDATE：acquireNextTrigger 对每个候选执行 UPDATE ... SET TRIGGER_STATE='ACQUIRED' WHERE TRIGGER_NAME=? AND TRIGGER_GROUP=? AND TRIGGER_STATE='WAITING'，rowsUpdated<=0 说明被别人抢走，跳过。
被选中的触发器立刻推进 NEXT_FIRE_TIME：triggerFired 先校验状态仍是 ACQUIRED，写 QRTZ_FIRED_TRIGGERS 置 EXECUTING，再调 trigger.triggered(cal) 计算下一次触发时间并写回；@DisallowConcurrentExecution 的作业会让同 job 的其他触发器进入 BLOCKED：

原码java

    Date prevFireTime = trigger.getPreviousFireTime();

    // call triggered - to update the trigger's next-fire-time state...
    trigger.triggered(cal);

    String state = STATE_WAITING;
    boolean force = true;

    if (job.isConcurrentExecutionDisallowed()) {
        state = STATE_BLOCKED;
        force = false;
        try {
            getDelegate().updateTriggerStatesForJobFromOtherState(conn, job.getKey(),
                    STATE_BLOCKED, STATE_WAITING);
            getDelegate().updateTriggerStatesForJobFromOtherState(conn, job.getKey(),
                    STATE_BLOCKED, STATE_ACQUIRED);
            getDelegate().updateTriggerStatesForJobFromOtherState(conn, job.getKey(),
                    STATE_PAUSED_BLOCKED, STATE_PAUSED);
        } catch (SQLException e) {
            throw new JobPersistenceException(
                    "Couldn't update states of blocked triggers: "
                            + e.getMessage(), e);
        }
    } 

    if (trigger.getNextFireTime() == null) {
        state = STATE_COMPLETE;
        force = true;
    }

    storeTrigger(conn, trigger, job, true, state, force, false);

    job.getJobDataMap().clearDirtyFlag();

    return new TriggerFiredBundle(job, trigger, cal, trigger.getKey().getGroup()
            .equals(Scheduler.DEFAULT_RECOVERY_GROUP), new Date(), trigger
            .getPreviousFireTime(), prevFireTime, trigger.getNextFireTime());
}

整批拿取与状态切换都在 TRIGGER_ACCESS 锁里（triggersFired、triggeredJobComplete 同样），行锁实现见下一节。

实验 quartzPersisted：建一个 60 秒后触发的触发器，TRIGGER_STATE=WAITING；调度器 shutdown(false) 后新起实例，QRTZ_JOB_DETAILS 仍是 1 行、触发器仍 WAITING、下次触发时间在未来——调度状态不在内存里。

行锁与集群互斥

isClustered=true 时 Quartz 强制启用数据库行锁（即使 useDBLocks 没显式设），默认实现是 StdRowLockSemaphore：对 QRTZ_LOCKS 的一行做 SELECT ... FOR UPDATE，没有行就插入（带重试，应对首次并发的插入竞争）：

原码java

public static final String SELECT_FOR_LOCK = "SELECT * FROM "
        + TABLE_PREFIX_SUBST + TABLE_LOCKS + " WHERE " + COL_SCHEDULER_NAME + " = " + SCHED_NAME_SUBST
        + " AND " + COL_LOCK_NAME + " = ? FOR UPDATE";

public static final String INSERT_LOCK = "INSERT INTO "
    + TABLE_PREFIX_SUBST + TABLE_LOCKS + "(" + COL_SCHEDULER_NAME + ", " + COL_LOCK_NAME + ") VALUES (" 
    + SCHED_NAME_SUBST + ", ?)";

锁的持有范围是一个事务：executeInNonManagedTXLock 先 obtainLock，再执行回调，提交后 releaseLock，异常则回滚并释放：

原码java

protected <T> T executeInNonManagedTXLock(
        String lockName, 
        TransactionCallback<T> txCallback, final TransactionValidator<T> txValidator) throws JobPersistenceException {
    boolean transOwner = false;
    Connection conn = null;
    try {
        if (lockName != null) {
            // If we aren't using db locks, then delay getting DB connection 
            // until after acquiring the lock since it isn't needed.
            if (getLockHandler().requiresConnection()) {
                conn = getNonManagedTXConnection();
            }

            transOwner = getLockHandler().obtainLock(conn, lockName);
        }

        if (conn == null) {
            conn = getNonManagedTXConnection();
        }

        final T result = txCallback.execute(conn);
        try {
            commitConnection(conn);
        } catch (JobPersistenceException e) {
            rollbackConnection(conn);
            if (txValidator == null || !retryExecuteInNonManagedTXLock(lockName, conn1 -> txValidator.validate(conn1, result))) {
                throw e;
            }
        }

        Long sigTime = clearAndGetSignalSchedulingChangeOnTxCompletion();
        if(sigTime != null && sigTime >= 0) {
            signalSchedulingChangeImmediately(sigTime);
        }

        return result;
    } catch (JobPersistenceException e) {
        rollbackConnection(conn);
        throw e;
    } catch (RuntimeException e) {
        rollbackConnection(conn);
        throw new JobPersistenceException("Unexpected runtime exception: "
                + e.getMessage(), e);
    } finally {
        try {
            releaseLock(lockName, transOwner);
        } finally {
            cleanupConnection(conn);
        }
    }
}

JobStoreTX 把这个方法直接当作实现，所以每次持锁都是“连接级事务锁”：

原码java

@Override
protected Object executeInLock(
        String lockName, 
        TransactionCallback txCallback) throws JobPersistenceException {
    return executeInNonManagedTXLock(lockName, txCallback, null);
}

实验 quartzCluster：同一库上两个 isClustered=true 的调度器（实例 u46-a/u46-b），一个 700ms 后触发的作业——两实例合计只触发 1 次；QRTZ_SCHEDULER_STATE 2 行（各自心跳），QRTZ_LOCKS 2 行（TRIGGER_ACCESS 与 STATE_ACCESS）。

非集群模式的对照更重要：isClustered=false 时两个实例不建锁行，实验改为观察“运行期死实例的残留”：插入一个心跳过期 20 秒、占着 ACQUIRED 触发的死实例行，等 4 秒——死实例心跳行还在、孤立触发仍是 ACQUIRED。因为非集群模式下，清理动作只在启动时的 recoverJobs() 里跑一次（把所有 ACQUIRED 放回 WAITING、清空 FIRED_TRIGGERS），运行期没有任何周期性接管。观察行 quartzNonCluster locks=0 deadRowKept=true orphanStillAcquired=true。

结论：“多实例部署”不是把同一份配置拷几遍就行，必须 isClustered=true（以及同一套表、同一 SCHED_NAME），否则等于几个各自为政的调度器共用一个库。

心跳与故障接管

集群模式下每个实例有一个 ClusterManager 线程，按 clusterCheckinInterval（默认 7500ms）向 QRTZ_SCHEDULER_STATE 写心跳，并检查其它实例是否过期。判据是“上次心跳 + max(间隔, 距我上次心跳的时间) + 7500ms”已经过去：

原码java

protected long calcFailedIfAfter(SchedulerStateRecord rec) {
    return rec.getCheckinTimestamp() +
        Math.max(rec.getCheckinInterval(), 
                (System.currentTimeMillis() - lastCheckin)) +
        7500L;
}

发现死实例后，接管者拿 STATE_ACCESS 与 TRIGGER_ACCESS 两把锁，做三件事：把该实例的 BLOCKED 触发器放回 WAITING、ACQUIRED 放回 WAITING、对 requestsRecovery=true 且执行中的作业补一条 RECOVERING_JOBS 组的恢复触发器（带 IGNORE_MISFIRE 指令与原始触发信息），最后删掉死实例的心跳行：

原码java

protected void clusterRecover(Connection conn, List<SchedulerStateRecord> failedInstances)
    throws JobPersistenceException {

    if (!failedInstances.isEmpty()) {

        long recoverIds = System.currentTimeMillis();

        logWarnIfNonZero(failedInstances.size(),
                "ClusterManager: detected " + failedInstances.size()
                        + " failed or restarted instances.");
        try {
            for (SchedulerStateRecord rec : failedInstances) {
                getLog().info("ClusterManager: Scanning for instance \"{}\"'s failed in-progress jobs.", rec.getSchedulerInstanceId());

                List<FiredTriggerRecord> firedTriggerRecs = getDelegate()
                        .selectInstancesFiredTriggerRecords(conn,
                                rec.getSchedulerInstanceId());

接管的动作在同一个循环里：对每个 firing 记录分类处理，requestsRecovery 的作业补一条带原始触发信息的恢复触发器，最后删除死实例的心跳行：

原码java

} else if (ftRec.isJobRequestsRecovery()) {
    // handle jobs marked for recovery that were not fully
    // executed..
    if (jobExists(conn, jKey)) {
        @SuppressWarnings("deprecation")
        SimpleTriggerImpl rcvryTrig = new SimpleTriggerImpl(
                "recover_"
                        + rec.getSchedulerInstanceId()
                        + "_"
                        + recoverIds++,
                Scheduler.DEFAULT_RECOVERY_GROUP,
                new Date(ftRec.getScheduleTimestamp()));
        rcvryTrig.setJobName(jKey.getName());
        rcvryTrig.setJobGroup(jKey.getGroup());
        rcvryTrig.setMisfireInstruction(SimpleTrigger.MISFIRE_INSTRUCTION_IGNORE_MISFIRE_POLICY);
        rcvryTrig.setPriority(ftRec.getPriority());
        JobDataMap jd = getDelegate().selectTriggerJobDataMap(conn, tKey.getName(), tKey.getGroup());
        jd.put(Scheduler.FAILED_JOB_ORIGINAL_TRIGGER_NAME, tKey.getName());
        jd.put(Scheduler.FAILED_JOB_ORIGINAL_TRIGGER_GROUP, tKey.getGroup());
        jd.put(Scheduler.FAILED_JOB_ORIGINAL_TRIGGER_FIRETIME_IN_MILLISECONDS, String.valueOf(ftRec.getFireTimestamp()));
        jd.put(Scheduler.FAILED_JOB_ORIGINAL_TRIGGER_SCHEDULED_FIRETIME_IN_MILLISECONDS, String.valueOf(ftRec.getScheduleTimestamp()));
        rcvryTrig.setJobDataMap(jd);

        rcvryTrig.computeFirstFireTime(null);
        storeTrigger(conn, rcvryTrig, null, false,
                STATE_WAITING, false, true);
        recoveredCount++;

实验 clusterTakeover：存活实例 checkinInterval=1000；人为插入心跳过期 20 秒的死实例与一条 ACQUIRED 触发；4 秒内死实例心跳行被删除、孤立触发被清理（clusterTakeover deadRowRemoved=true orphanCleaned=true）。

接管 ≠ 业务完成：恢复触发器只是让作业再执行一次；如果原执行已经产生了部分副作用，重复只能由业务幂等处理（见“调度成功不等于业务成功”）。

misfire 语义

触发器到点时调度器没来得及处理（停机、GC 停顿、线程池满、刚启动）叫 misfire。判据是“当前时间 − 阈值”已经超过 NEXT_FIRE_TIME，阈值默认 60000ms；MisfireHandler 每约一个阈值周期扫一次 WAITING 且已过期的触发器，交给触发器自己的 updateAfterMisfire：

原码java

protected boolean updateMisfiredTrigger(Connection conn,
        TriggerKey triggerKey, String newStateIfNotComplete, boolean forceState)
    throws JobPersistenceException {
    try {

        OperableTrigger trig = retrieveTrigger(conn, triggerKey);

        long misfireTime = System.currentTimeMillis();
        if (getMisfireThreshold() > 0) {
            misfireTime -= getMisfireThreshold();
        }

        if (trig.getNextFireTime().getTime() > misfireTime) {
            return false;
        }

        doUpdateOfMisfiredTrigger(conn, trig, forceState, newStateIfNotComplete, false);

        return true;

    } catch (Exception e) {
        throw new JobPersistenceException(
                "Couldn't update misfired trigger '" + triggerKey + "': " + e.getMessage(), e);
    }
}

CronTrigger 的两种指令：

原码java

public void updateAfterMisfire(org.quartz.Calendar cal) {
    int instr = getMisfireInstruction();

    if(instr == Trigger.MISFIRE_INSTRUCTION_IGNORE_MISFIRE_POLICY)
        return;

    if (instr == MISFIRE_INSTRUCTION_SMART_POLICY) {
        instr = MISFIRE_INSTRUCTION_FIRE_ONCE_NOW;
    }

    if (instr == MISFIRE_INSTRUCTION_DO_NOTHING) {
        Date newFireTime = getFireTimeAfter(new Date());
        while (newFireTime != null && cal != null
                && !cal.isTimeIncluded(newFireTime.getTime())) {
            newFireTime = getFireTimeAfter(newFireTime);
        }
        setNextFireTime(newFireTime);
    } else if (instr == MISFIRE_INSTRUCTION_FIRE_ONCE_NOW) {
        setNextFireTime(new Date());
    }
}

FIRE_ONCE_NOW（SMART 对 cron 的默认）：把下次触发时间改成“现在”，错过多少次都只补一次。
DO_NOTHING：把下次触发时间改成“现在之后的下一个日历点”，不补。

SimpleTrigger 的 SMART 按 repeatCount 三分支（0 → FIRE_NOW；无限 → RESCHEDULE_NEXT_WITH_REMAINING_COUNT；有限 → RESCHEDULE_NOW_WITH_EXISTING_COUNT），并且显式 FIRE_NOW 在 repeatCount != 0 时会被改写成 RESCHEDULE_NOW_WITH_REMAINING：

原码java

public void updateAfterMisfire(Calendar cal) {
    int instr = getMisfireInstruction();

    if(instr == Trigger.MISFIRE_INSTRUCTION_IGNORE_MISFIRE_POLICY)
        return;

    if (instr == Trigger.MISFIRE_INSTRUCTION_SMART_POLICY) {
        if (getRepeatCount() == 0) {
            instr = MISFIRE_INSTRUCTION_FIRE_NOW;
        } else if (getRepeatCount() == REPEAT_INDEFINITELY) {
            instr = MISFIRE_INSTRUCTION_RESCHEDULE_NEXT_WITH_REMAINING_COUNT;
        } else {
            // if (getRepeatCount() > 0)
            instr = MISFIRE_INSTRUCTION_RESCHEDULE_NOW_WITH_EXISTING_REPEAT_COUNT;
        }
    } else if (instr == MISFIRE_INSTRUCTION_FIRE_NOW && getRepeatCount() != 0) {
        instr = MISFIRE_INSTRUCTION_RESCHEDULE_NOW_WITH_REMAINING_REPEAT_COUNT;
    }

MISFIRE_INSTRUCTION_IGNORE_MISFIRE_POLICY = -1 是第四种语义：不判 misfire，“错过几次就快速补几次”直到追上。Quartz 的 Trigger 接口注释用 15 秒周期、错过 5 分钟举例——会连续触发 20 次。这是一把双刃剑：适合幂等、可追平的指标采集，不适合有外部副作用的业务。获取触发器的 SQL 里 MISFIRE_INSTR = -1 的触发器被排除在“过期下限”过滤之外，就是为了这个语义：

原码java

String SELECT_NEXT_TRIGGER_TO_ACQUIRE = "SELECT "
    + COL_TRIGGER_NAME + ", " + COL_TRIGGER_GROUP + ", "
    + COL_NEXT_FIRE_TIME + ", " + COL_PRIORITY + " FROM "
    + TABLE_PREFIX_SUBST + TABLE_TRIGGERS + " WHERE "
    + COL_SCHEDULER_NAME + " = " + SCHED_NAME_SUBST
    + " AND " + COL_TRIGGER_STATE + " = ? AND " + COL_NEXT_FIRE_TIME + " <= ? " 
    + "AND (" + COL_MISFIRE_INSTRUCTION + " = -1 OR (" +COL_MISFIRE_INSTRUCTION+ " <> -1 AND "+ COL_NEXT_FIRE_TIME + " >= ?)) "
    + "ORDER BY "+ COL_NEXT_FIRE_TIME + " ASC, " + COL_PRIORITY + " DESC";

实验：

misfireCron：两个 cron 触发器都回拨 5 分钟（阈值 2 秒），FIRE_ONCE_NOW 触发 once=1，DO_NOTHING 触发 nothing=0 且下次触发时间在未来。
misfireSimple：repeatCount=2 的 SimpleTrigger + NOW_WITH_EXISTING_COUNT，回拨 10 分钟后 fired=1——不是按错过的 60 轮补跑。

阈值与周期要按业务窗口设：批次窗口 10 分钟的任务，阈值设 1 分钟没意义；设 1 小时则停机一小时内的所有点都可能被当成一次。misfire 策略是调度器层的补偿，与 Batch 的重启无关——Batch 里“补哪一天”要看 JobInstance 的 identifying 参数。

调度成功不等于业务成功

四组证据：

Quartz 不重试失败作业。JobRunShell 捕获异常、通知监听器、照常推进触发器；JobExecutionException 的 refireImmediately=false 只是“不立即重跑”，不是“失败重试”。实验 quartzFailing：一次性触发器抛异常只执行 1 次；repeatCount=2 的触发器严格按计划执行 3 次，失败不追加次数。
调度器只保证“到点触发一次”（在行锁可用时）。执行结果、业务表终态都不在 Quartz 的元数据里；它连“成功/失败”都不记录。
Batch 的 exit status 与进程退出码是两回事。默认 JobLauncherApplicationRunner 在应用启动时跑 job，FAILED 不阻止 Spring 上下文正常启动；要靠 JobExecutionExitCodeGenerator（Boot 自动配置）把状态映射成退出码，或自己检查 JobExecution.getStatus()。
重复来源有三条：Batch restart（未提交块重跑）、Quartz misfire 补跑与集群接管（恢复触发器）、以及“任务里又套了一层重试”。业务幂等是最后防线。

实验 batchFaultTolerant 的对照也在这里：跳过坏条目让 step “成功”，但业务数据少了一行——step 成功不等于数据完整，skip 必须有审计。

观测与排障

Batch 侧看四张表：

表	看什么
`BATCH_JOB_INSTANCE`	`JOB_NAME`、`JOB_KEY`、实例数是否符合“每天一个”
`BATCH_JOB_EXECUTION`	`STATUS`、`EXIT_CODE`、`EXIT_MESSAGE`、`START_TIME/END_TIME`、同一实例的执行次数
`BATCH_STEP_EXECUTION`	`READ/WRITE/COMMIT/ROLLBACK_COUNT`、`READ/WRITE/PROCESS_SKIP_COUNT`、`STATUS`
`BATCH_STEP_EXECUTION_CONTEXT`	checkpoint（`SHORT_CONTEXT` 可读；长上下文在 `SERIALIZED_CONTEXT`）

典型排障顺序：“实例是不是新的”（JOB_KEY）→“这次为什么启动/被拒”（异常类型）→“哪一步失败”（STEP_EXECUTION）→“块提交到哪”（COMMIT/ROLLBACK 计数与上下文）→“重跑会不会重复”（业务表唯一键与最后一块边界）。

Quartz 侧看四张表：

表	看什么
`QRTZ_TRIGGERS`	`TRIGGER_STATE`（WAITING/ACQUIRED/BLOCKED/COMPLETE/ERROR/PAUSED）、`NEXT_FIRE_TIME`、`MISFIRE_INSTR`
`QRTZ_FIRED_TRIGGERS`	`STATE`（ACQUIRED/EXECUTING）、`INSTANCE_NAME`；残留的 ACQUIRED 是死实例或异常路径的痕迹
`QRTZ_SCHEDULER_STATE`	`LAST_CHECKIN_TIME` 与 `CHECKIN_INTERVAL`；过期的行说明 ClusterManager 没在跑
`QRTZ_LOCKS`	集群模式下应有两行 `TRIGGER_ACCESS`/`STATE_ACCESS`；缺失说明没开集群

BLOCKED 是正常状态（非并发作业正在运行），不是故障；长时间 BLOCKED 要先看 FIRED_TRIGGERS 里那条 EXECUTING 是否真的还在跑。

容量与设计取舍

Batch 的吞吐三参数：chunkSize、提交事务成本、reader 的批量能力。chunk 越大，提交次数越少、失败重做越多；chunk 越小，提交越频繁、scan 成本越可控。一个可算的起点：每条处理成本 t，chunk 大小 c，事务提交成本 s，则吞吐约 c / (c·t + s)；失败重做期望成本约 c/2 × 失败率。实验里 6 条数据、chunk=2 的“失败重读”正好是 2 条（失败块大小）。至少两种合理选择：小 chunk（如 10-50，适合失败代价高的外部调用）与大 chunk（如 500-1000，适合纯本地计算 + faultTolerant + skip）。

Quartz 的调度容量：每轮扫描的 SQL 成本 + 线程池大小 + 锁竞争。org.quartz.scheduler.batchTriggerAcquisitionMaxCount 默认 1；acquireTriggersWithinLock 默认 false——单条拿取时不加锁，靠条件 UPDATE 的 CAS 语义防重；批量拿取（maxCount>1）自动加锁。触发器量很大时，QRTZ_TRIGGERS 上的 (SCHED_NAME, TRIGGER_STATE, NEXT_FIRE_TIME) 索引与定期清理 COMPLETE 触发器是运维重点。

三种组合方式，责任不同：

方案	谁负责“到点”	谁负责“运行模型”	代价
只用 Batch + `@Scheduled`	内存定时器	Batch	多实例会重复启动；无 misfire 语义；进程退出丢失调度
Quartz 触发 Batch（Quartz Job 里调用 `JobOperator.start`）	Quartz（持久、集群）	Batch	两套元数据、两套幂等责任；作业失败不影响触发器推进
依赖平台调度器（xxl-job 等）	平台	Batch	组件更多；平台侧幂等与重试语义要单独核对（见 U85）

演进方向：任务数增长 → Quartz 的扫描与锁成为瓶颈时可换平台调度器；数据量增长 → Batch 的并行（多线程 step、本地分区）优先于远程分区（后者需要消息中间件与多进程，本票未运行）；合规要求增长 → 把 skip/失败/恢复都落审计表，而不是只依赖元数据表。

实验读法

对应 labs/batch-quartz 的 19 个测试，建议按观察行复读（全部来自同一次 exit 0 的运行）：

U46 batchIdentity ...：同一性来自名字 + identifying 参数。
U46 batchRestart firstStatus=FAILED rowsAfterFailure=2 finalRows=6 reads=10 commits=2 rollbacks=0：chunk 回滚与 checkpoint 续读。
U46 batchSkipWithoutFt status=FAILED writeSkip=0 / U46 batchFaultTolerant status=COMPLETED rows=3 writeSkips=1：fault-tolerant 开关的真实作用。
U46 batchStepSkip step1Runs=1 step2Runs=2 step1Executions=1：已完成 step 不重跑。
U46 batchConcurrentLaunch rejected=true executions=1：异步启动下的并发互斥。
U46 batchAbandonRecover abandoned=true recoverUnstuck=true restartNewExecution=true：abandon/recover/restart 三件套。
U46 quartzPersisted jobRows=1 triggerState=WAITING：调度状态跨进程。
U46 quartzCluster fired=1 states=2 locks=2 / U46 quartzNonCluster locks=0 deadRowKept=true orphanStillAcquired=true：集群开与不开的差别。
U46 clusterTakeover deadRowRemoved=true orphanCleaned=true：心跳过期与接管。
U46 misfireCron once=1 nothing=0 nothingNextInFuture=true / U46 misfireSimple fired=1 ...：misfire 两族语义。
U46 quartzFailing noRetry once=1 repeat=3：失败不重试。
U46 quartzNonConcurrent fired=2 triggerState=BLOCKED / U46 quartzShutdownWait waited=true executed=1：非并发与关闭等待。
U46 bootQuartz jdbcStore=LocalDataSourceJobStore clustered=false qrtzTables=11 / U46 bootBatchJdbc resourceless=false batchTables=9：Boot 自动配置的条件与默认值。

条件变化与未运行边界

未运行：远程分区（Remote Partitioning）与远程 chunk（spring-batch-integration，需要消息中间件与多进程）；Batch 的多线程 step 与本地分区；JSR-352 / Jakarta Batch 规范实现；Quartz 的 JobStoreCMT/JTA 与 Terracotta 商业版；RAMJobStore 的高可用（本来就不持久）。
未覆盖的拓扑：Quartz 多实例跨机器（本实验是同 JVM 多 Scheduler 对象共用同一库）、跨机房时钟偏差、数据库主从切换与网络分区。
未覆盖的故障模型：MySQL 主从切换、磁盘满、Quartz 实例在 EXECUTING 中途被 kill -9（本实验用插入过期行模拟死实例）、Batch 进程在 chunk 提交中途崩溃（用 STARTED + recover 模拟）。
实验操作说明：Quartz 的 misfire 用回拨 NEXT_FIRE_TIME 与 2 秒阈值构造；集群接管用回拨 LAST_CHECKIN_TIME 构造；这些都是对真实判据的受控加速，不是生产时间窗。quartzNonCluster 的观察点必须是“调度器启动之后”再投放死实例行，因为非集群模式的 recoverJobs() 会在启动时一次性清空 FIRED_TRIGGERS——这个顺序本身也是被实验证实才写下来的。
Boot 4.0.9 版本事实：当 DataSource 以 Bean 形式提供（而非 spring.datasource.url 属性）时，Quartz 的建表初始化器无法推断 @@platform@@ 平台占位符，必须显式给 spring.quartz.jdbc.schema 或 platform；否则 Failed to determine DatabaseDriver。这是“自动装配不等于交付保证”的一个具体例子。
教材完成不等于学习者掌握：本页的“已讲清”是审读结论，不是学习效果证明。

闭卷验收

本页 20 道闭卷题收进了下方自测面板：先回忆，再展开参考答案与评分点，最后给三档自评。

来源与版本

来源	版本	用途
spring-batch v6.0.6 源码	6.0.6	JobKey/launcher/step/chunk/容错/仓储/operator
Spring Batch 官方参考文档（6.0）	6.0	隔离级别建议、checkpoint 与重启语义
quartz v2.5.2 源码	2.5.2	调度循环、行锁、集群接管、misfire
Quartz 官方文档	2.5.x	misfire 指令、集群配置说明（与源码对照阅读）
Spring Boot 4.0.9 Batch/Quartz 自动配置源码	4.0.9	自动配置条件、默认值与拆分模块
本机运行	Batch 6.0.6 / Quartz 2.5.2 / Framework 7.0.10 / MySQL 8.4.7 / JDK 21.0.11	19 个测试、18 条观察行、exit 0，证据见 `labs/batch-quartz/evidence/u46-2026-10-31.json`

访问与运行日期：2026-10-31。源码片段为固定 tag 的连续行段，整文件 SHA-256 随每段元数据登记；产品文档、源码与实验是三类证据，不能互相替代。

持久任务：Spring Batch 6 的运行模型与 Quartz 2 的持久调度 ​

问题、术语与三个项目 ​

版本、工程与观察入口 ​

身份：JOB_NAME 与 identifying 参数 ​

运行状态机与“创建执行”的校验链 ​

step 执行：跳过与重启 ​

chunk 事务与 checkpoint ​

容错默认值：不开 faultTolerant 会怎样 ​

仓储事务与并发启动 ​

abandon、recover 与 stop ​

Quartz：持久调度循环 ​

行锁与集群互斥 ​

心跳与故障接管 ​

misfire 语义 ​

调度成功不等于业务成功 ​

观测与排障 ​

容量与设计取舍 ​

实验读法 ​

条件变化与未运行边界 ​

闭卷验收 ​

来源与版本 ​

持久任务：Spring Batch 6 的运行模型与 Quartz 2 的持久调度

问题、术语与三个项目

版本、工程与观察入口

身份：JOB_NAME 与 identifying 参数

运行状态机与“创建执行”的校验链

step 执行：跳过与重启

chunk 事务与 checkpoint

容错默认值：不开 faultTolerant 会怎样

仓储事务与并发启动

abandon、recover 与 stop

Quartz：持久调度循环

行锁与集群互斥

心跳与故障接管

misfire 语义

调度成功不等于业务成功

观测与排障

容量与设计取舍

实验读法

条件变化与未运行边界

闭卷验收

来源与版本