spring-batch day2 实战篇中

  本篇不探讨底层原理和一些代码含义，只介绍一些使用场景。配置接上篇。上篇介绍了一下Spring-batch的一些基本的操作单元，以及一些处理的API，本篇主要介绍一下spring-batch的ItemReader和ItemWriter

1.ItemReader读取程序中的数据源

• 首先自定义一个MyReader实现ItemReader接口
• 将MyReader绑定到Step，并由Step绑定一个Writer接口的实现用于输出
  具体代码如下：
  
  MyReader.java:

public class MyReader implements ItemReader<String> {private Iterator<String> iterator;public MyReader(List<String> data){this.iterator = data.iterator();}@Overridepublic String read() throws Exception, UnexpectedInputException, ParseException, NonTransientResourceException {if (iterator.hasNext()){return this.iterator.next();}return null;}
}

• 测试ItemReader
  ItemReaderDemo.java:

@Configuration
public class ItemReaderDemo {@Autowiredprivate JobBuilderFactory jobBuilderFactory;@Autowiredprivate StepBuilderFactory stepBuilderFactory;@Beanpublic Job itemReaderDemoJob(){return jobBuilderFactory.get("itemReaderDemoJob").start(step1()).build();}@Beanpublic Step step1() {return stepBuilderFactory.get("itemReaderDemoStep").<String,String>chunk(2).reader(itemReaderDemoRead()).writer(list -> {for (String s : list) {System.out.println(s);}}).build();}@Beanpublic MyReader itemReaderDemoRead() {List<String> data = Arrays.asList("java", "shiro", "springboot");return new MyReader(data);}
}

2.ItemReader读取数据库中的数据源

使用spring-batch提供的JdbcPagingItemReader读取数据

