了解 Spring Batch 中的 AsyncItemWriter，以及何时使用 MyBatis 编写块？

Question

在下面，我的写入是否一次提交到数据库1000——因为我在Spring Batch作业中的提交间隔设置为1000？MyBatis SqlSessionFactory被定义为BATCH执行：

<bean id="sqlSession" class="org.mybatis.spring.SqlSessionTemplate">
    <constructor-arg index="0" ref="sqlSessionFactory" />
    <constructor-arg index="1" value="BATCH" />
</bean>
<!-- define the SqlSessionFactory -->
<bean id="sqlSessionFactory" class="org.mybatis.spring.SqlSessionFactoryBean">
    <property name="dataSource" ref="dataSource" />
    <property name="typeAliasesPackage" value="org.my.domain" />
</bean>

我在调试日志中注意到：

2015-08-21 22:58:54,632 [main] DEBUG org.mybatis.spring.SqlSessionUtils - Fetched SqlSession [org.apache.ibatis.session.defaults.DefaultSqlSession@174b870c] from current transaction

在上面的 DEBUG 语句中，我认为它正在打开连接。然后在插入语句下方插入到一批 1，000 个中 - 或者它是否为每个插入打开了与 MS SQL 服务器的新连接？

2015-08-21 22:58:54,632 [main] DEBUG org.mybatis.spring.SqlSessionUtils - Fetched SqlSession [org.apache.ibatis.session.defaults.DefaultSqlSession@174b870c] from current transaction
KickoutMapper.insertKickoutTbl - ==> Parameters: 12143(Long), 10039(Long), 0(Integer), SUBSCRIBER4998(String), .....
2015-08-21 22:58:54,632 [main] DEBUG org.mybatis.spring.SqlSessionUtils - Releasing transactional SqlSession [org.apache.ibatis.session.defaults.DefaultSqlSession@174b870c]

最后它正在关闭事务（同样，我不确定它是否正在关闭与MS SQL服务器的连接，或者是否在每次插入后都这样做了）。同样，我想批量插入 1，000 条记录，并与数据库建立一条连接。

2015-08-21 22:58:55,376 [main] DEBUG org.mybatis.spring.SqlSessionUtils - Releasing transactional SqlSession [org.apache.ibatis.session.defaults.DefaultSqlSession@174b870c]

此外，AsyncItemWriter是否可以改为委托给JdbcBatchItemWriter？它是否比我配置的MyBatis更快？在下面的代码中，我委托给CompositeItemWriter，它使用MyBatis映射器写入两个单独的表。

我的用例是这样的:我需要从一个mec_mdw数据库表中读取并验证700万条记录。

处理器认为无效的mec_mdw记录将被插入mec_kickout表。此外，该特定记录的验证失败原因将插入mec_kickout_reason表中。我最初从中读取的mec_mdw表也将在处理结束时更新一些列。

这是迄今为止我使用异步处理和写入所做的，目前我只测试了2500个读取的mdw记录，在这个测试中，所有2500个记录都是故意无效的，因此被插入到mec_kickout表中，原始的mdw表2500行也在更新中。所有这些都在大约50秒内完成，在一台拥有16 Gb RAM的8核CPU笔记本电脑上，MS SQL服务器数据库需要网络呼叫。但我仍然不完全理解是否可以更快地完成。

    <job id="mecmdwvalidatorJob" xmlns="http://www.springframework.org/schema/batch">
    <step id="mdwvalidatorStep1">
        <tasklet>
            <chunk reader="pageItemReader" processor="asyncItemProcessor"
                writer="asynchItemWriter" commit-interval="1000" skip-limit="2147483647">
                <skippable-exception-classes> <!-- TODO -->
                    <include class="java.lang.Exception" />
                </skippable-exception-classes>
            </chunk>
        </tasklet>
    </step>
</job>

