0

在下文中，我的写入是否一次提交到数据库 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 表也将在处理结束时更新一些列。

这是我到目前为止使用异步处理和写入的内容，目前我正在测试仅读取 2,500 条 mdw 记录，在此测试中，所有 2,500 条记录都是故意无效的，因此被插入到 mec_kickout 表中，而原始记录mdw 表 2,500 行也正在更新。所有这一切都在大约 50 秒内完成，在一台 8 核 CPU 笔记本电脑上，具有 16 Gb 的 RAM，并且一个 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);
    }

}
4

1 回答 1

0

1）这里的博客帮助我理解了 Spring Batch 中的事务。

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

于 2015-08-24T07:33:17.913 回答