数据库Database - 基于游标的ItemReaders

6.9.1 基于Cursor的ItemReaders

使用游标(cursor)是大多数批处理开发人员默认采用的方法, 因为它是处理有关系的数据“流”在数据库级别的解决方案。Java 的 ResultSet 类其本质就是用面向对象的游标处理机制。 ResultSet 维护着一个指向当前数据行的 cursor。调用 ResultSet 的 next 方法则将游标移到下一行。

Spring Batch 基于 cursor 的 ItemReaders 在初始化时打开游标, 每次调用 read 时则将游标向前移动一行, 返回一个可用于进行处理的映射对象。最好将会调用 close 方法, 以确保所有资源都被释放。

Spring 的 JdbcTemplate 的解决办法, 是通过回调模式将 ResultSet 中所有行映射之后，在返回调用方法前关闭结果集来处理的。

但是,在批处理的时候就不一样了, 必须得等 step 执行完成才能调用close。下图描绘了基于游标的ItemReader是如何处理的, 使用的SQL语句非常简单, 而且都是类似的实现方式:

这个例子演示了基本的处理模式。 数据库中有一个 “FOO” 表,它有三个字段: ID, NAME, 以及 BAR , select 查询所有ID大于1但小于7的行。这样的话游标起始于 ID 为 2的行(第1行)。这一行的结果会被映射为一个Foo对象。再次调用read()则将光标移动到下一行, 也就是ID为3的Foo。 在所有行读取完毕之后这些结果将会被写出去, 然后这些对象就会被垃圾回收(假设没有其他引用指向他们)。

译注

Foo、Bar 都是英文中的任意代词,没有什么具体意义, 就如我们说的 张三,李四 一样

JdbcCursorItemReader

JdbcCursorItemReader 是基于 cursor 的Jdbc实现。它直接使用ResultSet，需要从数据库连接池中获取连接来执行SQL语句。我们的示例使用下面的数据库表:

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

我们一般使用领域对象来对应到每一行, 所以用 RowMapper 接口的实现来映射 CustomerCredit对象:

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;
    }
}

一般来说Spring的用户对 JdbcTemplate 都不陌生，而 JdbcCursorItemReader 使用其作为关键API接口, 我们一起来学习如何通过 JdbcTemplate 读取这一数据, 看看它与 ItemReader 有何区别。 为了演示方便, 我们假设CUSTOMER表有1000行数据。第一个例子将使用 JdbcTemplate:

//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());

当执行完上面的代码, customerCredits 这个 List 中将包含 1000 个 CustomerCredit 对象。 在 query 方法中, 先从 DataSource 获取一个连接, 然后用来执行给定的SQL, 获取结果后对 ResultSet 中的每一行调用一次 mapRow 方法。 让我们来对比一下 JdbcCursorItemReader 的实现:

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>

附加属性" class="reference-link">附加属性

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

需要整理

Table 6.2. JdbcCursorItemReader 的属性(Properties)

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 一样。查看全部附加属性请查看 附加属性, StoredProcedureItemReader 的附加属性也一样。