Spring Batch 高级元数据使用指南

核心概念
Spring Batch 的元数据管理机制是批处理系统的"控制中心"，通过 JobRepository 存储作业执行状态，JobLauncher 控制作业启动，而高级功能则通过 JobExplorer 和 JobOperator 提供更精细的控制能力。

一、元数据仓库高级访问

Spring Batch 通过 JobExplorer 提供元数据的只读访问能力，用于监控和查询作业状态：

// Kotlin 配置 JobExplorer
@Bean
fun jobExplorer(dataSource: DataSource): JobExplorer {
    val factoryBean = JobExplorerFactoryBean().apply {
        setDataSource(dataSource)
        setTablePrefix("BATCH_")  // [!code highlight] // 自定义表前缀
    }
    return factoryBean.`object`
}

表前缀的重要性

当存在多套批处理系统时，通过 setTablePrefix() 设置表前缀可避免元数据表冲突，例如 SYSTEM.BATCH_JOB_EXECUTION

常用查询方法示例

// 获取作业实例
val instances = jobExplorer.getJobInstances("csvImportJob", 0, 10)

// 获取运行中作业
val runningJobs = jobExplorer.findRunningJobExecutions("nightlyReportJob")

// 获取作业执行详情
val execution = jobExplorer.getJobExecution(12345L)

---

二、Job 注册中心 (JobRegistry)

实现作业的集中管理和动态注册，核心接口为 JobRegistry：

// 配置 MapJobRegistry
@Bean
fun jobRegistry(): JobRegistry = MapJobRegistry()

作业注册的三种方式

1. JobRegistrySmartInitializingSingleton (推荐)
   Spring 容器初始化完成后自动注册所有单例 Job：

@Bean
fun jobRegistryInitializer(jobRegistry: JobRegistry) = 
    JobRegistrySmartInitializingSingleton(jobRegistry)

2. 自动注册器 (AutomaticJobRegistrar)
   动态加载子上下文中的作业，避免 Bean 名称冲突：

@Bean
fun automaticRegistrar() = AutomaticJobRegistrar().apply {
    setJobLoader(DefaultJobLoader(jobRegistry()))
    setApplicationContextFactories(listOf(
        ClassPathXmlApplicationContextFactory("classpath:/jobs/import-job.xml")
    ))
    afterPropertiesSet()
}

CAUTION

JobRegistryBeanPostProcessor 已在 Spring Batch 5.2+ 废弃，请使用 JobRegistrySmartInitializingSingleton

---

三、Job 操作器 (JobOperator)

提供作业的全生命周期管理能力，包括启动、停止、重启等操作：

// Kotlin 配置 JobOperator
@Bean
fun jobOperator(
    jobExplorer: JobExplorer,
    jobRepository: JobRepository,
    jobRegistry: JobRegistry,
    jobLauncher: JobLauncher
) = SimpleJobOperator().apply {
    this.jobExplorer = jobExplorer
    this.jobRepository = jobRepository
    this.jobRegistry = jobRegistry
    this.jobLauncher = jobLauncher
}

关键操作示例

启动作业

// 启动新作业实例
val executionId = jobOperator.start("invoiceJob", "date=2023-11-01")

停止作业

// 停止运行中的作业
jobOperator.stop(executionId)

重启作业

// 重启失败的作业
jobOperator.restart(failedExecutionId)

---

四、Job 参数增量器

实现作业参数的自动递增，确保每次启动生成新实例：

class RunIdIncrementer : JobParametersIncrementer {
    override fun getNext(parameters: JobParameters?): JobParameters {
        val runId = parameters?.getLong("run.id", 1L) ?: 1L
        return JobParametersBuilder()
            .addLong("run.id", runId + 1, true)
            .toJobParameters()
    }
}

关联到 Job

@Bean
fun dataExportJob(jobRepository: JobRepository) = JobBuilder("dataExport", jobRepository)
    .incrementer(RunIdIncrementer())  // [!code highlight] // 绑定增量器
    .start(exportStep)
    .build()

---

五、作业停止与终止

操作类型	状态变更	是否可重启	适用场景
停止	STOPPED	✅ 是	优雅停止运行中的作业
终止	ABANDONED	❌ 否	服务器崩溃后的手动标记

// 停止作业执行
jobOperator.stop(executionId)

// 标记为终止 (不可重启)
jobRepository.updateExecutionStatus(executionId, BatchStatus.ABANDONED)

WARNING

当进程意外终止时（如 kill -9），必须手动标记作业状态为 ABANDONED，否则重启时会产生状态冲突！

---

六、最佳实践总结

1. 监控体系：使用 JobExplorer 构建作业监控面板
2. 动态作业：通过 AutomaticJobRegistrar 实现模块化作业加载
3. 安全停止：优先使用 JobOperator.stop() 而非强制中断
4. 参数设计：利用增量器确保参数唯一性
5. 状态处理：及时清理 ABANDONED 状态作业释放资源

企业级应用场景

在金融对账系统中，通过 JobOperator 实现：

• 定时启动日终对账作业
• 异常时自动重试三次
• 人工干预终止长时间卡住的任务
• 生成执行报告发送至监控平台

通过掌握这些高级元数据操作技巧，您将能构建出更健壮、更易维护的企业级批处理应用！