<bean id="fieldSetMapper" class="org.springframework.batch.item.file.mapping.BeanWrapperFieldSetMapper"> <property name="prototypeBeanName" value="player" /> </bean> <bean id="player" class="org.springframework.batch.sample.domain.Player" scope="prototype" />

对于 FieldSet 中的每个条目(entry), mapper都会在Player对象的新实例中查找相应的setter (因此,需要指定 prototype scope),和 Spring容器 查找 setter匹配属性名是一样的方式。FieldSet 中每个可用的字段都会被映射, 然后返回组装好的 Player 对象,不需要再手写代码。

UK21341EAH4121131.11customer1 UK21341EAH4221232.11customer2 UK21341EAH4321333.11customer3 UK21341EAH4421434.11customer4 UK21341EAH4521535.11customer5

虽然看起来像是一个很长的字段,但实际上代表了4个分开的字段:

1. ISIN : 唯一标识符,订购的商品编码 – 占12字符。
2. Quantity : 订购的商品数量 – 占3字符。
3. Price : 商品的价格 – 占5字符。
4. Customer : 订购商品的顾客Id – 占9字符。

<bean id="fixedLengthLineTokenizer" class="org.springframework.batch.io.file.transform.FixedLengthTokenizer"> <property name="names" value="ISIN,Quantity,Price,Customer" /> <property name="columns" value="1-12, 13-15, 16-20, 21-29" /> </bean>

因为 FixedLengthLineTokenizer 使用的也是 LineTokenizer 接口, 所以返回值同样是 FieldSet, 和使用分隔符基本上是一样的。这也就可以使用同样的方式来处理其输出, 例如使用 BeanWrapperFieldSetMapper。

USER;Smith;Peter;;T;20014539;F LINEA;1044391041ABC037.49G201XX1383.12H LINEB;2134776319DEF422.99M005LI

这个文件中有三种类型的记录, “USER”, “LINEA”, 以及 “LINEB”。 一行 “USER” 对应一个 User 对象。 “LINEA” 和 “LINEB”对应的都是 Line 对象, 只是 “LINEA” 包含的信息比“LINEB”要多。

<bean id="orderFileLineMapper" class="org.spr...PatternMatchingCompositeLineMapper"> <property name="tokenizers"> <map> <entry key="USER*" value-ref="userTokenizer" /> <entry key="LINEA*" value-ref="lineATokenizer" /> <entry key="LINEB*" value-ref="lineBTokenizer" /> </map> </property> <property name="fieldSetMappers"> <map> <entry key="USER*" value-ref="userFieldSetMapper" /> <entry key="LINE*" value-ref="lineFieldSetMapper" /> </map> </property> </bean>

在这个示例中, “LINEA” 和 “LINEB” 使用独立的 LineTokenizer，但使用同一个 FieldSetMapper.

<entry key="*" value-ref="defaultLineTokenizer" />

还有一个 PatternMatchingCompositeLineTokenizer 可用来单独解析。

IncorrectTokenCountException

tokenizer.setNames(new String[] {"A", "B", "C", "D"}); try { tokenizer.tokenize("a,b,c"); }catch(IncorrectTokenCountException e){ assertEquals(4, e.getExpectedCount()); assertEquals(3, e.getActualCount()); }

因为 tokenizer 配置了4列的名称,但在这个文件中只找到 3 个字段, 所以会抛出 IncorrectTokenCountException 异常。

tokenizer.setColumns(new Range[] { new Range(1, 5), new Range(6, 10), new Range(11, 15) }); try { tokenizer.tokenize("12345"); fail("Expected IncorrectLineLengthException"); } catch (IncorrectLineLengthException ex) { assertEquals(15, ex.getExpectedLength()); assertEquals(5, ex.getActualLength()); }

上面配置的范围是: 1-5 , 6-10 , 以及 11-15 , 因此预期的总长度是15。但在这里传入的行的长度是 5 ,所以会导致IncorrectLineLengthException 异常。之所以直接抛出异常, 而不是先去映射第一个字段的原因是为了更早发现处理失败, 而不再调用 FieldSetMapper 来读取第2列。但是呢，有些情况下, 行的长度并不总是固定的。 出于这个原因, 可以通过设置’strict’ 属性的值，不验证行的宽度:

tokenizer.setColumns(new Range[] { new Range(1, 5), new Range(6, 10) }); tokenizer.setStrict(false); FieldSet tokens = tokenizer.tokenize("12345"); assertEquals("12345", tokens.readString(0)); assertEquals("", tokens.readString(1));

上面示例和前一个几乎完全相同, 只是调用了 tokenizer.setStrict(false) 。这个设置告诉 tokenizer 在对一行进行解析(tokenizing)时不要去管(enforce)行的长度。然后就正确地创建了一个 FieldSet并返回。当然,剩下的值就只会包含空的token值。

public interface LineAggregator<T> { public String aggregate(T item); }

接口 LineAggregator 与 LineTokenizer 相互对应. LineTokenizer 接收 String ,处理后返回一个 FieldSet 对象, 而LineAggregator 则是接收一条记录,返回对应的 String.

public class PassThroughLineAggregator<T> implements LineAggregator<T> { public String aggregate(T item) { return item.toString(); } }

上面的实现对于需要直接转换为string的时候是很管用的,但是 FlatFileItemWriter 的一些优势也是很有必要的,比如 事务,以及支持重启特性等.