@Configuration
public class ItemReaderDemo {@Autowiredprivate JobBuilderFactory jobBuilderFactory;@Autowiredprivate StepBuilderFactory stepBuilderFactory;@Autowiredprivate DataSource dataSource;@Beanpublic Job itemReaderDemoJob(){return jobBuilderFactory.get("itemReaderDemoJob2").start(step1()).build();}@Beanpublic Step step1() {return stepBuilderFactory.get("itemReaderDemoStep").<User,User>chunk(2).reader(dbReader()).writer(list -> {for (User s : list) {System.out.println(s);}}).build();}@Beanpublic JdbcPagingItemReader<User> dbReader(){JdbcPagingItemReader<User> reader = new JdbcPagingItemReader<>();reader.setDataSource(dataSource);reader.setFetchSize(2);//把读取到的记录转化为User对象reader.setRowMapper(new RowMapper<User>() {@Overridepublic User mapRow(ResultSet resultSet, int i) throws SQLException {User user= new User();user.setUid(resultSet.getInt(1));user.setName(resultSet.getString(2));user.setAge(resultSet.getInt(3));user.setGender(resultSet.getString(4));return user;}});//指定sql语句MySqlPagingQueryProvider provider  = new MySqlPagingQueryProvider();provider.setSelectClause("uid,name,age,gender");provider.setFromClause("from user");//根据某个字段排序Map<String, Order> sort = new HashMap<>(1);sort.put("uid",Order.ASCENDING);provider.setSortKeys(sort);reader.setQueryProvider(provider);return reader;}
}

3.ItemReader读取普通文本文件中的数据源

使用FlatFileItemReader读取普通文本文件

@Configuration
public class ItemReaderDemo {@Autowiredprivate JobBuilderFactory jobBuilderFactory;@Autowiredprivate StepBuilderFactory stepBuilderFactory;@Autowiredprivate DataSource dataSource;@Beanpublic Job itemReaderDemoJob(){return jobBuilderFactory.get("itemReaderDemoJob3").start(step1()).build();}@Beanpublic Step step1() {return stepBuilderFactory.get("itemReaderDemoStep").<Customer,Customer>chunk(2).reader(dbReader()).writer(list -> {for (Customer s : list) {System.out.println(s);}}).build();}@Beanpublic FlatFileItemReader<Customer> dbReader(){FlatFileItemReader<Customer> reader = new FlatFileItemReader<>();reader.setResource(new ClassPathResource("Customer.txt"));//跳过第一行reader.setLinesToSkip(1);//解析数据DelimitedLineTokenizer tokenizer = new DelimitedLineTokenizer();tokenizer.setNames(new String[]{"cid","name","age","Birthday"});//将一行数据映射为对象DefaultLineMapper<Customer> mapper = new DefaultLineMapper<>();mapper.setLineTokenizer(tokenizer);mapper.setFieldSetMapper(new FieldSetMapper<Customer>() {@Overridepublic Customer mapFieldSet(FieldSet fieldSet) throws BindException {Customer customer = new Customer();customer.setCid(fieldSet.readInt("cid"));customer.setName(fieldSet.readString("name"));customer.setAge(fieldSet.readInt("age"));customer.setBirthday(fieldSet.readDate("Birthday"));return customer;}});mapper.afterPropertiesSet();reader.setLineMapper(mapper);return reader;}
}

文件结构如下图：

4.ItemReader读取xml文件中的数据源

• 需要导入处理xml文件的依赖包

@Configuration
public class ItemReaderDemo {@Autowiredprivate JobBuilderFactory jobBuilderFactory;@Autowiredprivate StepBuilderFactory stepBuilderFactory;@Autowiredprivate DataSource dataSource;@Beanpublic Job itemReaderDemoJob(){return jobBuilderFactory.get("itemReaderDemoJob4").start(step1()).build();}@Beanpublic Step step1() {return stepBuilderFactory.get("itemReaderDemoStep").<Customer,Customer>chunk(2).reader(dbReader()).writer(list -> {for (Customer s : list) {System.out.println(s);}}).build();}//指定作用域@Bean@StepScopepublic StaxEventItemReader<Customer> dbReader(){StaxEventItemReader<Customer> reader = new StaxEventItemReader<>();reader.setResource(new ClassPathResource("Customer.xml"));//指定需要处理的根标签reader.setFragmentRootElementName("Customer");XStreamMarshaller marshaller = new XStreamMarshaller();Map<String,Class> map = new HashMap<>();map.put("Customer",Customer.class);marshaller.setAliases(map);//默认的DateConverter不能满足我们的需求，需要自定义一个实现marshaller.setConverters(new Converter() {public boolean canConvert(Class arg0) {return Date.class == arg0;}public void marshal(Object arg0, HierarchicalStreamWriter arg1, MarshallingContext arg2) {}public Object unmarshal(HierarchicalStreamReader reader, UnmarshallingContext arg1) {GregorianCalendar calendar = new GregorianCalendar();SimpleDateFormat dateFm = new SimpleDateFormat("yyyy-MM-dd"); //格式化当前系统日期try {calendar.setTime(dateFm.parse(reader.getValue()));} catch (ParseException e) {throw new ConversionException(e.getMessage(), e);}return calendar.getTime();}});reader.setUnmarshaller(marshaller);return reader;}
}

5.多文件读取

使用MultiResourceItemReader读取文件

@Configuration
public class ItemReaderDemo {@Autowiredprivate JobBuilderFactory jobBuilderFactory;@Autowiredprivate StepBuilderFactory stepBuilderFactory;@Autowiredprivate DataSource dataSource;@Value("classpath:/file*.txt")private Resource[] fileResource;@Beanpublic Job itemReaderDemoJob(){return jobBuilderFactory.get("itemReaderDemoJob5").start(step1()).build();}@Beanpublic Step step1() {return stepBuilderFactory.get("itemReaderDemoStep").<Customer,Customer>chunk(2).reader(dbReader()).writer(list -> {for (Customer s : list) {System.out.println(s);}}).build();}//指定作用域@Bean@StepScopepublic MultiResourceItemReader<Customer> dbReader(){MultiResourceItemReader<Customer> reader = new MultiResourceItemReader();reader.setDelegate(flatReader());reader.setResources(fileResource);return reader;}@Bean@StepScopepublic FlatFileItemReader<Customer> flatReader(){FlatFileItemReader<Customer> reader = new FlatFileItemReader<>();reader.setResource(new ClassPathResource("Customer.txt"));//跳过第一行reader.setLinesToSkip(1);//解析数据DelimitedLineTokenizer tokenizer = new DelimitedLineTokenizer();tokenizer.setNames(new String[]{"cid","name","age","Birthday"});//将一行数据映射为对象DefaultLineMapper<Customer> mapper = new DefaultLineMapper<>();mapper.setLineTokenizer(tokenizer);mapper.setFieldSetMapper(new FieldSetMapper<Customer>() {@Overridepublic Customer mapFieldSet(FieldSet fieldSet) throws BindException {Customer customer = new Customer();customer.setCid(fieldSet.readInt("cid"));customer.setName(fieldSet.readString("name"));customer.setAge(fieldSet.readInt("age"));customer.setBirthday(fieldSet.readDate("Birthday"));return customer;}});mapper.afterPropertiesSet();reader.setLineMapper(mapper);return reader;}
}

6.带异常处理的Reader

public class RestartReader implements ItemStreamReader<String> {private Iterator<String> iterator;public RestartReader(List<String> data){this.iterator = data.iterator();}//ExecutionContext存储运行出错的行private ExecutionContext executionContext;private Long curLine;@Overridepublic String read() throws Exception, UnexpectedInputException, ParseException, NonTransientResourceException {if (iterator.hasNext()){return this.iterator.next();}return null;}@Overridepublic void open(ExecutionContext executionContext) throws ItemStreamException {this.executionContext = executionContext;if(executionContext.containsKey("curLine")){this.curLine = executionContext.getLong("curLine");}else{this.curLine = 0L;executionContext.put("curLine",this.curLine);System.out.println("start reading from curLine" + this.curLine+1);}}@Overridepublic void update(ExecutionContext executionContext) throws ItemStreamException {}@Overridepublic void close() throws ItemStreamException {}
}

7.使用ItemWriter将数据输出到数据库中

