Spring Batch 5迁移：如何禁用内部表依赖？

问题背景 #

从Spring Batch 4.x（对应Spring Boot 2.6.6）升级到5.0（对应Spring Boot 3.1.3）后，原批处理应用不依赖Spring Batch用于状态记录的内部表，但新版本默认会配置JDBC型JobRepository，导致执行作业时抛出SQL语法错误，提示找不到BATCH_JOB_INSTANCE等内部表。

原Job配置代码如下：

@Configuration
@EnableBatchProcessing
@RequiredArgsConstructor
public class UpdateAssignedProductsJob {

  private static final int DEFAULT_CHUNK_SIZE = 25;

  public static final String JOB_NAME = JobName.UPDATE_ASSIGNED_PRODUCTS;

  @Bean(UpdateAssignedProductsJob.JOB_NAME)
  public Job job(
      @Qualifier(UpdateAssignedProductsJob.JOB_NAME + "_STEP_ONE") Step stepOne,
      JobRepository jobRepository,
      JobExecutionListener jobExecutionListener,
      @Qualifier("ArticleFlatFilePublisher") ArticleFlatFilePublisher articleFlatFilePublisher
  ) {
    return new JobBuilder(UpdateAssignedProductsJob.JOB_NAME, jobRepository)
        .incrementer(new RunIdIncrementer())
        .flow(stepOne)
        .end()
        .listener(jobExecutionListener)
        .listener(articleFlatFilePublisher)
        .build();
  }

  @Bean(UpdateAssignedProductsJob.JOB_NAME + "_STEP_ONE")
  public Step stepOne(
      @Qualifier("UpdateAssignedProductsJobPagedItemReader") UpdateAssignedProductsJobPagedItemReader itemReader,
      JobRepository jobRepository, PlatformTransactionManager transactionManager,
      @Qualifier("ProductItemProcessor") ProductItemProcessor itemProcessor,
      @Qualifier("ArticleFlatFileWriter") ArticleFlatFileWriter itemWriter
  ) {
     return new StepBuilder(UpdateAssignedProductsJob.JOB_NAME + "_STEP_ONE", jobRepository)
        .<Product, Article>chunk(DEFAULT_CHUNK_SIZE, transactionManager)
        .reader(itemReader)
        .processor(itemProcessor)
        .writer(itemWriter)
        .build();
  }
}

执行作业时抛出的错误片段：

java.lang.IllegalStateException: Failed to execute CommandLineRunner
    at org.springframework.boot.SpringApplication.callRunner(SpringApplication.java:774) [spring-boot-3.1.3.jar:3.1.3]
    ...
Caused by: org.springframework.jdbc.BadSqlGrammarException: PreparedStatementCallback; bad SQL grammar [SELECT JOB_INSTANCE_ID, JOB_NAME
FROM BATCH_JOB_INSTANCE
WHERE JOB_NAME = ?
 and JOB_KEY = ?]
    at org.springframework.jdbc.support.SQLExceptionSubclassTranslator.doTranslate(SQLExceptionSubclassTranslator.java:101) ~[spring-jdbc-6.0.11.jar:6.0.11]
    ...

根据Spring Batch 5.0迁移文档说明：

Job repository/explorer配置更新。
基于Map的job repository/explorer实现在v4中被弃用，并在v5中完全移除。你应该改用基于JDBC的实现。除非你使用自定义的Job repository/explorer实现，否则@EnableBatchProcessing注解会配置一个基于JDBC的JobRepository，这需要应用上下文中存在一个DataSource Bean。DataSource可以指向H2、HSQL等嵌入式数据库，以配合内存型job repository使用。

解决方案 #

由于Spring Batch 5移除了原有的内存型MapJobRepository，需自定义不依赖数据库的JobRepository实现，具体步骤如下：

1. 实现内存型DAO（或空实现） #

如果不需要跟踪作业执行状态，可实现仅做内存存储的DAO（或空实现，所有方法不做持久化操作）。以下是JobInstanceDao的内存实现示例：

