JobRepository 详解

本章定位：深入理解 JobRepository——Spring Batch 状态管理的基石。它是如何记录每一次执行、如何支持断点续传、如何在内存和数据库模式间选择。

定义与作用

JobRepository 是 Spring Batch 的持久化机制，负责存储和管理 JobExecution、StepExecution、ExecutionContext 等元数据。它是实现状态管理、重启能力和历史追踪的物理基础。

飞翔科技架构师白歌在团队技术分享中强调：

"如果把 Spring Batch 比作一个物流系统，JobRepository 就是它的数据库——每一批货（Chunk）的签收记录、每一次运输（JobExecution）的状态、每一条路线（Step）的进度，都在里面。没有它，你无法知道哪些货已经送到、哪些还在路上。"

核心原理

JobRepository 的数据流

核心元数据表

表名	存储内容	关键字段	写入时机
`BATCH_JOB_INSTANCE`	Job 实例定义	JOB_INSTANCE_ID, JOB_NAME, JOB_KEY	Job 首次启动时
`BATCH_JOB_EXECUTION`	Job 执行记录	JOB_EXECUTION_ID, STATUS, START_TIME, END_TIME	启动时创建，结束时更新
`BATCH_JOB_EXECUTION_PARAMS`	每次执行的参数	JOB_EXECUTION_ID, KEY_NAME, KEY_TYPE, STRING_VAL	与 JobExecution 同时创建
`BATCH_STEP_EXECUTION`	Step 执行记录	STEP_EXECUTION_ID, STATUS, READ_COUNT, WRITE_COUNT	每个 Step 开始/结束时
`BATCH_STEP_EXECUTION_CONTEXT`	Step 执行上下文	STEP_EXECUTION_ID, SHORT_CONTEXT (JSON)	每个 Chunk 提交后
`BATCH_JOB_EXECUTION_CONTEXT`	Job 执行上下文	JOB_EXECUTION_ID, SHORT_CONTEXT (JSON)	Job 执行期间

内存 vs 数据库模式

模式	实现类	适用场景	优缺点
内存模式	`MapJobRepositoryFactoryBean`	测试、开发、演示	快但重启后数据丢失，无法查询历史
数据库模式	`JDBC JobRepository`	生产环境	持久化、支持重启、可查询历史

# 数据库模式配置（生产推荐）
spring.datasource.url=jdbc:mysql://localhost:3306/batch_meta
spring.batch.jdbc.initialize-schema=always  # 自动创建元数据表
spring.batch.jdbc.table-prefix=BATCH_       # 表名前缀

完整示例

场景一：飞翔科技——通过 JobExplorer 查询作业历史

背景：产品经理孔蓝需要一个管理界面查看所有批处理作业的执行历史。小崔用 JobExplorer（JobRepository 的只读查询接口）实现。

@RestController
@RequestMapping("/api/batch/history")
public class BatchHistoryController {

    private final JobExplorer jobExplorer;

    public BatchHistoryController(JobExplorer jobExplorer) {
        this.jobExplorer = jobExplorer;
    }

    @GetMapping("/jobs/{jobName}")
    public List<Map<String, Object>> getJobHistory(@PathVariable String jobName) {
        List<JobInstance> instances = jobExplorer.findJobInstancesByJobName(jobName, 0, 20);

        return instances.stream().map(instance -> {
            List<JobExecution> executions = jobExplorer.getJobExecutions(instance);
            return Map.of(
                "instanceId", instance.getId(),
                "executions", executions.stream().map(exec -> Map.of(
                    "executionId", exec.getId(),
                    "status", exec.getStatus().name(),
                    "startTime", exec.getStartTime(),
                    "endTime", exec.getEndTime(),
                    "exitCode", exec.getExitStatus().getExitCode()
                )).collect(Collectors.toList())
            );
        }).collect(Collectors.toList());
    }

