java batch 怎么写

Java中实现批处理（Batch Processing）有多种方式，具体取决于应用场景、数据量大小以及性能要求,以下是几种常见的实现方式及其详细步骤：

基础Java循环实现批处理

编写Java程序

创建一个简单的Java程序，用于演示如何通过循环实现批处理，假设我们需要处理一个包含大量数据的列表,并将其分批处理。

import java.util.ArrayList;
import java.util.List;
public class BatchProcessingExample {
    public static void main(String[] args) {
        // 模拟大量数据
        List<Integer> data = new ArrayList<>();
        for (int i = 0; i < 1000; i++) {
            data.add(i);
        }
        // 定义每批处理的数据量
        int batchSize = 100;
        // 分批处理数据
        for (int i = 0; i < data.size(); i += batchSize) {
            int end = Math.min(i + batchSize, data.size());
            List<Integer> batch = data.subList(i, end);
            processBatch(batch);
        }
    }
    private static void processBatch(List<Integer> batch) {
        // 模拟处理逻辑
        System.out.println("Processing batch: " + batch);
    }
}

编译和运行

使用javac命令编译Java程序，然后运行生成的.class文件，或者,你也可以将其打包为JAR文件并运行。

使用Spring Batch实现批处理

Spring Batch是一个功能强大的批处理框架，适用于企业级应用，它提供了丰富的功能，如事务管理、重试机制、并发执行等。

添加依赖

在Maven项目中，添加Spring Batch依赖：

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-batch</artifactId>
</dependency>

配置Spring Batch

创建一个配置类,定义Job和Step：

import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.core.launch.support.RunIdIncrementer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
@EnableBatchProcessing
public class BatchConfig {
    private final JobBuilderFactory jobBuilderFactory;
    private final StepBuilderFactory stepBuilderFactory;
    public BatchConfig(JobBuilderFactory jobBuilderFactory, StepBuilderFactory stepBuilderFactory) {
        this.jobBuilderFactory = jobBuilderFactory;
        this.stepBuilderFactory = stepBuilderFactory;
    }
    @Bean
    public Job userJob(Step userStep) {
        return jobBuilderFactory.get("userJob")
                .incrementer(new RunIdIncrementer())
                .flow(userStep)
                .end()
                .build();
    }
    @Bean
    public Step userStep(ItemReader<User> reader, ItemProcessor<User, ProcessedUser> processor, ItemWriter<ProcessedUser> writer) {
        return stepBuilderFactory.get("userStep")
                .<User, ProcessedUser>chunk(100)
                .reader(reader)
                .processor(processor)
                .writer(writer)
                .build();
    }
}

实现ItemReader、ItemProcessor和ItemWriter

这些组件分别负责读取数据、处理数据和写入数据。

import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import java.util.List;
@Configuration
public class BatchComponents {
    @Bean
    public ItemReader<User> reader() {
        // 实现从数据源读取数据的逻辑
        return new UserItemReader();
    }
    @Bean
    public ItemProcessor<User, ProcessedUser> processor() {
        // 实现数据处理逻辑
        return new UserProcessor();
    }
    @Bean
    public ItemWriter<ProcessedUser> writer() {
        // 实现数据写入逻辑
        return new ProcessedUserWriter();
    }
}

启动Job

可以通过命令行Runner或者在应用启动时自动执行Job。

import org.springframework.batch.core.Job;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.JobParameters;
import org.springframework.batch.core.launch.JobLauncher;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Component;
@Component
public class JobLauncherRunner implements CommandLineRunner {
    @Autowired
    private JobLauncher jobLauncher;
    @Autowired
    private Job importUserJob;
    @Override
    public void run(String... args) throws Exception {
        JobExecution execution = jobLauncher.run(importUserJob, new JobParameters());
        System.out.println("Job Exit Status: " + execution.getStatus());
    }
}

使用Easy Batch实现轻量级批处理

Easy Batch是一个轻量级的批处理框架，适用于简单的批处理需求，它提供了一些消除繁琐的任务模板代码，如读取、筛选、解析和验证输入数据。

添加依赖

在Maven项目中，添加Easy Batch依赖：

<dependency>
    <groupId>com.github.jmcortex</groupId>
    <artifactId>easy-batch</artifactId>
    <version>1.0</version>
</dependency>

编写批处理任务

使用Easy Batch的流畅API配置批处理任务。

import com.github.jmcortex.easybatch.core.filter.Filter;
import com.github.jmcortex.easybatch.core.record.Record;
import com.github.jmcortex.easybatch.core.writer.RecordWriter;
import com.github.jmcortex.easybatch.core.reader.FileRecordReader;
import com.github.jmcortex.easybatch.core.job.JobBuilder;
import com.github.jmcortex.easybatch.core.job.JobExecutor;
import com.github.jmcortex.easybatch.core.job.JobReport;
import java.io.File;
import java.util.Arrays;
import java.util.List;
public class EasyBatchExample {
    public static void main(String[] args) {
        // 定义数据源和目标文件路径
        File inputFile = new File("input.csv");
        File outputFile = new File("output.csv");
        // 创建Job并配置读取器、过滤器、处理器和写入器
        JobReport report = new JobExecutor()
                .execute(new JobBuilder()
                        .reader(new FileRecordReader(inputFile))
                        .filter(new MyFilter())
                        .mapper(new MyMapper())
                        .writer(new MyWriter(outputFile))
                        .build());
        // 输出处理结果
        System.out.println("Processed records: " + report.getTotalRecords());
    }
}

常见问题及解决方案

内存溢出（OutOfMemoryError）

问题：在处理大量数据时，一次性加载所有数据到内存中处理，容易引发OutOfMemoryError。
解决方案：使用分页读取或流式处理，分批次读取数据，减少内存占用，在Spring Batch中使用PagingItemReader或实现分页逻辑。

事务管理不当

问题：批量处理中，如果一个事务包含太多数据处理操作，一旦失败，回滚成本高，且可能影响数据库性能。
解决方案：合理设置chunkSize，控制每次提交的记录数量，平衡性能与事务安全性，在Spring Batch中设置chunk(100)表示每100条