Spring Batch 之 Sample（CSV文件操作）(四)

首先声明一点，该处的csv格式文件，必须新建txt 文件，然后修改后缀名，不能呢个由excel格式修改后缀名，因为该格式仍包含了其他很多信息，用txt打开的时候就看到了很多乱码似的字符！！！

本文将通过一个完整的实例，与大家一起讨论运用Spring Batch对CSV文件的读写操作。此实例的流程是：读取一个含有四个字段的CSV文件(ID,Name,Age,Score),对读取的字段做简单的处理，然后输出到另外一个CSV文件中。



JobLaunch类用来启动Job， CsvItemProcessor类用来对Reader取得的数据进行处理， Student类是一个POJO类，用来存放映射的数据。 inputFile.csv是数据读取文件， outputFile.csv是数据输出文件。

beans.xml文件配置如前篇文章，不再赘述。

batch.xml文件中Job配置如下：

<?xml version="1.0" encoding="UTF-8"?>
<bean:beans xmlns="http://www.springframework.org/schema/batch"
xmlns:bean="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:p="http://www.springframework.org/schema/p" xmlns:tx="http://www.springframework.org/schema/tx"
xmlns:aop="http://www.springframework.org/schema/aop" xmlns:context="http://www.springframework.org/schema/context"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.0.xsd http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-3.0.xsd http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop-3.0.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-2.5.xsd http://www.springframework.org/schema/batch http://www.springframework.org/schema/batch/spring-batch-2.1.xsd"> 
<bean:import resource="beans.xml"/>

<job id="csvJob">
<step id="csvStep">
<tasklet transaction-manager="transactionManager">
<chunk reader="csvItemReader" writer="csvItemWriter" processor="csvItemProcessor" commit-interval="1">
</chunk>
</tasklet>
</step>
</job>

<!-- 读取csv文件 -->
<bean:bean id="csvItemReader" class="org.springframework.batch.item.file.FlatFileItemReader" scope="step">
<bean:property name="resource" value="classpath:inputFile.csv"/>
<bean:property name="lineMapper">
<bean:bean
class="org.springframework.batch.item.file.mapping.DefaultLineMapper">
<bean:property name="lineTokenizer" ref="lineTokenizer"/>
<bean:property name="fieldSetMapper">
<bean:bean
class="org.springframework.batch.item.file.mapping.BeanWrapperFieldSetMapper">
<bean:property name="prototypeBeanName" value="student"></bean:property>
</bean:bean>
</bean:property>
</bean:bean>
</bean:property>
</bean:bean>

<bean:bean id="student" class="com.yss.sample.Student"></bean:bean>
<bean:bean id="csvItemProcessor" class="com.yss.sample.CsvItemProcessor" scope="step"></bean:bean>

<!-- lineTokenizer -->
<bean:bean id="lineTokenizer" class="org.springframework.batch.item.file.transform.DelimitedLineTokenizer">
<bean:property name="delimiter" value=","/>
<bean:property name="names">
<bean:list>
<bean:value>Id</bean:value>
<bean:value>name</bean:value>
<bean:value>age</bean:value>
<bean:value>score</bean:value>
</bean:list>
</bean:property>
</bean:bean>

<!-- 写CSV文件 -->
<bean:bean id="csvItemWriter" class="org.springframework.batch.item.file.FlatFileItemWriter" scope="step">
<bean:property name="resource" value="file:src/outputFile.csv"/>
<bean:property name="lineAggregator">
<bean:bean class="org.springframework.batch.item.file.transform.DelimitedLineAggregator">
<bean:property name="delimiter" value=","></bean:property>
<bean:property name="fieldExtractor">
<bean:bean
class="org.springframework.batch.item.file.transform.BeanWrapperFieldExtractor">
<bean:property name="names" value="name,age,score"></bean:property>
</bean:bean>
</bean:property>
</bean:bean>
</bean:property>
</bean:bean>

</bean:beans>

这个文件里配置了这次运行的JOB：csvJob。本Job包含一个Step，完成一个完整的CSV文件读写功能。分别由 csvItemReader完成CSV文件的读操作，由 csvItemProcessor完成对取得数据的处理，由
csvItemWriter完成对CSV文件的写操作。

csvItemReader实现的是Spring Batch提供FlatFileItemReader类，此类主要用于Flat文件的读操作。它包含两个必要的属性 resource和 lineMapper。前者指定要读取的文件的位置，后者是将文件的每一行映射成一个Pojo对象。其中 lineMapper也有两个重要属性 lineTokenizer和 fieldSetMapper， lineTokenizer将文件的一行分解成一个 FieldSet，然后由 fieldSetMapper映射成Pojo对象。

