如何通过Spring Batch读取数据并在Job外部获取数据进行后续处理？

Got it, let's break down how to solve this problem perfectly for your use case. The core idea is to capture the data read by Spring Batch's ItemReader and make it accessible outside the job context, so you can run your custom validation (against that external list you can't access in the job) and handle JPA inserts afterward. Here are two reliable approaches:

Approach 1: Use a Thread-Safe Data Holder (For Small to Medium Data Volumes) #

This method stores the read data in a thread-safe component that lives outside the job's scope, so you can retrieve it once the job finishes.

Step 1: Create a Thread-Safe Data Holder Component #

First, make a singleton component to hold your data—this needs to be thread-safe since Spring Batch might run steps in parallel or use chunk processing:

@Component
public class BatchDataHolder {
    private final ConcurrentHashMap<Long, YourDataModel> storedItems = new ConcurrentHashMap<>();

    // Add items as they're read
    public void addItem(YourDataModel item) {
        storedItems.put(item.getId(), item); // Use a unique key to avoid duplicates
    }

    // Fetch all collected items after the job
    public Collection<YourDataModel> getAllItems() {
        return storedItems.values();
    }

    // Clear the holder before each job run to avoid stale data
    public void reset() {
        storedItems.clear();
    }
}

Step 2: Build a Custom ItemWriter to Store Data #

Instead of writing to a database directly, use an ItemWriter that pushes data into your holder:

@Component
public class StoringItemWriter implements ItemWriter<YourDataModel> {

    private final BatchDataHolder dataHolder;

    // Constructor injection (preferred over @Autowired)
    public StoringItemWriter(BatchDataHolder dataHolder) {
        this.dataHolder = dataHolder;
    }

    @Override
    public void write(List<? extends YourDataModel> items) throws Exception {
        items.forEach(dataHolder::addItem);
    }
}

Step 3: Configure Your Batch Job #

Set up your job to use your existing ItemReader and the new StoringItemWriter:

@Configuration
public class BatchJobConfig {

    private final JobBuilderFactory jobBuilderFactory;
    private final StepBuilderFactory stepBuilderFactory;
    private final ItemReader<YourDataModel> customItemReader; // Your existing reader
    private final StoringItemWriter storingItemWriter;

    public BatchJobConfig(JobBuilderFactory jobBuilderFactory, StepBuilderFactory stepBuilderFactory,
                          ItemReader<YourDataModel> customItemReader, StoringItemWriter storingItemWriter) {
        this.jobBuilderFactory = jobBuilderFactory;
        this.stepBuilderFactory = stepBuilderFactory;
        this.customItemReader = customItemReader;
        this.storingItemWriter = storingItemWriter;
    }

    @Bean
    public Step dataReadingStep() {
        return stepBuilderFactory.get("dataReadingStep")
                .<YourDataModel, YourDataModel>chunk(100) // Adjust chunk size to fit your needs
                .reader(customItemReader)
                .writer(storingItemWriter)
                .build();
    }

    @Bean
    public Job dataExtractionJob() {
        return jobBuilderFactory.get("dataExtractionJob")
                .incrementer(new RunIdIncrementer())
                .flow(dataReadingStep())
                .end()
                .build();
    }
}

Step 4: Execute the Job and Process Data Externally #

Create a service to launch the job, fetch the stored data, run your validation, and handle JPA inserts:

@Service
@Slf4j
public class PostJobProcessingService {

    private final JobLauncher jobLauncher;
    private final Job dataExtractionJob;
    private final BatchDataHolder dataHolder;
    private final ExternalValidationService validationService; // Your service with the inaccessible list
    private final JpaEntityRepository jpaRepository; // Your JPA repo for final inserts

    public PostJobProcessingService(JobLauncher jobLauncher, Job dataExtractionJob,
                                    BatchDataHolder dataHolder, ExternalValidationService validationService,
                                    JpaEntityRepository jpaRepository) {
        this.jobLauncher = jobLauncher;
        this.dataExtractionJob = dataExtractionJob;
        this.dataHolder = dataHolder;
        this.validationService = validationService;
        this.jpaRepository = jpaRepository;
    }

    public void runJobAndProcessData() throws JobExecutionException {
        // Reset the holder to avoid leftover data from previous runs
        dataHolder.reset();

        // Launch the batch job
        JobParameters jobParams = new JobParametersBuilder()
                .addLong("runTimestamp", System.currentTimeMillis())
                .toJobParameters();
        JobExecution jobExecution = jobLauncher.run(dataExtractionJob, jobParams);

        // Only process if the job completed successfully
        if (BatchStatus.COMPLETED.equals(jobExecution.getStatus())) {
            Collection<YourDataModel> readItems = dataHolder.getAllItems();
            log.info("Fetched {} items from batch job", readItems.size());

            // Run validation and JPA inserts
            readItems.forEach(item -> {
                // Check if xyz exists in your external list
                boolean isItemValid = validationService.isXyzValid(item.getXyz());
                if (isItemValid) {
                    // Convert your data model to JPA entity and save
                    JpaEntity entity = mapToJpaEntity(item);
                    jpaRepository.save(entity);
                } else {
                    log.warn("Skipping invalid item with xyz: {}", item.getXyz());
                }
            });
        } else {
            log.error("Batch job failed with status: {}", jobExecution.getStatus());
            throw new RuntimeException("Job execution failed - cannot process data");
        }
    }

    // Helper method to map your batch data model to JPA entity
    private JpaEntity mapToJpaEntity(YourDataModel item) {
        JpaEntity entity = new JpaEntity();
        entity.setId(item.getId());
        entity.setXyz(item.getXyz());
        // Map other fields as needed
        return entity;
    }
}

Approach 2: Use Temporary Storage (For Large Data Volumes) #

If you're dealing with huge datasets that would cause memory issues with the in-memory holder, use a temporary database table or Redis to store the read data:

• Temporary Table: Configure an ItemWriter to insert data into a temp table (with a job-specific identifier). After the job completes, query the temp table for all items linked to the job ID, run your validation, insert into final tables, then delete the temp records.
• Redis: Use a Redis hash or set to store items with a job-specific key. After the job, fetch all items by the key, process them, then delete the key from Redis.

Key Notes to Avoid Issues #

• Thread Safety: If using parallel step execution (taskExecutor), ensure your data holder or temporary storage handles concurrent writes properly.
• Job Restarts: Always clear/delete temporary data before launching a job to avoid processing stale records from previous runs.
• Transaction Management: For temporary tables, ensure the ItemWriter uses proper transactions to guarantee data consistency if the job fails mid-run.

内容的提问来源于stack exchange，提问作者R.Henderson