处理多个StoredProcedureItemReader调用

Question

我必须在Spring批处理作业中实现以下用例：

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

我提出了以下策略：

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

@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>

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

Answer 1

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

您可以在Spring Batch文档的Common Batch Patterns一节中阅读更多关于驱动查询模式的信息：http://docs.Spring.io/spring-batch/trunk/reference/html/Patterns.html