这种方式与DB的读操作非常类似。lineMapper类似于ResultSet，文件中的一行类似于Table中的一条记录，被封装成的FieldSet，类似于RowMapper。至于怎么将一条记录封装，这个工作由lineTokenizer的继承类DelimitedLineTokenizer完成。DelimitedLineTokenizer的delimiter属性决定文件的一行数据按照什么分解，默认的是“，”， names属性标示分解的每个字段的名字，传给fieldSetMapper（本实例用的是BeanWrapperFieldSetMapper）的时候，就可以按照这个名字取得相应的值。fieldSetMapper的属性prototypeBeanName，是映射Pojo类的名字。设置了此属性后，框架就会将lineTokenizer分解成的一个FieldSet映射成Pojo对象，映射是按照名字来完成的(lineTokenizer分解时标注的名字与Pojo对象中字段的名字对应)。

总之，FlatFileItemReader读取一条记录由以下四步完成：1，从resource指定的文件中读取一条记录；2，lineTokenizer将这条记录按照delimiter分解成Fileset，每个字段的名字由names属性取得；3，将分解成的Fileset传递给fieldSetMapper，由其按照名字映射成Pojo对象；4，最终由FlatFileItemReader将映射成的Pojo对象返回，框架将返回的对象传递给Processor。

csvItemProcessor实现的是ItemProcessor类。此类接受Reader映射成的Pojo对象，可以对此对象做相应的业务逻辑处理，然后返回，框架就会将返回的结果传递给Writer进行写操作。具体实现代码如下：

package com.yss.sample;

import org.springframework.batch.item.ItemProcessor;
import org.springframework.stereotype.Component;

@Component("csvItemProcessor")
public class CsvItemProcessor  implements ItemProcessor<Student,Student>{

/**
* 对取到的数据进行简单的处理。
*
* @param student
*            处理前的数据。
* @return 处理后的数据。
* @exception Exception
*                处理是发生的任何异常。
*/
@Override
public Student process(Student student) throws Exception {

System.out.println(student.getId());
System.out.println(student.getName());
System.out.println(student.getAge());
System.out.println(student.getScore());

/* 合并ID和名字 */
student.setName(student.getId() + "--" + student.getName());
/* 年龄加2 */
student.setAge(student.getAge() + 2);
/* 分数加10 */
student.setScore(student.getScore() + 10);
/* 将处理后的结果传递给writer */
return student;
}

}

csvItemWriter实现的是FlatFileItemWriter类。此类与FlatFileItemReader类相似，也有两个重要的属性：resource和lineAggregator。前者是要输出的文件的路径，后者和lineTokenizer类似。lineAggregator(本实例用DelimitedLineAggregator类)也有两个重要的属性：delimiter和fieldExtractor。Delimiter标示输出的字段以什么分割，后者将Pojo对象组装成由Pojo对象的字段组成的一个字符串。同样FlatFileItemWriter写一条记录也有以下四步完成：1，Processor传递过来一个对象给lineAggregator；2，lineAggregator将其这个对象转化成一个数组；3，再由lineAggregator的属性fieldExtractor将数组转化成按照delimiter分割一个字符串；4，将这个字符串输出。

这样，一条数据的读、处理、写操作就基本完成了。当然，读和写也可以自己写类来处理，只是要注意继承FlatFileItemReader和FlatFileItemWriter就可以了。

实例中用到的Student类代码如下：

package com.yss.sample;

public class Student {

private String id;

private String name;

private int age;

private float score;

public String getId() {
return id;
}

public void setId(String id) {
this.id = id;
}

public String getName() {
return name;
}

public void setName(String name) {
this.name = name;
}

public int getAge() {
return age;
}

public void setAge(int age) {
this.age = age;
}

public float getScore() {
return score;
}

public void setScore(float score) {
this.score = score;
}

}

实例中用到的输入数据如下：



实例输出结果如下：



本文的配置要注意以下两点：

1， 注意Writer的resource要写成“file:******”形式，不能用“classpath:******”形式。

2， 如果将Job配置中commit-interval属性配置为大于1时，每次commit的都是最后一条记录，前面读取的被覆盖了。具体原因不明，如果将Reader的fieldSetMapper属性自己重写，就可以解决这个问题。(注：student bean添加scope属性可以解决此问题：scope："prototype".2011/12/16)

下次，将和大家一起讨论关于XML文件的读写问题。