java - 处理多个 StoredProcedureItemReader 调用

Question

我必须在 Spring Batch 作业中实现以下用例:

1. 通过StoredProcedureItemReader读取提供程序列表
2. 迭代列表并为步骤 1 中找到的每个提供程序(作为输入参数)调用另一个 StoredProcedureItemReader。
3. 第二个 SP 的输出将写入 CSV。

我提出了以下策略:

1. 第 1 步开始
2. SP ItemReader 返回提供者列表。
3. 在 ItemWriter 中，将提供程序保存到 ExecutionContext
4. 第 1 步结束
5. 第 2 步开始
6. 另一个 SP ItemReader 从 ExecutionContext 访问提供程序
7. 另一个 ItemWriter 使用 FlatFileItemWriter 将响应写入 CSV

我无法理解第二个 SP ItemReader 将如何访问第一个 SP ItemReader 返回的提供程序列表。分区在这里有帮助吗？另外，是否有更好的策略来实现这一目标？

编辑1:

这是 ItemProcessor 的原始实现(作为丰富):

@Scope("step")
public class FetchReportFromProviderProcessor implements
        ItemProcessor<RiscProvider, List<SogReportRecord>>, ItemStream {

    StoredProcedureItemReader<SogReportRecord> reader = new StoredProcedureItemReader<SogReportRecord>();
    private DataSource dataSource;

    @Value("#{jobParameters['date']}")
    private String date;

    @Override
    public List<SogReportRecord> process(final RiscProvider item) throws Exception {

        SogReportRecord record = null;
        List<SogReportRecord> records = new ArrayList<SogReportRecord>();

        SqlParameter[] sqlParameters = new SqlParameter[] {new SqlParameter(OracleTypes.CURSOR)};           

        reader.setParameters(sqlParameters);
        reader.setPreparedStatementSetter(new PreparedStatementSetter() {

            @Override
            public void setValues(PreparedStatement ps) throws SQLException {
                ps.setString(0, item.getPrefix());
                ps.setString(1, date);
            }
        });

        while( (record = reader.read()) != null ) {
            records.add(record);
        }

        return records;
    }

    public DataSource getDataSource() {
        return dataSource;
    }

    public void setDataSource(DataSource dataSource) {
        this.dataSource = dataSource;
    }

    @Override
    public void open(ExecutionContext executionContext)
            throws ItemStreamException {
        reader.setDataSource(dataSource);
        reader.setProcedureName("RISC_GET_DAYMOVEINOUT");
        reader.open(executionContext);      
    }

    @Override
    public void update(ExecutionContext executionContext)
            throws ItemStreamException {
        reader.update(executionContext);        
    }

    @Override
    public void close() throws ItemStreamException {
        reader.close();     
    }

}

以及 XML 部分:

<batch:job id="SOG_MOVEINOUT_REPORT_GENERATOR">
    <batch:step id="GET_REPORTS">
        <batch:tasklet>
            <batch:chunk reader="getProviders"
                processor="fetchRecordsFromProvider"
                writer="sogReportWriter" commit-interval="500" />
        </batch:tasklet>
    </batch:step>
</batch:job>

<!-- Reader to fetch list of providers -->
<bean id="getProviders" class="org.springframework.batch.item.database.StoredProcedureItemReader">
    <property name="dataSource" ref="dataSource" />
    <property name="procedureName" value="RISC_GET_PROVIDER" />
    <property name="parameters">
        <list>
            <bean class="org.springframework.jdbc.core.SqlOutParameter">
                <constructor-arg index="0" value="providers" />
                <constructor-arg index="1">
                    <util:constant static-field="oracle.jdbc.OracleTypes.CURSOR" />
                </constructor-arg>
            </bean>
        </list>
    </property>
    <property name="refCursorPosition" value="1" />
    <property name="rowMapper">
        <bean class="com.kpn.risc.ProviderRowMapper" />
    </property>
</bean>

<bean id="fetchRecordsFromProvider" class="com.kpn.risc.FetchReportFromProviderProcessor">
    <property name="dataSource" ref="dataSource" />
</bean>

<bean id="sogReportWriter" class="org.springframework.batch.item.file.FlatFileItemWriter" scope="step">
    <property name="resource" value="file:///${batch.job.report.dir}/report-#{stepExecutionContext['provider']}.csv" />
    <property name="lineAggregator">
        <bean class="org.springframework.batch.item.file.transform.DelimitedLineAggregator">
            <property name="fieldExtractor">
                <bean class="org.springframework.batch.item.file.transform.PassThroughFieldExtractor" />
            </property>
        </bean>
    </property>
</bean>

在上面的代码中，处理器将其工作委托(delegate)给 SP ItemReader。但在调用 read() 方法之前，读取器无法正确初始化。是否可以或建议在 ItemProcessor 内调用 ItemReader？

Answer 1

您所描述的内容直接属于批处理的驱动查询模式。本质上，一个查询定义了另一个查询用来驱动其查询的 id(或者在您的情况下为提供者)。通常，ItemReader 读取 id 并将每个 ID 传递给 ItemProcessor 进行丰富(“其他”查询)。然后结果被传递给 ItemWriter。

您可以在 Spring Batch 文档的通用批处理模式部分中阅读有关驱动查询模式的更多信息:http://docs.spring.io/spring-batch/trunk/reference/html/patterns.html