  自定义MyWriter实现ItemWriter接口

MyWriter.java:

@Component("myWriter")
public class MyWriter implements ItemWriter<String> {@Overridepublic void write(List<? extends String> list) throws Exception {System.out.println(list.size());for (String s : list) {System.out.println(s);}}
}

ItemWriterDemo.java:

@Configuration
public class ItemWriterDemo {@Autowiredprivate JobBuilderFactory jobBuilderFactory;@Autowiredprivate StepBuilderFactory stepBuilderFactory;@Autowired@Qualifier("myWriter")private ItemWriter<String> myWriter;@Beanpublic Job itemReaderDemoJob(){return jobBuilderFactory.get("itemWriterDemoJob3").start(step1()).build();}@Beanpublic Step step1() {return stepBuilderFactory.get("itemReaderDemoStep2").<String,String>chunk(2).reader(itemReaderDemoRead()).writer(myWriter).build();}@Beanpublic MyReader itemReaderDemoRead() {List<String> data = Arrays.asList("java", "shiro", "springboot");return new MyReader(data);}
}

8.将数据输出到数据库中

使用JdbcBatchItemWriter处理数据

@Configuration
public class ItemWriterDemo {@Autowiredprivate JobBuilderFactory jobBuilderFactory;@Autowiredprivate StepBuilderFactory stepBuilderFactory;@Autowiredprivate DataSource dataSource;@Beanpublic Job itemReaderDemoJob(){return jobBuilderFactory.get("itemWriterDemoJob4").start(step1()).build();}@Beanpublic Step step1() {return stepBuilderFactory.get("itemReaderDemoStep3").<Customer,Customer>chunk(2).reader(dbReader()).writer(itemWriter()).build();}//指定作用域@Bean@StepScopepublic StaxEventItemReader<Customer> dbReader(){StaxEventItemReader<Customer> reader = new StaxEventItemReader<>();reader.setResource(new ClassPathResource("Customer.xml"));//指定需要处理的根标签reader.setFragmentRootElementName("Customer");XStreamMarshaller marshaller = new XStreamMarshaller();Map<String,Class> map = new HashMap<>();map.put("Customer",Customer.class);marshaller.setAliases(map);//默认的DateConverter不能满足我们的需求，需要自定义一个实现marshaller.setConverters(new Converter() {public boolean canConvert(Class arg0) {return Date.class == arg0;}public void marshal(Object arg0, HierarchicalStreamWriter arg1, MarshallingContext arg2) {}public Object unmarshal(HierarchicalStreamReader reader, UnmarshallingContext arg1) {GregorianCalendar calendar = new GregorianCalendar();SimpleDateFormat dateFm = new SimpleDateFormat("yyyy-MM-dd"); //格式化当前系统日期try {calendar.setTime(dateFm.parse(reader.getValue()));} catch (ParseException e) {throw new ConversionException(e.getMessage(), e);}return calendar.getTime();}});reader.setUnmarshaller(marshaller);return reader;}@Beanpublic JdbcBatchItemWriter<Customer> itemWriter(){JdbcBatchItemWriter<Customer> writer = new JdbcBatchItemWriter<>();writer.setDataSource(dataSource);writer.setSql("insert into customer (cid,name,age,Birthday) values (:cid,:name,:age,:Birthday)");writer.setItemSqlParameterSourceProvider(new BeanPropertyItemSqlParameterSourceProvider<>());return writer;}
}

9.输出到文件：

使用FlatFileItemWriter输出：

@Configuration
public class ItemWriterDemo {@Autowiredprivate JobBuilderFactory jobBuilderFactory;@Autowiredprivate StepBuilderFactory stepBuilderFactory;@Autowiredprivate DataSource dataSource;@Beanpublic Job itemReaderDemoJob(){return jobBuilderFactory.get("itemWriterDemoJob10").start(step1()).build();}@Beanpublic Step step1() {return stepBuilderFactory.get("itemReaderDemoStep3").<Customer,Customer>chunk(2).reader(dbReader()).writer(itemWriter()).build();}//指定作用域@Bean@StepScopepublic StaxEventItemReader<Customer> dbReader(){StaxEventItemReader<Customer> reader = new StaxEventItemReader<>();reader.setResource(new ClassPathResource("Customer.xml"));//指定需要处理的根标签reader.setFragmentRootElementName("Customer");XStreamMarshaller marshaller = new XStreamMarshaller();Map<String,Class> map = new HashMap<>();map.put("Customer",Customer.class);marshaller.setAliases(map);//默认的DateConverter不能满足我们的需求，需要自定义一个实现marshaller.setConverters(new Converter() {public boolean canConvert(Class arg0) {return Date.class == arg0;}public void marshal(Object arg0, HierarchicalStreamWriter arg1, MarshallingContext arg2) {}public Object unmarshal(HierarchicalStreamReader reader, UnmarshallingContext arg1) {GregorianCalendar calendar = new GregorianCalendar();SimpleDateFormat dateFm = new SimpleDateFormat("yyyy-MM-dd"); //格式化当前系统日期try {calendar.setTime(dateFm.parse(reader.getValue()));} catch (ParseException e) {throw new ConversionException(e.getMessage(), e);}return calendar.getTime();}});reader.setUnmarshaller(marshaller);return reader;}@Beanpublic FlatFileItemWriter<Customer> itemWriter(){FlatFileItemWriter<Customer> writer = new FlatFileItemWriter<>();//不能写到类路径下writer.setResource(new PathResource("D:\\test.txt"));writer.setLineAggregator(new LineAggregator<Customer>() {ObjectMapper mapper = new ObjectMapper();@Overridepublic String aggregate(Customer item) {String str = null;try {str = mapper.writeValueAsString(item);} catch (JsonProcessingException e) {e.printStackTrace();}System.out.println(str);return str;}});return writer;}
}

10.输出到XML文件

使用StaxEventItemWriter实现文件写入：

@Configuration
public class ItemWriterDemo {@Autowiredprivate JobBuilderFactory jobBuilderFactory;@Autowiredprivate StepBuilderFactory stepBuilderFactory;@Autowiredprivate DataSource dataSource;@Beanpublic Job itemReaderDemoJob(){return jobBuilderFactory.get("itemWriterDemoJob11").start(step1()).build();}@Beanpublic Step step1() {return stepBuilderFactory.get("itemReaderDemoStep3").<Customer,Customer>chunk(2).reader(dbReader()).writer(itemWriter()).build();}//指定作用域@Bean@StepScopepublic StaxEventItemReader<Customer> dbReader(){StaxEventItemReader<Customer> reader = new StaxEventItemReader<>();reader.setResource(new ClassPathResource("Customer.xml"));//指定需要处理的根标签reader.setFragmentRootElementName("Customer");XStreamMarshaller marshaller = new XStreamMarshaller();Map<String,Class> map = new HashMap<>();map.put("Customer",Customer.class);marshaller.setAliases(map);//默认的DateConverter不能满足我们的需求，需要自定义一个实现marshaller.setConverters(new Converter() {public boolean canConvert(Class arg0) {return Date.class == arg0;}public void marshal(Object arg0, HierarchicalStreamWriter arg1, MarshallingContext arg2) {}public Object unmarshal(HierarchicalStreamReader reader, UnmarshallingContext arg1) {GregorianCalendar calendar = new GregorianCalendar();SimpleDateFormat dateFm = new SimpleDateFormat("yyyy-MM-dd"); //格式化当前系统日期try {calendar.setTime(dateFm.parse(reader.getValue()));} catch (ParseException e) {throw new ConversionException(e.getMessage(), e);}return calendar.getTime();}});reader.setUnmarshaller(marshaller);return reader;}@Beanpublic StaxEventItemWriter<Customer> itemWriter() {StaxEventItemWriter<Customer> writer = new StaxEventItemWriter<>();//不能写到类路径下writer.setResource(new PathResource("D:\\text.xml"));XStreamMarshaller marshaller = new XStreamMarshaller();Map<String,Class> map = new HashMap<>();map.put("customer",Customer.class);marshaller.setAliases(map);writer.setRootTagName("customers");writer.setMarshaller(marshaller);try {writer.afterPropertiesSet();} catch (Exception e) {e.printStackTrace();}return writer;}
}

11.输出到多个文件：

使用CompositeItemWriter实现

@Configuration
public class ItemWriterDemo {@Autowiredprivate JobBuilderFactory jobBuilderFactory;@Autowiredprivate StepBuilderFactory stepBuilderFactory;@Autowiredprivate DataSource dataSource;@Beanpublic Job itemReaderDemoJob(){return jobBuilderFactory.get("itemWriterDemoJob12").start(step1()).build();}@Beanpublic Step step1() {return stepBuilderFactory.get("itemReaderDemoStep3").<Customer,Customer>chunk(2).reader(dbReader()).writer(itemWriter()).build();}//指定作用域@Bean@StepScopepublic StaxEventItemReader<Customer> dbReader(){StaxEventItemReader<Customer> reader = new StaxEventItemReader<>();reader.setResource(new ClassPathResource("Customer.xml"));//指定需要处理的根标签reader.setFragmentRootElementName("Customer");XStreamMarshaller marshaller = new XStreamMarshaller();Map<String,Class> map = new HashMap<>();map.put("Customer",Customer.class);marshaller.setAliases(map);//默认的DateConverter不能满足我们的需求，需要自定义一个实现marshaller.setConverters(new Converter() {public boolean canConvert(Class arg0) {return Date.class == arg0;}public void marshal(Object arg0, HierarchicalStreamWriter arg1, MarshallingContext arg2) {}public Object unmarshal(HierarchicalStreamReader reader, UnmarshallingContext arg1) {GregorianCalendar calendar = new GregorianCalendar();SimpleDateFormat dateFm = new SimpleDateFormat("yyyy-MM-dd"); //格式化当前系统日期try {calendar.setTime(dateFm.parse(reader.getValue()));} catch (ParseException e) {throw new ConversionException(e.getMessage(), e);}return calendar.getTime();}});reader.setUnmarshaller(marshaller);return reader;}@Beanpublic CompositeItemWriter<Customer> itemWriter() {CompositeItemWriter<Customer> writer = new CompositeItemWriter<>();writer.setDelegates(Arrays.asList(jsonFileWriter(),xmlFileWriter()));return writer;}@Beanpublic FlatFileItemWriter<Customer> jsonFileWriter(){FlatFileItemWriter<Customer> writer = new FlatFileItemWriter<>();//不能写到类路径下writer.setResource(new PathResource("D:\\test.txt"));writer.setLineAggregator(new LineAggregator<Customer>() {ObjectMapper mapper = new ObjectMapper();@Overridepublic String aggregate(Customer item) {String str = null;try {str = mapper.writeValueAsString(item);} catch (JsonProcessingException e) {e.printStackTrace();}System.out.println(str);return str;}});return writer;}@Beanpublic StaxEventItemWriter<Customer> xmlFileWriter() {StaxEventItemWriter<Customer> writer = new StaxEventItemWriter<>();//不能写到类路径下writer.setResource(new PathResource("D:\\text.xml"));XStreamMarshaller marshaller = new XStreamMarshaller();Map<String,Class> map = new HashMap<>();map.put("customer",Customer.class);marshaller.setAliases(map);writer.setRootTagName("customers");writer.setMarshaller(marshaller);try {writer.afterPropertiesSet();} catch (Exception e) {e.printStackTrace();}return writer;}
}

12.文件分类输出

将数据根据不同的类型输出到不同的文件

@Configuration
public class ItemWriterDemo {@Autowiredprivate JobBuilderFactory jobBuilderFactory;@Autowiredprivate StepBuilderFactory stepBuilderFactory;@Autowiredprivate DataSource dataSource;@Beanpublic Job itemReaderDemoJob(){return jobBuilderFactory.get("itemWriterDemoJob26").start(step1()).build();}@Beanpublic Step step1() {return stepBuilderFactory.get("itemReaderDemoStep3").<Customer,Customer>chunk(2).reader(dbReader()).writer(itemWriter())//需要把流添加上去，不然会报must open before write to 错误.stream(jsonFileWriter).stream(xmlFileWriter).build();}//指定作用域@Bean@StepScopepublic StaxEventItemReader<Customer> dbReader(){StaxEventItemReader<Customer> reader = new StaxEventItemReader<>();reader.setResource(new ClassPathResource("Customer.xml"));//指定需要处理的根标签reader.setFragmentRootElementName("Customer");XStreamMarshaller marshaller = new XStreamMarshaller();Map<String,Class> map = new HashMap<>();map.put("Customer",Customer.class);marshaller.setAliases(map);//默认的DateConverter不能满足我们的需求，需要自定义一个实现marshaller.setConverters(new Converter() {public boolean canConvert(Class arg0) {return Date.class == arg0;}public void marshal(Object arg0, HierarchicalStreamWriter arg1, MarshallingContext arg2) {}public Object unmarshal(HierarchicalStreamReader reader, UnmarshallingContext arg1) {GregorianCalendar calendar = new GregorianCalendar();SimpleDateFormat dateFm = new SimpleDateFormat("yyyy-MM-dd"); //格式化当前系统日期try {calendar.setTime(dateFm.parse(reader.getValue()));} catch (ParseException e) {throw new ConversionException(e.getMessage(), e);}return calendar.getTime();}});reader.setUnmarshaller(marshaller);return reader;}@Beanpublic ClassifierCompositeItemWriter<Customer> itemWriter() {ClassifierCompositeItemWriter<Customer> writer = new ClassifierCompositeItemWriter<>();writer.setClassifier(new Classifier<Customer, ItemWriter<? super Customer>>() {//org.springframework.batch.item.WriterNotOpenException: Writer must be open before it can be written to@Overridepublic ItemWriter<? super Customer> classify(Customer customer) {if(customer.getCid()%2==0){return jsonFileWriter();}return xmlFileWriter();}});return writer;}@Autowired@Qualifier("jsonFileWriter")private ItemStreamWriter<Customer> jsonFileWriter;@Beanpublic FlatFileItemWriter<Customer> jsonFileWriter(){FlatFileItemWriter<Customer> writer = new FlatFileItemWriter<>();//不能写到类路径下writer.setResource(new PathResource("D:\\test.txt"));writer.setLineAggregator(new LineAggregator<Customer>() {ObjectMapper mapper = new ObjectMapper();@Overridepublic String aggregate(Customer item) {String str = null;try {str = mapper.writeValueAsString(item);} catch (JsonProcessingException e) {e.printStackTrace();}System.out.println(str);return str;}});return writer;}@Autowired@Qualifier("xmlFileWriter")private ItemStreamWriter<Customer> xmlFileWriter;@Beanpublic StaxEventItemWriter<Customer> xmlFileWriter() {StaxEventItemWriter<Customer> writer = new StaxEventItemWriter<>();//不能写到类路径下writer.setResource(new PathResource("D:\\text.xml"));XStreamMarshaller marshaller = new XStreamMarshaller();Map<String,Class> map = new HashMap<>();map.put("customer",Customer.class);marshaller.setAliases(map);writer.setRootTagName("customers");writer.setMarshaller(marshaller);try {writer.afterPropertiesSet();} catch (Exception e) {e.printStackTrace();}return writer;}
}