    @GetMapping("/steps/{jobExecutionId}")
    public List<Map<String, Object>> getStepDetails(
            @PathVariable Long jobExecutionId) {
        JobExecution jobExec = jobExplorer.getJobExecution(jobExecutionId);
        if (jobExec == null) return List.of();

        return jobExec.getStepExecutions().stream().map(step -> Map.of(
            "stepName", step.getStepName(),
            "status", step.getStatus().name(),
            "readCount", step.getReadCount(),
            "writeCount", step.getWriteCount(),
            "commitCount", step.getCommitCount(),
            "skipCount", step.getSkipCount(),
            "startTime", step.getStartTime(),
            "endTime", step.getEndTime()
        )).collect(Collectors.toList());
    }
}

运行结果：

GET /api/batch/history/jobs/orderProcessingJob
→ 200 OK
[
  {
    "instanceId": 15,
    "executions": [
      {"executionId": 42, "status": "COMPLETED", "startTime": "...", "endTime": "..."},
      {"executionId": 41, "status": "FAILED", "startTime": "...", "endTime": "..."}
    ]
  }
]

GET /api/batch/history/steps/42
→ 200 OK
[
  {"stepName": "cleanupStep", "status": "COMPLETED", "readCount": 0, "writeCount": 0},
  {"stepName": "importStep", "status": "COMPLETED", "readCount": 5000, "writeCount": 4980,
   "commitCount": 100, "skipCount": 20},
  {"stepName": "reportStep", "status": "COMPLETED", "readCount": 0, "writeCount": 0}
]

场景二：自定义 JobRepository 配置

在需要自定义序列化器或表前缀时的配置：

@Configuration
@EnableBatchProcessing
public class CustomJobRepositoryConfig {

    @Bean
    public JobRepository customJobRepository(DataSource dataSource,
                                               PlatformTransactionManager tx) throws Exception {
        JobRepositoryFactoryBean factory = new JobRepositoryFactoryBean();
        factory.setDataSource(dataSource);
        factory.setTransactionManager(tx);
        factory.setTablePrefix("CUSTOM_BATCH_");  // 自定义表前缀
        factory.setDatabaseType("MYSQL");
        factory.setMaxVarCharLength(2500);        // ExecutionContext 最大长度
        factory.afterPropertiesSet();
        return factory.getObject();
    }
}

易错场景与避坑

反例一：生产环境使用 H2 内存数据库

# ❌ 生产环境：H2 内存模式 → 重启后元数据全部丢失
spring.datasource.url=jdbc:h2:mem:batch

后果：失败 Job 无法重启，历史无法查询。

正确做法：生产环境使用 MySQL/PostgreSQL 等持久化数据库。

反例二：ExecutionContext 过大导致序列化失败

// ❌ 在 ExecutionContext 中存入大对象
executionContext.put("bigList", hugeListOfObjects);

问题：ExecutionContext 序列化为 JSON 存储在 SHORT_CONTEXT 列中，列有长度限制（默认 2500 字符）。超大数据可能导致截断或写入失败。

正确做法：ExecutionContext 只存简单的恢复点（行号、ID），不存完整数据对象。

面试高频考点

Q1：JobRepository 存储在内存和数据库的区别？

内存模式（MapJobRepositoryFactoryBean）适合测试环境，应用重启后元数据丢失。数据库模式持久化元数据，支持生产环境的状态查询、作业重启和历史追踪。生产环境必须使用数据库模式。

Q2：如何查看 Step 的只读、写入和跳过记录数？

通过 JobExplorer 查询 BATCH_STEP_EXECUTION 表，其中 READ_COUNT、WRITE_COUNT、SKIP_COUNT、COMMIT_COUNT 字段精确记录了每次 Step 执行的统计信息。也可以通过 Spring Cloud Data Flow 的 UI 可视化查看。

Q3：ExecutionContext 什么时候被持久化？

每个 Chunk 提交后，框架调用 ItemStream.update(executionContext)，然后 JobRepository 将更新后的 ExecutionContext 持久化到 BATCH_STEP_EXECUTION_CONTEXT 表。这意味着即使在 Chunk 中间崩溃，上一个 Chunk 的进度已经被安全保存。

上一章：JobLauncher 详解下一章：Step 详解