简单的文件写入示例

1. 将要写出的对象传递给 LineAggregator 以获取一个字符串(String).
2.  将返回的 String 写入配置指定的文件中.

下面是 FlatFileItemWriter 中对应的代码:

public void write(T item) throws Exception { write(lineAggregator.aggregate(item) + LINE_SEPARATOR); }

简单的配置如下所示:

<bean id="itemWriter" class="org.spr...FlatFileItemWriter"> <property name="resource" value="file:target/test-outputs/output.txt" /> <property name="lineAggregator"> <bean class="org.spr...PassThroughLineAggregator"/> </property> </bean>

属性提取器 FieldExtractor

1. 从文件中读取一行.
2. 将这一行字符串传递给 LineTokenizer#tokenize() 方法, 以获取 FieldSet 对象
3. 将分词器返回的 FieldSet 传给一个 FieldSetMapper 映射器, 然后将 ItemReader#read() 方法得到的结果 return。

文件的写入也很类似, 但步骤正好相反:

1. 将要写入的对象传递给 writer
2. 将领域对象的属性域转换为数组
3. 将结果数组合并(aggregate)为一行字符串

public interface FieldExtractor<T> { Object[] extract(T item); }

FieldExtractor 的实现类应该根据传入对象的属性创建一个数组， 稍后使用分隔符将各个元素写入文件，或者作为 field-width line 的一部分.

BeanWrapperFieldExtractor<Name> extractor = new BeanWrapperFieldExtractor<Name>(); extractor.setNames(new String[] { "first", "last", "born" }); String first = "Alan"; String last = "Turing"; int born = 1912; Name n = new Name(first, last, born); Object[] values = extractor.extract(n); assertEquals(first, values[0]); assertEquals(last, values[1]); assertEquals(born, values[2]);

这个 extractor 实现只有一个必需的属性,就是 names , 里面用来存放要映射字段的名字。 就像 BeanWrapperFieldSetMapper 需要字段名称来将 FieldSet 中的 field 映射到对象的 setter 方法一样, BeanWrapperFieldExtractor 需要 names 映射 getter 方法来创建一个对象数组。值得注意的是, names的顺序决定了field在数组中的顺序。

