当前位置：首页 > news >正文

Spring Batch 高级篇-并行步骤

news 2025/8/19 5:42:07

引言

概念

案例

转视频版

引言

接着上篇：Spring Batch 高级篇-多线程步骤，了解Spring Batch多线程步骤后，接下来一起学习一下Spring Batch 高级功能-并行步骤

概念

并行步骤，指的是某2个或者多个步骤同时执行。比如下图

图中，流程从步骤1执行，然后执行步骤2，步骤3，当步骤2/3执行结束之后，在执行步骤4.

设想一种场景，当读取2个或者多个互不关联的文件时，可以多个文件同时读取，这个就是并行步骤。

案例

需求：现有user-parallel.txt, user-parallel.json 2个文件将它们中数据读入内存

1>编写user-parallel.txt, user-parallel.json

6#zhangsan#14
7#lisi#13
8#wangwu#12
9#zhaoliu#11
10#qianqi#10

[{"id":1, "name":"dafei", "age":18},{"id":2, "name":"xiaofei", "age":17},{"id":3, "name":"zhongfei", "age":16},{"id":4, "name":"laofei", "age":15},{"id":5, "name":"feifei", "age":14}
]

2>编写实体对象

@Getter
@Setter
@ToString
public class User {private Long id;private String name;private int age;
}

3>代码实现

package com.langfeiyes.batch._36_step_parallel;import com.fasterxml.jackson.databind.ObjectMapper;
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.job.builder.FlowBuilder;
import org.springframework.batch.core.job.flow.Flow;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.file.FlatFileItemReader;
import org.springframework.batch.item.file.builder.FlatFileItemReaderBuilder;
import org.springframework.batch.item.json.JacksonJsonObjectReader;
import org.springframework.batch.item.json.JsonItemReader;
import org.springframework.batch.item.json.builder.JsonItemReaderBuilder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;
import org.springframework.core.io.ClassPathResource;
import org.springframework.core.task.SimpleAsyncTaskExecutor;import java.util.List;@SpringBootApplication
@EnableBatchProcessing
public class ParallelStepJob {@Autowiredprivate JobBuilderFactory jobBuilderFactory;@Autowiredprivate StepBuilderFactory stepBuilderFactory;@Beanpublic JsonItemReader<User> jsonItemReader(){ObjectMapper objectMapper = new ObjectMapper();JacksonJsonObjectReader<User> jsonObjectReader = new JacksonJsonObjectReader<>(User.class);jsonObjectReader.setMapper(objectMapper);return new JsonItemReaderBuilder<User>().name("userJsonItemReader").jsonObjectReader(jsonObjectReader).resource(new ClassPathResource("user-parallel.json")).build();}@Beanpublic FlatFileItemReader<User> flatItemReader(){return new FlatFileItemReaderBuilder<User>().name("userItemReader").resource(new ClassPathResource("user-parallel.txt")).delimited().delimiter("#").names("id", "name", "age").targetType(User.class).build();}@Beanpublic ItemWriter<User> itemWriter(){return new ItemWriter<User>() {@Overridepublic void write(List<? extends User> items) throws Exception {items.forEach(System.err::println);}};}@Beanpublic Step jsonStep(){return stepBuilderFactory.get("jsonStep").<User, User>chunk(2).reader(jsonItemReader()).writer(itemWriter()).build();}@Beanpublic Step flatStep(){return stepBuilderFactory.get("step2").<User, User>chunk(2).reader(flatItemReader()).writer(itemWriter()).build();}@Beanpublic Job parallelJob(){//线程1-读user-parallel.txtFlow parallelFlow1 = new FlowBuilder<Flow>("parallelFlow1").start(flatStep()).build();//线程2-读user-parallel.jsonFlow parallelFlow2 = new FlowBuilder<Flow>("parallelFlow2").start(jsonStep()).split(new SimpleAsyncTaskExecutor()).add(parallelFlow1).build();return jobBuilderFactory.get("parallel-step-job").start(parallelFlow2).end().build();}public static void main(String[] args) {SpringApplication.run(ParallelStepJob.class, args);}
}

结果

User(id=6, name=zhangsan, age=14)
User(id=7, name=lisi, age=13)
User(id=8, name=wangwu, age=12)
User(id=9, name=zhaoliu, age=11)
User(id=1, name=dafei, age=18)
User(id=2, name=xiaofei, age=17)
User(id=10, name=qianqi, age=10)
User(id=3, name=zhongfei, age=16)
User(id=4, name=laofei, age=15)
User(id=5, name=feifei, age=14)

解析：

1：jsonItemReader() flatItemReader() 定义2个读入操作，分别读json格式跟普通文本格式

2：parallelJob() 配置job，需要指定并行的flow步骤，先是parallelFlow1然后是parallelFlow2 ， 2个步骤间使用.split(new SimpleAsyncTaskExecutor()) 隔开，表示线程池开启2个线程，分别处理parallelFlow1， parallelFlow2 2个步骤。

到这，本篇就结束了，欲知后事如何，请听下回分解~