在Spring Batch中，是否可以将多个JDBCBatchItemWriter配置为并行写入？

Question

在spring批处理作业中，项目处理器将项目读取器读取的对象拆分为七个长度可变的列表。这些列表必须写入DB中的七个表，任何错误（如DB出于任何原因拒绝记录）都必须导致事务在所有七个表上回滚。

目前，我使用这七个列表创建一个包装对象，这些列表将传递给自定义项目编写器。作者获取所有这些项目，创建自己的七个列表，以便它只有七个批处理写入（使用基于JdbcTemboard的DAO）用于项目处理器返回的一批包装对象。

我的编写器按顺序为每个表调用insert函数，我希望加快速度。我想知道是否可以将列表并行地写入各自的表中，以便总的执行时间是最长的写入时间。我不能妥协的一个要求是，这必须在单个事务中，如果任何写入程序有任何异常，则需要回滚该事务。

Answer 1

这里有一个简单的解决方案，它利用TaskExecitor并扩展到CompositeItemWriterorg.springframework.batch.item.support.。

package de.incompleteco.spring.batch.item.support;

import java.util.List;

import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.support.CompositeItemWriter;
import org.springframework.core.task.TaskExecutor;
import org.springframework.util.Assert;

import de.incompleteco.spring.domain.SimpleEntity;

public class ParallelCompositeItemWriter extends CompositeItemWriter<SimpleEntity> {

    private List<ItemWriter<? super SimpleEntity>> delegates;

    private TaskExecutor taskExecutor;

    @Override
    public void write(final List<? extends SimpleEntity> item) throws Exception {
        for (final ItemWriter<? super SimpleEntity> writer : delegates) {
            taskExecutor.execute(new Runnable()  {
                @Override
                public void run() {
                    try {
                        writer.write(item);
                    } catch (Throwable t) {
                        rethrow(t);
                    }   
                }

                private void rethrow(Throwable t) {
                    if (t instanceof RuntimeException) {
                        throw (RuntimeException) t;
                    }
                    else if (t instanceof Error) {
                        throw (Error) t;
                    }
                    throw new IllegalStateException(t);
                }       
            });
        }//end for
    }


    public void setTaskExecutor(TaskExecutor taskExecutor) {
        this.taskExecutor = taskExecutor;
    }

    @Override
    public void setDelegates(List<ItemWriter<? super SimpleEntity>> delegates) {
        this.delegates = delegates;
        super.setDelegates(delegates);
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        super.afterPropertiesSet();
        Assert.notNull(taskExecutor,"Task executor needs to be set");
    }



}

示例配置如下所示；

<batch:job id="simpleJob">
    <batch:step id="simpleJob.step1">
        <batch:tasklet>
            <batch:chunk reader="reader" writer="writer" commit-interval="10"/>
        </batch:tasklet>
    </batch:step>
</batch:job>

<bean id="reader" class="org.springframework.batch.item.support.IteratorItemReader">
    <constructor-arg ref="itemList"/>
</bean>

<bean id="writer" class="de.incompleteco.spring.batch.item.support.ParallelCompositeItemWriter">
    <property name="delegates" ref="writerDelegates"/>
    <property name="taskExecutor" ref="writerTaskExecutor"/>
</bean>

<util:list id="writerDelegates">
    <bean class="org.springframework.batch.item.database.JdbcBatchItemWriter">
        <property name="dataSource" ref="dataSource1"/>
        <property name="sql" value="insert into test_table (idcol,stuff) values (:idCol,:stuff)"/>
        <property name="itemSqlParameterSourceProvider">
            <bean class="org.springframework.batch.item.database.BeanPropertyItemSqlParameterSourceProvider"/>
        </property>
    </bean>
    <bean class="org.springframework.batch.item.database.JdbcBatchItemWriter">
        <property name="dataSource" ref="dataSource2"/>
        <property name="sql" value="insert into test_table (idcol,stuff) values (:idCol,:stuff)"/>
        <property name="itemSqlParameterSourceProvider">
            <bean class="org.springframework.batch.item.database.BeanPropertyItemSqlParameterSourceProvider"/>
        </property>
    </bean>     
</util:list>

<util:list id="itemList">
    <bean class="de.incompleteco.spring.domain.SimpleEntity">
        <constructor-arg value="stuff1"/>
    </bean>
    <bean class="de.incompleteco.spring.domain.SimpleEntity">
        <constructor-arg value="stuff2"/>
    </bean>     
    <bean class="de.incompleteco.spring.domain.SimpleEntity">
        <constructor-arg value="stuff3"/>
    </bean>     
</util:list>

<task:executor id="writerTaskExecutor" pool-size="3"/>


<bean id="dataSource1" class="bitronix.tm.resource.jdbc.PoolingDataSource" init-method="init" destroy-method="close">
    <property name="className" value="org.h2.jdbcx.JdbcDataSource" />
    <property name="uniqueName" value="#{T(System).currentTimeMillis()}" />
    <property name="allowLocalTransactions" value="true"/>
    <property name="maxPoolSize" value="2" />
    <property name="driverProperties">
        <props>
            <prop key="URL">jdbc:h2:mem:a;DB_CLOSE_DELAY=-1</prop>
        </props>
    </property>
</bean> 

<bean id="dataSource2" class="bitronix.tm.resource.jdbc.PoolingDataSource" init-method="init" destroy-method="close">
    <property name="className" value="org.h2.jdbcx.JdbcDataSource" />
    <property name="uniqueName" value="#{T(System).currentTimeMillis()}" />
    <property name="allowLocalTransactions" value="true"/>
    <property name="maxPoolSize" value="2" />
    <property name="driverProperties">
        <props>
            <prop key="URL">jdbc:h2:mem:b;DB_CLOSE_DELAY=-1</prop>
        </props>
    </property>
</bean>     

<jdbc:initialize-database  data-source="dataSource1">
    <jdbc:script location="classpath:/META-INF/sql/schema-h2.sql"/>
</jdbc:initialize-database>

<jdbc:initialize-database  data-source="dataSource2">
    <jdbc:script location="classpath:/META-INF/sql/schema-h2.sql"/>
</jdbc:initialize-database>
<!-- XA transaction -->

<bean id="btmConfig" factory-method="getConfiguration" class="bitronix.tm.TransactionManagerServices"/>

<bean id="BitronixTransactionManager" factory-method="getTransactionManager"
    class="bitronix.tm.TransactionManagerServices" depends-on="btmConfig" destroy-method="shutdown" />

<bean id="transactionManager" class="org.springframework.transaction.jta.JtaTransactionManager">
    <property name="transactionManager" ref="BitronixTransactionManager" />
    <property name="userTransaction" ref="BitronixTransactionManager" />
</bean>

此示例使用以下内容：；

• Bitronix JTA支持跨多个数据库的交易
• 将一个简单实体的非常简单的模型转换为一个简单的jdbc记录

（数据库中的内容非常粗糙，只是一个例子）