<bean id="pageItemReader"
    class="org.springframework.batch.item.database.JdbcPagingItemReader">
    <property name="dataSource" ref="dataSource" />
    <property name="queryProvider">
        <bean
            class="org.springframework.batch.item.database.support.SqlPagingQueryProviderFactoryBean">
            <property name="dataSource" ref="dataSource" />
            <property name="selectClause"
                value="select MDW_ID,FK_LOG_FILE_ID,TAX_YEAR,SUBS_TYPE_CD,SUB_FIRST_NM,SUB_MIDDLE_NM,SUB_LAST_NM,SUB_SUFFIX,SUB_DOB,SUB_ADDR1,SUB_ADDR2,SUB_CITY,SUB_STATE,SUB_PROVINCE,SUB_ZIP,SUB_ZIP4,SUB_COUNTRY_CD,SUB_COUNTRY,SUB_F_POSTAL_CD,LOB,SUB_SSN,GRP_EMP_NAME1,GRP_EMP_NAME2,GRP_EIN,GRP_ADDR1,GRP_ADDR2,GRP_CITY,GRP_STATE,GRP_PROVINCE,GRP_ZIP,GRP_ZIP4,GRP_COUNTRY_CD,GRP_COUNTRY,GRP_F_POSTAL_CD,ISSUER_NAME1,ISSUER_NAME2,ISSUER_PHONE,ISSUER_ADDR1,ISSUER_ADDR2,ISSUER_CITY,ISSUER_PROVINCE,ISSUER_ZIP,ISSUER_ZIP4,ISSUER_COUNTRY_CD,ISSUER_COUNTRY,ISSUER_F_POSTAL_CD,MEM_FIRST_NM,MEM_MIDDLE_NM,MEM_LAST_NM,MEM_SUFFIX,MEM_SSN,MEM_DOB,MEM_START_DATE,MEM_END_DATE,REGION_CD,SUB_MRN,SUB_MRN_PREFIX,MEM_MRN,MRN_PREFIX,PID,SUB_GRP_ID,SUB_GRP_NAME,INVALID_ADDR_FL" />
            <property name="fromClause"
                value="from MEC_MDW JOIN MEC_FILE_LOG on MEC_FILE_LOG.LOG_FILE_ID=MEC_MDW.FK_LOG_FILE_ID  " />
            <property name="whereClause" value="where MEC_FILE_LOG.STATUS=:status" />
            <property name="sortKey" value="MDW_ID" />
        </bean>
    </property>
    <property name="parameterValues">
        <map>
            <entry key="status" value="READY TO VALIDATE" />
        </map>
    </property>
    <property name="pageSize" value="1000" />
    <property name="rowMapper" ref="mdwRowMapper" />
</bean>
<bean id="mdwRowMapper" class="org.my.rowmapper.MdwRowMapper" />

<bean id="asyncItemProcessor"
    class="org.springframework.batch.integration.async.AsyncItemProcessor">
    <property name="delegate">
        <bean
            class="org.my.itemprocessor.MdwValidatingItemProcessor">
            <property name="validator">
                <bean
                    class="org.springframework.validation.beanvalidation.LocalValidatorFactoryBean" />
            </property>
        </bean>
    </property>
    <property name="taskExecutor" ref="taskExecutor" />
    <!-- <property name="taskExecutor"> -->
    <!-- <bean class="org.springframework.core.task.SimpleAsyncTaskExecutor" 
        /> -->
    <!-- </property> -->
</bean>

<task:executor id="taskExecutor" pool-size="10" />

<bean id="asynchItemWriter"
    class="org.springframework.batch.integration.async.AsyncItemWriter">
    <property name="delegate" ref="customerCompositeWriter">
    </property>
</bean>

<bean id="customerCompositeWriter"
    class="org.springframework.batch.item.support.CompositeItemWriter">
    <property name="delegates">
        <list>
            <ref bean="itemWriter1" />
            <ref bean="itemWriter2" />
        </list>
    </property>
</bean>

<bean id="itemWriter1" class="org.my.writer.MdwWriter" />
<bean id="itemWriter2" class="org.my.writer.KickoutWriter" />

</beans>

处理器将具有业务逻辑，现在它只有 Bean 验证，在 MecMdw 域对象中查找空属性。最终，我还需要适配器代码来查找其他表中的帐户 ID（因此更多的数据库连接乐趣！我认为这个适配器DAO逻辑将进入处理器

public class MdwValidatingItemProcessor implements ItemProcessor<MecMdw, MecMdw> {  

private Validator validator;  

public void setValidator(Validator validator) {  
     this.validator = validator;  
}  

public MecMdw process(MecMdw item) throws Exception {  

     BindingResult results = BindAndValidate(item);

     if (results.hasErrors())  {

         item.setKick_out_fl('Y');

         buildValidationException(results,item);

         return item;  

     }

     return item;  
}  

private BindingResult BindAndValidate(MecMdw item) {

     DataBinder binder = new DataBinder(item);

     binder.setValidator(validator);  

     binder.validate();  

     return binder.getBindingResult();  

}  

private void buildValidationException(BindingResult results, MecMdw item) {  

     List<String> listOfErrors = new ArrayList<String>();

     for (ObjectError error : results.getAllErrors()) {  
          listOfErrors.add(error.toString());  
     }  

     item.setValidationErrors(listOfErrors);
}  

MdwWriter和KickoutWriter正在使用MyBatis DAO写入数据库。

public class MdwWriter<MecMdw> implements ItemWriter<MecMdw> {

@Autowired
MdwMapper mdwMapper;


@Override
public void write(List<? extends MecMdw> items) throws Exception {
    for(MecMdw item : items){
        mdwMapper.setMecMdwRecordAsKickOut((org.my.domain.MecMdw) item);

    }
}

这是KickoutWriter.java

public class KickoutWriter<MecMdw> implements ItemWriter<MecMdw> {

@Autowired
KickoutMapper kickoutMapper;

@Override
public void write(List<? extends MecMdw> items) throws Exception {
    for(MecMdw item : items){

        kickoutMapper.insertKickoutTbl((org.my.domain.MecMdw) item);
    }

}

Answer 1

1）这里的博客帮助我了解了Spring Batch中的交易。

2）关于性能，您是否尝试过使用Spring Batch的多线程功能？使用线程向上扩展可能足以满足性能要求。