Spring Batch能否使用NoSQL数据库存储批处理元数据？求示例

是的，Spring Batch支持使用NoSQL数据库（如MongoDB、Firestore等）存储批处理元数据。Spring Batch的元数据管理核心依赖JobRepository接口，默认提供了JDBC实现，但允许开发者通过实现该接口（或利用官方扩展模块）适配NoSQL存储。

Spring Batch 4.3及以上版本提供了官方的MongoDB元数据存储支持，无需完全自定义，步骤如下：

1. 添加依赖 #

在Maven pom.xml中引入Spring Batch和MongoDB批处理模块：

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-batch</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.batch</groupId>
        <artifactId>spring-batch-mongodb</artifactId>
        <version>5.1.0</version> <!-- 匹配你的Spring Batch版本 -->
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-data-mongodb</artifactId>
    </dependency>
</dependencies>

2. 配置MongoDB版JobRepository #

创建配置类，使用MongoJobRepositoryFactoryBean生成适配MongoDB的JobRepository：

import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.repository.support.MongoJobRepositoryFactoryBean;
import org.springframework.batch.support.transaction.ResourcelessTransactionManager;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.transaction.PlatformTransactionManager;

@Configuration
public class BatchMongoConfig {

    @Bean
    public JobRepository jobRepository(MongoTemplate mongoTemplate) throws Exception {
        MongoJobRepositoryFactoryBean factoryBean = new MongoJobRepositoryFactoryBean();
        factoryBean.setMongoTemplate(mongoTemplate);
        factoryBean.setTransactionManager(transactionManager());
        return factoryBean.getObject();
    }

    @Bean
    public PlatformTransactionManager transactionManager() {
        // MongoDB单节点不支持事务，使用无操作事务管理器；副本集环境可替换为MongoTransactionManager
        return new ResourcelessTransactionManager();
    }
}

3. 配置JobLauncher #

将自定义的JobRepository注入JobLauncher：

import org.springframework.batch.core.launch.JobLauncher;
import org.springframework.batch.core.launch.support.SimpleJobLauncher;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class BatchLauncherConfig {

    @Bean
    public JobLauncher jobLauncher(JobRepository jobRepository) throws Exception {
        SimpleJobLauncher jobLauncher = new SimpleJobLauncher();
        jobLauncher.setJobRepository(jobRepository);
        jobLauncher.afterPropertiesSet();
        return jobLauncher;
    }
}

Firestore没有官方的Spring Batch元数据模块，需要自定义实现JobRepository接口，核心是将元数据实体（JobInstance、JobExecution、StepExecution等）持久化到Firestore集合中。

示例框架：

import com.google.cloud.firestore.Firestore;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.JobInstance;
import org.springframework.batch.core.JobParameters;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.repository.JobInstanceAlreadyExistsException;
import org.springframework.batch.core.repository.JobParametersInvalidException;
import org.springframework.stereotype.Component;

import java.util.UUID;

@Component
public class FirestoreJobRepository implements JobRepository {

    private final Firestore firestore;

    public FirestoreJobRepository(Firestore firestore) {
        this.firestore = firestore;
    }

    @Override
    public JobInstance createJobInstance(String jobName, JobParameters jobParameters) throws JobInstanceAlreadyExistsException, JobParametersInvalidException {
        // 生成唯一ID，检查是否存在重复的JobInstance
        Long instanceId = UUID.randomUUID().getMostSignificantBits() & Long.MAX_VALUE;
        JobInstance jobInstance = new JobInstance(instanceId, jobName);
        // 写入Firestore集合
        firestore.collection("job_instances").document(instanceId.toString()).set(jobInstance);
        return jobInstance;
    }

    @Override
    public void updateJobInstance(JobInstance jobInstance) {
        firestore.collection("job_instances").document(jobInstance.getId().toString()).set(jobInstance);
    }

    @Override
    public JobExecution createJobExecution(JobInstance jobInstance, JobParameters jobParameters) throws JobExecutionAlreadyRunningException, JobRestartException, JobInstanceAlreadyCompleteException {
        // 实现JobExecution创建逻辑
        Long executionId = UUID.randomUUID().getMostSignificantBits() & Long.MAX_VALUE;
        JobExecution execution = new JobExecution(jobInstance, executionId, jobParameters);
        firestore.collection("job_executions").document(executionId.toString()).set(execution);
        return execution;
    }

    // 实现JobRepository剩余的方法（updateJobExecution、createStepExecution等）
}

配置类中注册该Bean：

import com.google.cloud.firestore.Firestore;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class BatchFirestoreConfig {

    @Bean
    public JobRepository firestoreJobRepository(Firestore firestore) {
        return new FirestoreJobRepository(firestore);
    }
}

注意：自定义实现时需要处理元数据的一致性，比如通过Firestore的字段版本控制实现乐观锁，避免并发更新冲突。

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