门户网站建设 存在的问题,网站开发 发表文章,ICP网站忘记密码,做任务 送科比网站目录
引言
概念 案例
转视频版 引言
接着上篇#xff1a;Spring Batch ItemWriter组件#xff0c;了解Spring Batch ItemWriter处理组件后#xff0c;接下来一起学习一下Spring Batch 高级功能-多线程步骤
概念
默认的情况下#xff0c;步骤基本上在单线程中执行Spring Batch ItemWriter组件了解Spring Batch ItemWriter处理组件后接下来一起学习一下Spring Batch 高级功能-多线程步骤
概念
默认的情况下步骤基本上在单线程中执行那能不能在多线程环境执行呢答案肯定是yes但是也要注意多线程环境步骤执行一定要慎重。原因多线程环境下步骤是要设置不可重启。
Spring Batch 的多线程步骤是使用Spring 的 TaskExecutor(任务执行器)实现的。约定每一个块开启一个线程独立执行。 案例
需求分5个块处理user-thread.txt文件
1编写user-thread.txt文件
1#dafei#18
2#xiaofei#16
3#laofei#20
4#zhongfei#19
5#feifei#15
6#zhangsan#14
7#lisi#13
8#wangwu#12
9#zhaoliu#11
10#qianqi#10
2定义实体对象
Getter
Setter
ToString
public class User {private Long id;private String name;private int age;
}
3完整代码
package com.langfeiyes.batch._35_step_thread;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.item.ItemWriter;
import org.springframework.batch.item.file.FlatFileItemReader;
import org.springframework.batch.item.file.builder.FlatFileItemReaderBuilder;
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 ThreadStepJob {Autowiredprivate JobBuilderFactory jobBuilderFactory;Autowiredprivate StepBuilderFactory stepBuilderFactory;Beanpublic FlatFileItemReaderUser userItemReader(){System.out.println(Thread.currentThread());FlatFileItemReaderUser reader new FlatFileItemReaderBuilderUser().name(userItemReader).saveState(false) //防止状态被覆盖.resource(new ClassPathResource(user-thread.txt)).delimited().delimiter(#).names(id, name, age).targetType(User.class).build();return reader;}Beanpublic ItemWriterUser itemWriter(){return new ItemWriterUser() {Overridepublic void write(List? extends User items) throws Exception {items.forEach(System.err::println);}};}Beanpublic Step step(){return stepBuilderFactory.get(step1).User, Userchunk(2).reader(userItemReader()).writer(itemWriter()).taskExecutor(new SimpleAsyncTaskExecutor()).build();}Beanpublic Job job(){return jobBuilderFactory.get(thread-step-job).start(step()).build();}public static void main(String[] args) {SpringApplication.run(ThreadStepJob.class, args);}
}4结果
User(id2, namexiaofei, age16)
User(id5, namefeifei, age15)
User(id4, namezhongfei, age19)
User(id7, namelisi, age13)
User(id1, namedafei, age18)
User(id6, namezhangsan, age14)
User(id3, namelaofei, age20)
User(id8, namewangwu, age12)
User(id9, namezhaoliu, age11)
User(id10, nameqianqi, age10)
解析
1userItemReader() 加上saveState(false) Spring Batch 提供大部分的ItemReader是有状态的作业重启基本通过状态来确定作业停止位置而在多线程环境中如果对象维护状态被多个线程访问可能存在线程间状态相互覆盖问题。所以设置为false表示关闭状态但这也意味着作业不能重启了。
2step() 方法加上.taskExecutor(new SimpleAsyncTaskExecutor()) 为作业步骤添加了多线程处理能力以块为单位一个块一个线程观察上面的结果很明显能看出输出的顺序是乱序的。改变 job 的名字再执行会发现输出数据每次都不一样。
到这本篇就结束了欲知后事如何请听下回分解~
转视频版
看文字不过瘾可以切换视频版Spring Batch高效批处理框架实战