import org.springframework.batch.core.JobInstance;
import org.springframework.batch.core.JobParameters;
import org.springframework.batch.core.repository.dao.JobInstanceDao;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicLong;

public class InMemoryJobInstanceDao implements JobInstanceDao {

    private final Map<Long, JobInstance> instancesById = new ConcurrentHashMap<>();
    private final Map<String, Map<JobParameters, JobInstance>> instancesByNameAndParameters = new ConcurrentHashMap<>();
    private final AtomicLong idGenerator = new AtomicLong(1);

    @Override
    public JobInstance createJobInstance(String jobName, JobParameters jobParameters) {
        JobInstance instance = new JobInstance(idGenerator.getAndIncrement(), jobName);
        instancesById.put(instance.getId(), instance);
        instancesByNameAndParameters.computeIfAbsent(jobName, k -> new ConcurrentHashMap<>())
                .put(jobParameters, instance);
        return instance;
    }

    @Override
    public JobInstance getJobInstance(Long jobInstanceId) {
        return instancesById.get(jobInstanceId);
    }

    @Override
    public JobInstance getJobInstance(String jobName, JobParameters jobParameters) {
        return instancesByNameAndParameters.getOrDefault(jobName, Collections.emptyMap()).get(jobParameters);
    }

    @Override
    public List<JobInstance> getJobInstances(String jobName, int start, int count) {
        return instancesById.values().stream()
                .filter(instance -> instance.getJobName().equals(jobName))
                .skip(start)
                .limit(count)
                .toList();
    }

    @Override
    public List<String> getJobNames() {
        return new ArrayList<>(instancesByNameAndParameters.keySet());
    }

    @Override
    public int getJobInstanceCount(String jobName) {
        return instancesByNameAndParameters.getOrDefault(jobName, Collections.emptyMap()).size();
    }
}

同理，需实现JobExecutionDao、StepExecutionDao、ExecutionContextDao的内存版本（或空实现）。

2. 配置自定义JobRepository #

创建配置类，将自定义DAO注入SimpleJobRepository，并配置无资源事务管理器（无需数据库事务）：

import org.springframework.batch.core.JobLauncher;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.repository.dao.*;
import org.springframework.batch.core.repository.support.SimpleJobRepository;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.support.ResourcelessTransactionManager;
import org.springframework.core.task.SyncTaskExecutor;

@Configuration
public class BatchCustomConfig {

    @Bean
    public JobRepository jobRepository() {
        JobInstanceDao jobInstanceDao = new InMemoryJobInstanceDao();
        JobExecutionDao jobExecutionDao = new InMemoryJobExecutionDao(); // 自定义的内存实现
        StepExecutionDao stepExecutionDao = new InMemoryStepExecutionDao(); // 自定义的内存实现
        ExecutionContextDao executionContextDao = new InMemoryExecutionContextDao(); // 自定义的内存实现
        return new SimpleJobRepository(jobInstanceDao, jobExecutionDao, stepExecutionDao, executionContextDao);
    }

    @Bean
    public PlatformTransactionManager transactionManager() {
        // 无需数据库事务，使用无资源事务管理器
        return new ResourcelessTransactionManager();
    }

    @Bean
    public JobLauncher jobLauncher(JobRepository jobRepository) {
        SimpleJobLauncher jobLauncher = new SimpleJobLauncher();
        jobLauncher.setJobRepository(jobRepository);
        jobLauncher.setTaskExecutor(new SyncTaskExecutor());
        return jobLauncher;
    }
}

3. 调整Job配置 #

确保@EnableBatchProcessing不会覆盖自定义JobRepository（或直接移除@EnableBatchProcessing，手动配置所有必要Bean），原Job配置类中注入自定义的JobRepository和PlatformTransactionManager即可正常使用。

注意事项 #

• 该方案适用于不需要作业重启、状态跟踪的场景，生产环境若需状态持久化，建议使用嵌入式数据库或配置Spring Batch自动创建内部表。
• 若使用Spring Boot，需确保自定义的JobRepository和PlatformTransactionManager优先被加载，避免默认JDBC版本覆盖。

内容的提问来源于stack exchange，提问作者CodeBot