public class CustomerCredit { private int id; private String name; private BigDecimal credit; //getters and setters removed for clarity }

因为使用到了领域对象,所以必须提供 FieldExtractor 接口的实现，当然也少不了要使用的分隔符:

<bean id="itemWriter" class="org.springframework.batch.item.file.FlatFileItemWriter"> <property name="resource" ref="outputResource" /> <property name="lineAggregator"> <bean class="org.spr...DelimitedLineAggregator"> <property name="delimiter" value=","/> <property name="fieldExtractor"> <bean class="org.spr...BeanWrapperFieldExtractor"> <property name="names" value="name,credit"/> </bean> </property> </bean> </property> </bean>

在这种情况下, 本章前面提到过的 BeanWrapperFieldExtractor 被用来将 CustomerCredit 中的 name 和 credit 字段转换为一个对象数组, 然后在各个字段之间用逗号分隔写入文件。

<bean id="itemWriter" class="org.springframework.batch.item.file.FlatFileItemWriter"> <property name="resource" ref="outputResource" /> <property name="lineAggregator"> <bean class="org.spr...FormatterLineAggregator"> <property name="fieldExtractor"> <bean class="org.spr...BeanWrapperFieldExtractor"> <property name="names" value="name,credit" /> </bean> </property> <property name="format" value="%-9s%-2.0f" /> </bean> </property> </bean>

上面的示例大部分看起来是一样的, 只有 format 属性的值不同:

<property name="format" value="%-9s%-2.0f" />

底层实现采用 Java 5 提供的 Formatter 。Java的 Formatter (格式化) 基于C语言的 printf 函数功能。关于如何配置formatter 请参考 Formatter 的javadoc.

1.7 XML Item Readers and Writers

<?xml version="1.0" encoding="UTF-8"?> <records> <trade xmlns="http://springframework.org/batch/sample/io/oxm/domain"> <isin>XYZ0001</isin> <quantity>5</quantity> <price>11.39</price> <customer>Customer1</customer> </trade> <trade xmlns="http://springframework.org/batch/sample/io/oxm/domain"> <isin>XYZ0002</isin> <quantity>2</quantity> <price>72.99</price> <customer>Customer2c</customer> </trade> <trade xmlns="http://springframework.org/batch/sample/io/oxm/domain"> <isin>XYZ0003</isin> <quantity>9</quantity> <price>99.99</price> <customer>Customer3</customer> </trade> </records>

• Root Element Name 片段根元素的名称就是要映射的对象。上面的示例代表的是 trade 的值。
• Resource Spring Resource 代表了需要读取的文件。
• Unmarshaller Spring OXM提供的Unmarshalling 用于将 XML片段映射为对象.

<bean id="itemReader" class="org.springframework.batch.item.xml.StaxEventItemReader"> <property name="fragmentRootElementName" value="trade" /> <property name="resource" value="data/iosample/input/input.xml" /> <property name="unmarshaller" ref="tradeMarshaller" /> </bean>

请注意,在上面的例子中,我们选用一个 XStreamMarshaller, 里面接受一个id为 aliases 的 map, 将首个entry的 key 值作为文档片段的name(即根元素), 将 value 作为绑定的对象类型。类似于FieldSet, 后面的其他元素映射为对象内部的字段名/值对。在配置文件中,我们可以像下面这样使用Spring配置工具来描述所需的alias:

<bean id="tradeMarshaller" class="org.springframework.oxm.xstream.XStreamMarshaller"> <property name="aliases"> <util:map id="aliases"> <entry key="trade" value="org.springframework.batch.sample.domain.Trade" /> <entry key="price" value="java.math.BigDecimal" /> <entry key="name" value="java.lang.String" /> </util:map> </property> </bean>

当 reader 读取到XML资源的一个新片段时(匹配默认的标签名称)。reader 根据这个片段构建一个独立的XML(或至少看起来是这样),并将 document 传给反序列化器(通常是一个Spring OXM Unmarshaller 的包装类)将XML映射为一个Java对象。

StaxEventItemReader xmlStaxEventItemReader = new StaxEventItemReader() Resource resource = new ByteArrayResource(xmlResource.getBytes()) Map aliases = new HashMap(); aliases.put("trade","org.springframework.batch.sample.domain.Trade"); aliases.put("price","java.math.BigDecimal"); aliases.put("customer","java.lang.String"); Marshaller marshaller = new XStreamMarshaller(); marshaller.setAliases(aliases); xmlStaxEventItemReader.setUnmarshaller(marshaller); xmlStaxEventItemReader.setResource(resource); xmlStaxEventItemReader.setFragmentRootElementName("trade"); xmlStaxEventItemReader.open(new ExecutionContext()); boolean hasNext = true CustomerCredit credit = null; while (hasNext) { credit = xmlStaxEventItemReader.read(); if (credit == null) { hasNext = false; } else { System.out.println(credit); } }

1.7.2 StaxEventItemWriter

<bean id="itemWriter" class="org.springframework.batch.item.xml.StaxEventItemWriter"> <property name="resource" ref="outputResource" /> <property name="marshaller" ref="customerCreditMarshaller" /> <property name="rootTagName" value="customers" /> <property name="overwriteOutput" value="true" /> </bean>

上面配置了3个必需的属性,以及1个可选属性 overwriteOutput = true , (本章前面提到过) 用来指定一个已存在的文件是否可以被覆盖。应该注意的是, writer 使用的 marshaller 和前面讲的 reading 示例中是完全相同的:

<bean id="customerCreditMarshaller" class="org.springframework.oxm.xstream.XStreamMarshaller"> <property name="aliases"> <util:map id="aliases"> <entry key="customer" value="org.springframework.batch.sample.domain.CustomerCredit" /> <entry key="credit" value="java.math.BigDecimal" /> <entry key="name" value="java.lang.String" /> </util:map> </property> </bean>

StaxEventItemWriter staxItemWriter = new StaxEventItemWriter() FileSystemResource resource = new FileSystemResource("data/outputFile.xml") Map aliases = new HashMap(); aliases.put("customer","org.springframework.batch.sample.domain.CustomerCredit"); aliases.put("credit","java.math.BigDecimal"); aliases.put("name","java.lang.String"); Marshaller marshaller = new XStreamMarshaller(); marshaller.setAliases(aliases); staxItemWriter.setResource(resource); staxItemWriter.setMarshaller(marshaller); staxItemWriter.setRootTagName("trades"); staxItemWriter.setOverwriteOutput(true); ExecutionContext executionContext = new ExecutionContext(); staxItemWriter.open(executionContext); CustomerCredit Credit = new CustomerCredit(); trade.setPrice(11.39); credit.setName("Customer1"); staxItemWriter.write(trade);

1.8 多个数据输入文件

file-1.txt file-2.txt ignored.txt

<bean id="multiResourceReader" class="org.spr...MultiResourceItemReader"> <property name="resources" value="classpath:data/input/file-*.txt" /> <property name="delegate" ref="flatFileItemReader" /> </bean>

1.9 数据库(Database)

1.9.1 基于Cursor的ItemReaders

JdbcCursorItemReader

CREATE TABLE CUSTOMER ( ID BIGINT IDENTITY PRIMARY KEY, NAME VARCHAR(45), CREDIT FLOAT );

public class CustomerCreditRowMapper implements RowMapper { public static final String ID_COLUMN = "id"; public static final String NAME_COLUMN = "name"; public static final String CREDIT_COLUMN = "credit"; public Object mapRow(ResultSet rs, int rowNum) throws SQLException { CustomerCredit customerCredit = new CustomerCredit(); customerCredit.setId(rs.getInt(ID_COLUMN)); customerCredit.setName(rs.getString(NAME_COLUMN)); customerCredit.setCredit(rs.getBigDecimal(CREDIT_COLUMN)); return customerCredit; } }

//For simplicity sake, assume a dataSource has already been obtained JdbcTemplate jdbcTemplate = new JdbcTemplate(dataSource); List customerCredits = jdbcTemplate.query("SELECT ID, NAME, CREDIT from CUSTOMER", new CustomerCreditRowMapper());

JdbcCursorItemReader itemReader = new JdbcCursorItemReader(); itemReader.setDataSource(dataSource); itemReader.setSql("SELECT ID, NAME, CREDIT from CUSTOMER"); itemReader.setRowMapper(new CustomerCreditRowMapper()); int counter = 0; ExecutionContext executionContext = new ExecutionContext(); itemReader.open(executionContext); Object customerCredit = new Object(); while(customerCredit != null){ customerCredit = itemReader.read(); counter++; } itemReader.close(executionContext);

运行这段代码后 counter 的值将变成 1000。如果上面的代码将返回的 customerCredit 放入 List, 则结果将和使用JdbcTemplate 的例子完全一致。 但是呢, 使用 ItemReader 的强大优势在于, 它允许数据项变成 “流式(streamed)”。 调用一次 read 方法, 通过ItemWriter写出数据对象, 然后再通过 read 获取下一项。 这使得 item 读取和写出可以进行 “分块(chunks)”, 并且周期性地提交, 这才是高性能批处理的本质。此外,它可以很容易地通过配置注入到某个 Spring Batch Step中:

<bean id="itemReader" class="org.spr...JdbcCursorItemReader"> <property name="dataSource" ref="dataSource"/> <property name="sql" value="select ID, NAME, CREDIT from CUSTOMER"/> <property name="rowMapper"> <bean class="org.springframework.batch.sample.domain.CustomerCreditRowMapper"/> </property> </bean>

因为在Java中有很多种不同的方式来打开游标, 所以 JdbcCustorItemReader 有许多可以设置的属性 :

HibernateCursorItemReader

使用 Spring 的程序员需要作出一个重要的决策，即是否使用ORM解决方案,这决定了是否使用 JdbcTemplate 或
HibernateTemplate , Spring Batch开发者也面临同样的选择。HibernateCursorItemReader 是 Hibernate 的游标实现。 其实在批处理中使用 Hibernate 那是相当有争议。这很大程度上是因为 Hibernate 最初就是设计了用来开发在线程序的。

但也不是说Hibernate就不能用来进行批处理。最简单的解决办法就是使用一个 StatelessSession (无状态会话), 而不使用标准 session 。这样就去掉了在批处理场景中 Hibernate 那些恼人的缓存、脏检查等等。

更多无状态会话与正常hibernate会话之间的差异, 请参考你使用的 hibernate 版本对应的文档。HibernateCursorItemReader 允许您声明一个HQL语句, 并传入 SessionFactory , 然后每次调用 read 时就会返回一个对象， 和 JdbcCursorItemReader 一样。下面的示例配置也使用和 JDBC reader 相同的数据库表:

HibernateCursorItemReader itemReader = new HibernateCursorItemReader(); itemReader.setQueryString("from CustomerCredit"); //For simplicity sake, assume sessionFactory already obtained. itemReader.setSessionFactory(sessionFactory); itemReader.setUseStatelessSession(true); int counter = 0; ExecutionContext executionContext = new ExecutionContext(); itemReader.open(executionContext); Object customerCredit = new Object(); while(customerCredit != null){ customerCredit = itemReader.read(); counter++; } itemReader.close(executionContext);

这里配置的 ItemReader 将以完全相同的方式返回CustomerCredit对象，和 JdbcCursorItemReader 没有区别, 如果
hibernate 映射文件正确的话。 useStatelessSession 属性的默认值为 true , 这里明确设置的目的只是为了引起你的注意,我们可以通过他来进行切换。 还值得注意的是 可以通过 setFetchSize 设置底层 cursor 的 fetchSize 属性 。与JdbcCursorItemReader一样,配置很简单:

<bean id="itemReader" class="org.springframework.batch.item.database.HibernateCursorItemReader"> <property name="sessionFactory" ref="sessionFactory" /> <property name="queryString" value="from CustomerCredit" /> </bean>

StoredProcedureItemReader

有时候使用存储过程来获取游标数据是很有必要的。 StoredProcedureItemReader 和 JdbcCursorItemReader 其实差不多，只是不再执行一个查询来获取游标，而是执行一个存储过程, 由存储过程返回一个游标。 存储过程有三种返回游标的方式:

1. 作为一个 ResultSet 返回(SQL Server, Sybase, DB2, Derby 以及 MySQL支持)
2. 作为一个 out 参数返回 ref-cursor (Oracle和PostgreSQL使用这种方式)
3. 作为存储函数(stored function)的返回值

下面是一个基本的配置示例, 还是使用上面 “客户信用” 的例子:

<bean id="reader" class="o.s.batch.item.database.StoredProcedureItemReader"> <property name="dataSource" ref="dataSource"/> <property name="procedureName" value="sp_customer_credit"/> <property name="rowMapper"> <bean class="org.springframework.batch.sample.domain.CustomerCreditRowMapper"/> </property> </bean>

这个例子依赖于存储过程提供一个 ResultSet 作为返回结果(方式1)。

如果存储过程返回一个ref-cursor(方式2),那么我们就需要提供返回的ref-cursor(out 参数)的位置。下面的示例中,第一个参数是返回的ref-cursor:

<bean id="reader" class="o.s.batch.item.database.StoredProcedureItemReader"> <property name="dataSource" ref="dataSource"/> <property name="procedureName" value="sp_customer_credit"/> <property name="refCursorPosition" value="1"/> <property name="rowMapper"> <bean class="org.springframework.batch.sample.domain.CustomerCreditRowMapper"/> </property> </bean>

如果存储函数的返回值是一个游标(方式 3), 则需要将 function 属性设置为 true , 默认为 false 。如下面所示:

<bean id="reader" class="o.s.batch.item.database.StoredProcedureItemReader"> <property name="dataSource" ref="dataSource"/> <property name="procedureName" value="sp_customer_credit"/> <property name="function" value="true"/> <property name="rowMapper"> <bean class="org.springframework.batch.sample.domain.CustomerCreditRowMapper"/> </property> </bean>

在所有情况下,我们都需要定义 RowMapper 以及 DataSource, 还有存储过程的名字。

如果存储过程/函数需要传入参数, 那么必须声明并通过 parameters 属性来设置值。下面是一个关于 Oracle 的示例, 其中声明了三个参数。 第一个是 out 参数,用来返回 ref-cursor, 第二第三个参数是 in 型参数, 类型都是 INTEGER :

<bean id="reader" class="o.s.batch.item.database.StoredProcedureItemReader"> <property name="dataSource" ref="dataSource"/> <property name="procedureName" value="spring.cursor_func"/> <property name="parameters"> <list> <bean class="org.springframework.jdbc.core.SqlOutParameter"> <constructor-arg index="0" value="newid"/> <constructor-arg index="1"> <util:constant static-field="oracle.jdbc.OracleTypes.CURSOR"/> </constructor-arg> </bean> <bean class="org.springframework.jdbc.core.SqlParameter"> <constructor-arg index="0" value="amount"/> <constructor-arg index="1"> <util:constant static-field="java.sql.Types.INTEGER"/> </constructor-arg> </bean> <bean class="org.springframework.jdbc.core.SqlParameter"> <constructor-arg index="0" value="custid"/> <constructor-arg index="1"> <util:constant static-field="java.sql.Types.INTEGER"/> </constructor-arg> </bean> </list> </property> <property name="refCursorPosition" value="1"/> <property name="rowMapper" ref="rowMapper"/> <property name="preparedStatementSetter" ref="parameterSetter"/> </bean>

除了参数声明, 我们还需要指定一个 PreparedStatementSetter 实现来设置参数值。这和上面的 JdbcCursorItemReader 一样。

1.9.2 可分页的 ItemReader

另一种是使用数据库游标执行多次查询,每次查询只返回一部分结果。 我们将这一部分称为一页(a page)。 分页时每次查询必须指定想要这一页的起始行号和想要返回的行数。

JdbcPagingItemReader

分页 ItemReader 的一个实现是 JdbcPagingItemReader 。 JdbcPagingItemReader 需要一个 PagingQueryProvider 来负责提供获取每一页所需的查询SQL。由于每个数据库都有不同的分页策略, 所以我们需要为各种数据库使用对应的PagingQueryProvider 。 也有自动检测所使用数据库类型的 SqlPagingQueryProviderFactoryBean ,会根据数据库类型选用适当的 PagingQueryProvider 实现。 这简化了配置,同时也是推荐的最佳实践。

SqlPagingQueryProviderFactoryBean 需要指定一个 select 子句以及一个 from 子句(clause). 当然还可以选择提供 where子句. 这些子句加上所需的排序列 sortKey 被组合成为一个 SQL 语句(statement).

在 reader 被打开以后, 每次调用 read 方法则返回一个 item,和其他的 ItemReader一样. 使用分页是因为可能需要额外的行.

下面是一个类似 ‘customer credit’ 示例的例子,使用上面提到的基于 cursor的ItemReaders:

<bean id="itemReader" class="org.spr...JdbcPagingItemReader"> <property name="dataSource" ref="dataSource"/> <property name="queryProvider"> <bean class="org.spr...SqlPagingQueryProviderFactoryBean"> <property name="selectClause" value="select id, name, credit"/> <property name="fromClause" value="from customer"/> <property name="whereClause" value="where status=:status"/> <property name="sortKey" value="id"/> </bean> </property> <property name="parameterValues"> <map> <entry key="status" value="NEW"/> </map> </property> <property name="pageSize" value="1000"/> <property name="rowMapper" ref="customerMapper"/> </bean>

这里配置的ItemReader将返回CustomerCredit对象, 必须指定使用的RowMapper。 ‘ pageSize ‘属性决定了每次数据库查询返回的实体数量。

‘ parameterValues ‘属性可用来为查询指定参数映射map。如果在where子句中使用了命名参数,那么这些entry的key应该和命名参数一一对应。如果使用传统的 ‘?’ 占位符, 则每个entry的key就应该是占位符的数字编号,和JDBC占位符一样索引都是从1开始。

JpaPagingItemReader

另一个分页ItemReader的实现是 JpaPagingItemReader 。JPA没有 Hibernate 中StatelessSession 之类的概念,所以我们必须使用JPA规范提供的其他功能。因为JPA支持分页,所以在使用JPA来处理分页时这是一种很自然的选择。读取每页后, 实体将会分离而且持久化上下文将会被清除,以允许在页面处理完成后实体会被垃圾回收。

JpaPagingItemReader 允许您声明一个JPQL语句,并传入一个 EntityManagerFactory 。然后就和其他的 ItemReader 一样,每次调用它的 read 方法都会返回一个 item. 当需要更多实体,则内部就会自动发生分页。下面是一个示例配置,和上面的JDBC reader一样,都是 ‘customer credit’:

<bean id="itemReader" class="org.spr...JpaPagingItemReader"> <property name="entityManagerFactory" ref="entityManagerFactory"/> <property name="queryString" value="select c from CustomerCredit c"/> <property name="pageSize" value="1000"/> </bean>

这里配置的ItemReader和前面所说的 JdbcPagingItemReader 返回一样的 CustomerCredit对象, 假设 Customer 对象有正确的JPA注解或者ORM映射文件。 ‘ pageSize ‘ 属性决定了每次查询时读取的实体数量。

IbatisPagingItemReader

如果使用 IBATIS/MyBatis, 则可以使用 IbatisPagingItemReader, 顾名思义, 也是一种实现分页的ItemReader。IBATIS不对分页提供直接支持, 但通过提供一些标准变量就可以为IBATIS查询提供分页支持。

下面是和上面的示例同样功能的配置,使用IbatisPagingItemReader来读取CustomerCredits:

<bean id="itemReader" class="org.spr...IbatisPagingItemReader"> <property name="sqlMapClient" ref="sqlMapClient"/> <property name="queryId" value="getPagedCustomerCredits"/> <property name="pageSize" value="1000"/> </bean>

上述 IbatisPagingItemReader 配置引用了一个IBATIS查询,名为“getPagedCustomerCredits”。如果使用MySQL,那么查询XML应该类似于下面这样。

<select id="getPagedCustomerCredits" resultMap="customerCreditResult"> select id, name, credit from customer order by id asc LIMIT #_skiprows#, #_pagesize# </select>

_skiprows 和 _pagesize 变量都是 IbatisPagingItemReader 提供的,还有一个 _page 变量,需要时也可以使用。分页查询的语法根据数据库不同使用。下面是使用Oracle的一个例子(但我们需要使用CDATA来包装某些特殊符号,因为是放在XML文档中嘛):

<select id="getPagedCustomerCredits" resultMap="customerCreditResult"> select * from ( select * from ( select t.id, t.name, t.credit, ROWNUM ROWNUM_ from customer t order by id )) where ROWNUM_ <![CDATA[ > ]]> ( #_page# * #_pagesize# ) ) where ROWNUM <![CDATA[ <= ]]> #_pagesize# </select>

1.9.3 Database ItemWriters

虽然文本文件和XML都有自己特定的 ItemWriter, 但数据库和他们并不一样。这是因为事务提供了所需的全部功能。 对于文件来说 ItemWriters 是必要的, 因为如果需要事务特性,他们必须充当这种角色, 跟踪输出的 item,并在适当的时间flushing/clearing。使用数据库时不需要这个功能,因为写已经包含在事务之中。 用户可以自己创建实现ItemWriter接口的DAO, 或使用一个处理常见问题的自定义ItemWriter,无论哪种方式,都不会有任何问题。 需要注意的一件事是批量输出时的性能和错误处理能力。 在使用hibernate作为ItemWriter 时是最常见的, 但在使用Jdbc batch 模式时可能也会存在同样的问题。批处理数据库输出没有任何固有的缺陷,如果我们注意 flush 并且数据没有错误的话。 但是,在写出时如果发生了什么错误,就可能会引起混乱,因为没有办法知道是哪个item引起的异常, 甚至是否某个单独的 item 负有责任,如下图所示:

如果 items 在输出之前有缓冲, 则遇到任何错误将不会立刻抛出, 直到缓冲区刷新之后,提交之前才会抛出。例如, 我们假设每一块写出20个item, 第15个 item 会抛出 DataIntegrityViolationException 。如果与 Step 有关, 则20项数据都会写入成功, 因为没有办法知道会出现错误,直到全部写入完成。一旦调用 Session#flush()， 就会清空缓冲区buffer, 而异常也将被放出来。在这一点上, Step无能为力, 事务也必须回滚。 通常, 异常会导致 item 被跳过(取决于 skip/retry 策略), 然后该item就不会被输出。 然而,在批处理的情况下, 是没有办法知道到底是哪一项引起的问题, 在错误发生时整个缓冲区都将被写出。解决这个问题的唯一方法就是在每一个 item之后 flush一下:

这种用法是很常见的, 尤其是在使用Hibernate时,ItemWriter的简单实现建议, 在每次调用 write() 之后执行 flush。这样做可以让跳过 items 变得可靠, 而Spring Batch 在错误发生后会在内部关注适当粒度的ItemWriter调用。

1.10 重用已存在的 Service

批处理系统通常是与其他应用程序相结合的方式使用。最常见的是与一个在线应用系统结合, 但也支持与瘦客户端集成,通过移动每个程序所使用的批量数据。由于这些原因,所以很多用户想要在批处理作业中重用现有的DAO或其他服务。Spring容器通过注入一些必要的类就可以实现这些重用。但可能需要现有的服务作为 ItemReader 或者 ItemWriter, 也可以适配另一个Spring Batch类, 或其本身就是一个 step 主要的ItemReader。为每个需要包装的服务编写一个适配器类是很简单的, 而因为这是很普遍的需求,所以 Spring Batch 提供了实现: ItemReaderAdapter 和 ItemWriterAdapter 。两个类都实现了标准的Spring方法委托模式调用，设置也相当简单。下面是一个reader的示例:

<bean id="itemReader" class="org.springframework.batch.item.adapter.ItemReaderAdapter"> <property name="targetObject" ref="fooService" /> <property name="targetMethod" value="generateFoo" /> </bean> <bean id="fooService" class="org.springframework.batch.item.sample.FooService" />

特别需要注意的是, targetMethod 必须和 read 方法行为对等: 如果不存在则返回null, 否则返回一个 Object。 其他的值会使框架不知道何时该结束处理, 或者引起无限循环或不正确的失败,这取决于 ItemWriter 的实现。 ItemWriter 的实现同样简单:

<bean id="itemWriter" class="org.springframework.batch.item.adapter.ItemWriterAdapter"> <property name="targetObject" ref="fooService" /> <property name="targetMethod" value="processFoo" /> </bean> <bean id="fooService" class="org.springframework.batch.item.sample.FooService" />

1.11 输入校验

在本章中, 已经讨论了很多种用来解析 input 的方法。 如果格式不对,那这些基本的实现都是抛出异常。 如果数据丢失一部分,FixedLengthTokenizer 也会抛出异常。 同样, 使用 FieldSetMapper 时,如果读取超出 RowMapper 索引范围的值,又或者返回值类型不匹配,都会抛出异常。 所有的异常都会在 read 返回之前抛出。 然而, 他们不能确定返回的item是否是合法的。 例如, 如果其中一个字段是 age , 很显然不能是负数。 解析为数字是没问题的, 因为确实存在这个数, 所以就不会抛出异常。 因为当下已经有大量的第三方验证框架, 所以 Spring Batch 并不提供另一个验证框架, 而是提供了一个非常简单的接口, 其他框架可以实现这个接口来提供兼容:

public interface Validator { void validate(Object value) throws ValidationException; }

The contract is that the validate method will throw an exception if the object is invalid, and return normally if it is valid.Spring Batch provides an out of the box ItemProcessor:

约定是如果对象无效则 validate 方法抛出一个异常, 如果对象合法那就正常返回。 Spring Batch 提供了开箱即用的ItemProcessor:

<bean class="org.springframework.batch.item.validator.ValidatingItemProcessor"> <property name="validator" ref="validator" /> </bean> <bean id="validator" class="org.springframework.batch.item.validator.SpringValidator"> <property name="validator"> <bean id="orderValidator" class="org.springmodules.validation.valang.ValangValidator"> <property name="valang"> <value> <![CDATA[ { orderId : ? > 0 AND ? <= 9999999999 : 'Incorrect order ID' : 'error.order.id' } { totalLines : ? = size(lineItems) : 'Bad count of order lines' : 'error.order.lines.badcount'} { customer.registered : customer.businessCustomer = FALSE OR ? = TRUE : 'Business customer must be registered' : 'error.customer.registration'} { customer.companyName : customer.businessCustomer = FALSE OR ? HAS TEXT : 'Company name for business customer is mandatory' :'error.customer.companyname'} ]]> </value> </property> </bean> </property> </bean>

这个示例展示了一个简单的 ValangValidator, 用来校验 order 对象。 这样写目的是为了尽可能多地演示如何使用 Valang 来添加校验程序。

1.12 不保存执行状态

默认情况下,所有 ItemReader 和 ItemWriter 在提交之前都会把当前状态信息保存到 ExecutionContext 中。 但有时我们又不希望保存这些信息。 例如,许多开发者使用处理指示器(process indicator)让数据库读取程序 ‘可重复运行(rerunnable)’。 在数据表中添加一个附加列来标识该记录是否已被处理。 当某条记录被读取/写入时,就将标志位从 false 变为 true , 然后只要在SQL语句的where子句中包含一个附加条件, 如 ” where PROCESSED_IND = false “, 就可确保在任务重启后只查询到未处理过的记录。 这种情况下,就不需要保存任何状态信息, 比如当前 row number 什么的, 因为在重启后这些信息都没用了。 基于这种考虑, 所有的 readers 和 writers 都含有一个 saveState 属性:

<bean id="playerSummarizationSource" class="org.spr...JdbcCursorItemReader"> <property name="dataSource" ref="dataSource" /> <property name="rowMapper"> <bean class="org.springframework.batch.sample.PlayerSummaryMapper" /> </property> <property name="saveState" value="false" /> <property name="sql"> <value> SELECT games.player_id, games.year_no, SUM(COMPLETES), SUM(ATTEMPTS), SUM(PASSING_YARDS), SUM(PASSING_TD), SUM(INTERCEPTIONS), SUM(RUSHES), SUM(RUSH_YARDS), SUM(RECEPTIONS), SUM(RECEPTIONS_YARDS), SUM(TOTAL_TD) from games, players where players.player_id = games.player_id group by games.player_id, games.year_no </value> </property> </bean>

上面配置的这个 ItemReader 在任何情况下都不会将 entries（状态信息）存放到 ExecutionContext 中.

1.13 创建自定义 ItemReaders 与 ItemWriters

到目前为止,本章已将 Spring Batch 中基本的读取(reading)和写入(writing)概念讲完, 还对一些常用的实现进行了讨论。然而,这些都是相当普通的, 还有很多潜在的场景可能没有现成的实现。本节将通过一个简单的例子,来演示如何创建自定义的 ItemReader 和 ItemWriter ,并且如何正确地实现和使用。 ItemReader 同时也将 ItemStream , 以说明如何让reader(读取器)或writer(写入器)支持重启(restartable)。

1.13.1 自定义 ItemReader 示例

为了实现这个目的,我们实现一个简单的 ItemReader , 从给定的list中读取数据。 我们将实现最基本的 ItemReader 功能, read:

public class CustomItemReader<T> implements ItemReader<T>{ List<T> items; public CustomItemReader(List<T> items) { this.items = items; } public T read() throws Exception, UnexpectedInputException,NoWorkFoundException, ParseException { if (!items.isEmpty()) { return items.remove(0); } return null; } }

这是一个简单的类, 传入一个 items list, 每次读取时删除其中的一条并返回。 如果list里面没有内容,则将返回null, 从而满足ItemReader 的基本要求, 测试代码如下所示:

List<String> items = new ArrayList<String>(); items.add("1"); items.add("2"); items.add("3"); ItemReader itemReader = new CustomItemReader<String>(items); assertEquals("1", itemReader.read()); assertEquals("2", itemReader.read()); assertEquals("3", itemReader.read()); assertNull(itemReader.read());

使 ItemReader 支持重启

现在剩下的问题就是让 ItemReader 变为可重启的。到目前这一步,如果发生掉电之类的情况,那么必须重新启动ItemReader,而且是从头开始。在很多时候这是允许的,但有时侯更好的处理办法是让批处理作业在上次中断的地方重新开始。判断的关键是根据 reader 是有状态的还是无状态的。 无状态的 reader 不需要考虑重启的情况, 但有状态的则需要根据其最后一个已知的状态来重新启动。出于这些原因, 官方建议尽可能地让 reader 成为无状态的,使开发者不需要考虑重新启动的情况。

如果需要保存状态信息,那应该使用 ItemStream 接口:

public class CustomItemReader<T> implements ItemReader<T>, ItemStream { List<T> items; int currentIndex = 0; private static final String CURRENT_INDEX = "current.index"; public CustomItemReader(List<T> items) { this.items = items; } public T read() throws Exception, UnexpectedInputException,ParseException { if (currentIndex < items.size()) { return items.get(currentIndex++); } return null; } public void open(ExecutionContext executionContext) throws ItemStreamException { if(executionContext.containsKey(CURRENT_INDEX)){ currentIndex = new Long(executionContext.getLong(CURRENT_INDEX)).intValue(); }else{ currentIndex = 0; } } public void update(ExecutionContext executionContext) throws ItemStreamException { executionContext.putLong(CURRENT_INDEX, new Long(currentIndex).longValue()); } public void close() throws ItemStreamException {} }

每次调用 ItemStream 的 update 方法时, ItemReader 的当前 index 都会被保存到给定的 ExecutionContext 中,key 为’ current.index ‘。 当调用 ItemStream 的 open 方法时, ExecutionContext会检查是否包含该 key 对应的条目。 如果找到key, 那么当前索引 index 就好移动到该位置。这是一个相当简单的例子,但它仍然符合通用原则:

ExecutionContext executionContext = new ExecutionContext(); ((ItemStream)itemReader).open(executionContext); assertEquals("1", itemReader.read()); ((ItemStream)itemReader).update(executionContext); List<String> items = new ArrayList<String>(); items.add("1"); items.add("2"); items.add("3"); itemReader = new CustomItemReader<String>(items); ((ItemStream)itemReader).open(executionContext); assertEquals("2", itemReader.read());

大多数ItemReaders具有更加复杂的重启逻辑。 例如 JdbcCursorItemReader , 存储了游标(Cursor)中最后所处理的行的row id。

还值得注意的是 ExecutionContext 中使用的 key 不应该过于简单。这是因为 ExecutionContext 被一个 Step 中的所有ItemStreams 共用。在大多数情况下,使用类名加上 key 的方式应该就足以保证唯一性。然而,在极端情况下, 同一个类的多个ItemStream 被用在同一个Step中时( 如需要输出两个文件的情况),就需要更加具备唯一性的name标识。出于这个原因,SpringBatch 的许多 ItemReader 和 ItemWriter 实现都有一个 setName() 方法, 允许覆盖默认的 key name。

1.13.2 自定义 ItemWriter 示例

自定义实现 ItemWriter 和上一小节所讲的 ItemReader 有很多方面是类似, 但也有足够多的不同之处。 但增加可重启特性在本质上是一样的, 所以本节的示例就不再讨论这一点。和 ItemReader 示例一样, 为了简单我们使用的参数也是 List :

public class CustomItemWriter<T> implements ItemWriter<T> { List<T> output = TransactionAwareProxyFactory.createTransactionalList(); public void write(List<? extends T> items) throws Exception { output.addAll(items); } public List<T> getOutput() { return output; } }

让 ItemWriter 支持重新启动

要让 ItemWriter 支持重新启动，我们将会使用和 ItemReader 相同的过程, 实现并添加 ItemStream 接口来同步 execution context。 在示例子中我们可能要记录处理过的items数量,并添加为到 footer 记录。 我们可以在 ItemWriter 的实现类中同时实现 ItemStream , 以便在 stream 重新打开时从执行上下文中取回原来的数据重建计数器。

实际开发中, 如果自定义 ItemWriter restartable(支持重启),则会委托另一个 writer(例如, 在写入文件时), 否则会写入到关系型数据库(支持事务的资源)中, 此时 ItemWriter 不需要 restartable特性,因为自身是无状态的。 如果你的 writer 有状态, 则应该实现2个接口: ItemStream 和 ItemWriter 。 请记住, writer客户端需要知道 ItemStream 的存在, 所以需要在 xml 配置文件中将其注册为 stream.