Geek asks and answers

In Spring Batch, can multiple JdbcBatchItemWriters be configured to write in parallel?

In my spring batch job, my product processor splits the object that the item reader reads into seven lists of variable lengths. These lists should be recorded in seven tables in the database, and any errors (for example, rejection of db records for any reason) should lead to a rollback of the transaction in all seven tables.

I am currently creating a wrapped object with these seven lists that are passed to the item producer. The writer takes all of these elements, creates his seven lists, so that he has only seven batch records (using the JdbcTemplate based DAO) for the batch of wrapped objects returned by the element processor.

My writer sequentially calls the insert function for each of these tables, which I would like to speed up. I was wondering if it is possible to write lists to the corresponding tables in parallel so that the total execution time is the time of the longest record. One of the requirements that I cannot compromise is that it must be in one transaction, which should be discarded if any of the authors have any exceptions.

+5
source share
1 answer

here's a simple solution using TaskExecutor and extension on org.springframework.batch.item.support.CompositeItemWriter.

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");
    }



}

An example configuration would look something like this:

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

This example uses the following:

  • Bitronix JTA for multi-database transaction support
  • very simple model of a simple object in a simple jdbc record

( )

+6

More articles:


All Articles