diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/AbstractPagingItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/AbstractPagingItemReader.java index 8025c9400..e64cb989b 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/AbstractPagingItemReader.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/AbstractPagingItemReader.java @@ -15,47 +15,68 @@ */ package org.springframework.batch.item.database; -import org.springframework.batch.item.support.AbstractItemCountingItemStreamItemReader; -import org.springframework.beans.factory.InitializingBean; -import org.springframework.util.ClassUtils; -import org.springframework.util.Assert; +import java.util.List; +import java.util.concurrent.atomic.AtomicInteger; + import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; - -import java.util.List; +import org.springframework.batch.item.support.AbstractItemCountingItemStreamItemReader; +import org.springframework.beans.factory.InitializingBean; +import org.springframework.util.Assert; +import org.springframework.util.ClassUtils; /** - * Abstract {@link org.springframework.batch.item.ItemReader} for to extend when reading database records in a paging - * fashion. - * - * Implementations should execute queries using paged requests of a size specified in {@link #setPageSize(int)}. - * Additional pages are requested when needed as {@link #read()} method is called, returning an - * object corresponding to current position. - * + * Abstract {@link org.springframework.batch.item.ItemReader} for to extend when + * reading database records in a paging fashion. + * + * Implementations should execute queries using paged requests of a size + * specified in {@link #setPageSize(int)}. Additional pages are requested when + * needed as {@link #read()} method is called, returning an object corresponding + * to current position. + * * @author Thomas Risberg * @since 2.0 */ -public abstract class AbstractPagingItemReader extends AbstractItemCountingItemStreamItemReader implements InitializingBean { +public abstract class AbstractPagingItemReader extends AbstractItemCountingItemStreamItemReader implements + InitializingBean { protected Log logger = LogFactory.getLog(getClass()); - protected boolean initialized = false; + private volatile boolean initialized = false; - protected int current = 0; + private int pageSize = 10; - protected int page = 0; + private volatile AtomicInteger current = new AtomicInteger(0); - protected int pageSize = 10; + private volatile int page = 0; - protected List results; + protected volatile List results; + + private Object lock = new Object(); public AbstractPagingItemReader() { setName(ClassUtils.getShortName(AbstractPagingItemReader.class)); } /** - * The number of rows to retreive at a time. - * + * The current page number. + * @return the current page + */ + public int getPage() { + return page; + } + + /** + * The page size configured for this reader. + * @return the page size + */ + public int getPageSize() { + return pageSize; + } + + /** + * The number of rows to retrieve at a time. + * * @param pageSize the number of rows to fetch per page */ public void setPageSize(int pageSize) { @@ -73,22 +94,26 @@ public abstract class AbstractPagingItemReader extends AbstractItemCountingIt @Override protected T doRead() throws Exception { - if (results == null || current >= pageSize) { + if (results == null || current.get() >= pageSize) { if (logger.isDebugEnabled()) { - logger.debug("Reading page " + page); + logger.debug("Reading page " + getPage()); } - doReadPage(); - - if (current >= pageSize) { - current = 0; + synchronized (lock) { + if (results == null || current.get() >= pageSize) { + doReadPage(); + page++; + current.set(0); + } } - page++; + + } - if (current < results.size()) { - return results.get(current++); + int next = current.getAndIncrement(); + if (next < results.size()) { + return results.get(next); } else { return null; @@ -102,7 +127,6 @@ public abstract class AbstractPagingItemReader extends AbstractItemCountingIt protected void doOpen() throws Exception { Assert.state(!initialized, "Cannot open an already opened ItemReader, call close first"); - initialized = true; } @@ -111,22 +135,22 @@ public abstract class AbstractPagingItemReader extends AbstractItemCountingIt protected void doClose() throws Exception { initialized = false; - current = 0; + current.set(0); page = 0; results = null; - } + } @Override protected void jumpToItem(int itemIndex) throws Exception { page = itemIndex / pageSize; - current = itemIndex % pageSize; + current.set(itemIndex % pageSize); doJumpToPage(itemIndex); if (logger.isDebugEnabled()) { - logger.debug("Jumping to page " + page + " and index " + current); + logger.debug("Jumping to page " + getPage() + " and index " + current); } } diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/IbatisPagingItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/IbatisPagingItemReader.java index 5fe0b26de..359bd08ee 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/IbatisPagingItemReader.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/IbatisPagingItemReader.java @@ -22,6 +22,7 @@ import org.springframework.orm.ibatis.SqlMapClientTemplate; import java.util.Map; import java.util.HashMap; +import java.util.concurrent.CopyOnWriteArrayList; import com.ibatis.sqlmap.client.SqlMapClient; @@ -59,7 +60,9 @@ import com.ibatis.sqlmap.client.SqlMapClient; *

* *

- * The implementation is *not* thread-safe. + * The implementation is thread-safe, but remember to use + * saveState=false if used in a multi-threaded client (no restart + * available). *

* * @author Thomas Risberg @@ -115,10 +118,16 @@ public class IbatisPagingItemReader extends AbstractPagingItemReader { if (parameterValues != null) { parameters.putAll(parameterValues); } - parameters.put("_page", page); - parameters.put("_pagesize", pageSize); - parameters.put("_skiprows", page * pageSize); - results = sqlMapClientTemplate.queryForList(queryId, parameters); + parameters.put("_page", getPage()); + parameters.put("_pagesize", getPageSize()); + parameters.put("_skiprows", getPage() * getPageSize()); + if (results == null) { + results = new CopyOnWriteArrayList(); + } + else { + results.clear(); + } + results.addAll(sqlMapClientTemplate.queryForList(queryId, parameters)); } @Override diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/JdbcPagingItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/JdbcPagingItemReader.java index c47b0efe5..dba296a4c 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/JdbcPagingItemReader.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/JdbcPagingItemReader.java @@ -16,42 +16,58 @@ package org.springframework.batch.item.database; +import java.sql.ResultSet; +import java.sql.SQLException; +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.SortedMap; +import java.util.TreeMap; +import java.util.concurrent.CopyOnWriteArrayList; + +import javax.sql.DataSource; + import org.springframework.beans.factory.InitializingBean; -import org.springframework.util.ClassUtils; -import org.springframework.util.Assert; import org.springframework.jdbc.core.JdbcTemplate; import org.springframework.jdbc.core.RowCallbackHandler; import org.springframework.jdbc.core.RowMapper; import org.springframework.jdbc.core.simple.SimpleJdbcTemplate; - -import javax.sql.DataSource; -import java.util.ArrayList; -import java.util.Map; -import java.util.List; -import java.util.LinkedHashMap; -import java.util.SortedMap; -import java.util.TreeMap; -import java.sql.ResultSet; -import java.sql.SQLException; +import org.springframework.util.Assert; +import org.springframework.util.ClassUtils; /** - * {@link org.springframework.batch.item.ItemReader} for reading database records using JDBC in a paging - * fashion. - * - * It executes the SQL built by the {@link PagingQueryProvider} to retrieve requested data. - * The query is executed using paged requests of a size specified in {@link #setPageSize(int)}. - * Additional pages are requested when needed as {@link #read()} method is called, returning an - * object corresponding to current position. - * - * The performance of the paging depends on the database specific features available to limit the number - * of returned rows. - * - * Setting a fairly large page size and using a commit interval that matches the page size should provide - * better performance. - * - * The implementation is *not* thread-safe. - * + *

+ * {@link org.springframework.batch.item.ItemReader} for reading database + * records using JDBC in a paging fashion. + *

+ * + *

+ * It executes the SQL built by the {@link PagingQueryProvider} to retrieve + * requested data. The query is executed using paged requests of a size + * specified in {@link #setPageSize(int)}. Additional pages are requested when + * needed as {@link #read()} method is called, returning an object corresponding + * to current position. + *

+ * + *

+ * The performance of the paging depends on the database specific features + * available to limit the number of returned rows. + *

+ * + *

+ * Setting a fairly large page size and using a commit interval that matches the + * page size should provide better performance. + *

+ * + *

+ * The implementation is thread-safe, but remember to use + * saveState=false if used in a multi-threaded client (no restart + * available). + *

+ * * @author Thomas Risberg + * @author Dave Syer * @since 2.0 */ public class JdbcPagingItemReader extends AbstractPagingItemReader implements InitializingBean { @@ -103,20 +119,24 @@ public class JdbcPagingItemReader extends AbstractPagingItemReader impleme /** * The row mapper implementation to be used by this reader - * - * @param rowMapper a {@link org.springframework.jdbc.core.simple.ParameterizedRowMapper} implementation + * + * @param rowMapper a + * {@link org.springframework.jdbc.core.simple.ParameterizedRowMapper} + * implementation */ public void setRowMapper(RowMapper rowMapper) { this.rowMapper = rowMapper; } /** - * The parameter values to be used for the query execution. If you use named parameters then the - * key should be the name used in the query clause. If you use "?" placeholders then the key should be - * the relative index that the parameter appears in the query string built using the select, from and - * where cluases specified. - * - * @param parameterValues the values keyed by the parameter named/index used in the query string. + * The parameter values to be used for the query execution. If you use named + * parameters then the key should be the name used in the query clause. If + * you use "?" placeholders then the key should be the relative index that + * the parameter appears in the query string built using the select, from + * and where cluases specified. + * + * @param parameterValues the values keyed by the parameter named/index used + * in the query string. */ public void setParameterValues(Map parameterValues) { this.parameterValues = parameterValues; @@ -133,49 +153,34 @@ public class JdbcPagingItemReader extends AbstractPagingItemReader impleme if (fetchSize != VALUE_NOT_SET) { jdbcTemplate.setFetchSize(fetchSize); } - jdbcTemplate.setMaxRows(pageSize); + jdbcTemplate.setMaxRows(getPageSize()); this.simpleJdbcTemplate = new SimpleJdbcTemplate(jdbcTemplate); Assert.notNull(queryProvider); queryProvider.init(dataSource); - this.firstPageSql = queryProvider.generateFirstPageQuery(pageSize); - this.remainingPagesSql = queryProvider.generateRemainingPagesQuery(pageSize); + this.firstPageSql = queryProvider.generateFirstPageQuery(getPageSize()); + this.remainingPagesSql = queryProvider.generateRemainingPagesQuery(getPageSize()); } @Override protected void doReadPage() { - if (results == null) { - results = new ArrayList(); - } - else { - results.clear(); - } - - if (page == 0) { + PagingRowCallbackHandler rowCallback = new PagingRowCallbackHandler(); + if (getPage() == 0) { if (logger.isDebugEnabled()) { logger.debug("SQL used for reading first page: [" + firstPageSql + "]"); } if (parameterValues != null && parameterValues.size() > 0) { if (this.queryProvider.isUsingNamedParameters()) { simpleJdbcTemplate.getNamedParameterJdbcOperations().query(firstPageSql, - getParameterMap(parameterValues, null), - new RowCallbackHandler() { - @SuppressWarnings("unchecked") - public void processRow(ResultSet rs) throws SQLException { - startAfterValue = rs.getObject(1); - results.add((T) rowMapper.mapRow(rs, results.size())); - } - }); + getParameterMap(parameterValues, null), rowCallback); } else { simpleJdbcTemplate.getJdbcOperations().query(firstPageSql, - getParameterList(parameterValues, null).toArray(), - new PagingRowCallbackHandler()); + getParameterList(parameterValues, null).toArray(), rowCallback); } } else { - simpleJdbcTemplate.getJdbcOperations().query(firstPageSql, - new PagingRowCallbackHandler()); + simpleJdbcTemplate.getJdbcOperations().query(firstPageSql, rowCallback); } } @@ -185,36 +190,41 @@ public class JdbcPagingItemReader extends AbstractPagingItemReader impleme } if (this.queryProvider.isUsingNamedParameters()) { simpleJdbcTemplate.getNamedParameterJdbcOperations().query(remainingPagesSql, - getParameterMap(parameterValues, startAfterValue), - new PagingRowCallbackHandler()); + getParameterMap(parameterValues, startAfterValue), rowCallback); } else { simpleJdbcTemplate.getJdbcOperations().query(remainingPagesSql, - getParameterList(parameterValues, startAfterValue).toArray(), - new PagingRowCallbackHandler()); + getParameterList(parameterValues, startAfterValue).toArray(), rowCallback); } } + if (results == null) { + results = new CopyOnWriteArrayList(); + } + else { + results.clear(); + } + results.addAll(rowCallback.getResults()); + } @Override protected void doJumpToPage(int itemIndex) { - if (page > 0) { + if (getPage() > 0) { String jumpToItemSql; - jumpToItemSql = queryProvider.generateJumpToItemQuery(itemIndex, pageSize); + jumpToItemSql = queryProvider.generateJumpToItemQuery(itemIndex, getPageSize()); if (logger.isDebugEnabled()) { logger.debug("SQL used for jumping: [" + jumpToItemSql + "]"); } - startAfterValue = simpleJdbcTemplate.getJdbcOperations().queryForObject(jumpToItemSql, - new RowMapper() { - public Object mapRow(ResultSet rs, int i) throws SQLException { - return rs.getObject(1); - } - }); + startAfterValue = simpleJdbcTemplate.getJdbcOperations().queryForObject(jumpToItemSql, new RowMapper() { + public Object mapRow(ResultSet rs, int i) throws SQLException { + return rs.getObject(1); + } + }); } @@ -251,6 +261,12 @@ public class JdbcPagingItemReader extends AbstractPagingItemReader impleme } private class PagingRowCallbackHandler implements RowCallbackHandler { + private final List results = new ArrayList(); + + public List getResults() { + return results; + } + @SuppressWarnings("unchecked") public void processRow(ResultSet rs) throws SQLException { startAfterValue = rs.getObject(1); diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/JpaPagingItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/JpaPagingItemReader.java index 43a7fe508..03cab065d 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/JpaPagingItemReader.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/JpaPagingItemReader.java @@ -18,6 +18,7 @@ package org.springframework.batch.item.database; import java.util.HashMap; import java.util.Map; +import java.util.concurrent.CopyOnWriteArrayList; import javax.persistence.EntityManager; import javax.persistence.EntityManagerFactory; @@ -29,38 +30,60 @@ import org.springframework.util.Assert; import org.springframework.util.ClassUtils; /** - * {@link org.springframework.batch.item.ItemReader} for reading database records built on top of JPA. - * - * It executes the JPQL {@link #setQueryString(String)} to retrieve requested data. The query is - * executed using paged requests of a size specified in {@link #setPageSize(int)}. Additional pages - * are requested when needed as {@link #read()} method is called, returning - * an object corresponding to current position. - * - * The performance of the paging depends on the JPA implementation and its use of database specific - * features to limit the number of returned rows. - * - * Setting a fairly large page size and using a commit interval that matches the page size should provide - * better performance. - * - * In order to reduce the memory usage for large results the persistence context is flushed and cleared - * after each page is read. This causes any entities read to be detached. If you make changes to the - * entities and want the changes persisted then you must explicitly merge the entities. - * - * The reader must be configured with an {@link javax.persistence.EntityManagerFactory}. All entity access - * is performed within a new transaction, independent of any existing Spring managed transactions. - * - * The implementation is *not* thread-safe. - * + *

+ * {@link org.springframework.batch.item.ItemReader} for reading database + * records built on top of JPA. + *

+ * + *

+ * It executes the JPQL {@link #setQueryString(String)} to retrieve requested + * data. The query is executed using paged requests of a size specified in + * {@link #setPageSize(int)}. Additional pages are requested when needed as + * {@link #read()} method is called, returning an object corresponding to + * current position. + *

+ * + *

+ * The performance of the paging depends on the JPA implementation and its use + * of database specific features to limit the number of returned rows. + *

+ * + *

+ * Setting a fairly large page size and using a commit interval that matches the + * page size should provide better performance. + *

+ * + *

+ * In order to reduce the memory usage for large results the persistence context + * is flushed and cleared after each page is read. This causes any entities read + * to be detached. If you make changes to the entities and want the changes + * persisted then you must explicitly merge the entities. + *

+ * + *

+ * The reader must be configured with an + * {@link javax.persistence.EntityManagerFactory}. All entity access is + * performed within a new transaction, independent of any existing Spring + * managed transactions. + *

+ * + *

+ * The implementation is thread-safe, but remember to use + * saveState=false if used in a multi-threaded client (no restart + * available). + *

+ * + * * @author Thomas Risberg * @since 2.0 */ public class JpaPagingItemReader extends AbstractPagingItemReader { private EntityManagerFactory entityManagerFactory; - + private EntityManager entityManager; - private final Map jpaPropertyMap = new HashMap(); + private final Map jpaPropertyMap = new HashMap(); private String queryString; @@ -76,8 +99,9 @@ public class JpaPagingItemReader extends AbstractPagingItemReader { /** * The parameter values to be used for the query execution. - * - * @param parameterValues the values keyed by the parameter named used in the query string. + * + * @param parameterValues the values keyed by the parameter named used in + * the query string. */ public void setParameterValues(Map parameterValues) { this.parameterValues = parameterValues; @@ -99,8 +123,7 @@ public class JpaPagingItemReader extends AbstractPagingItemReader { @Override protected void doOpen() throws Exception { super.doOpen(); - entityManager = - entityManagerFactory.createEntityManager(jpaPropertyMap); + entityManager = entityManagerFactory.createEntityManager(jpaPropertyMap); if (entityManager == null) { throw new DataAccessResourceFailureException("Unable to obtain an EntityManager"); } @@ -116,9 +139,8 @@ public class JpaPagingItemReader extends AbstractPagingItemReader { entityManager.flush(); entityManager.clear(); - Query query = entityManager.createQuery(queryString) - .setFirstResult(page * pageSize) - .setMaxResults(pageSize); + Query query = entityManager.createQuery(queryString).setFirstResult(getPage() * getPageSize()).setMaxResults( + getPageSize()); if (parameterValues != null) { for (Map.Entry me : parameterValues.entrySet()) { @@ -126,8 +148,14 @@ public class JpaPagingItemReader extends AbstractPagingItemReader { } } - results = query.getResultList(); - + if (results == null) { + results = new CopyOnWriteArrayList(); + } + else { + results.clear(); + } + results.addAll(query.getResultList()); + tx.commit(); } @@ -140,5 +168,5 @@ public class JpaPagingItemReader extends AbstractPagingItemReader { entityManager.close(); super.doClose(); } - + } diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/database/IbatisPagingItemReaderAsyncTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/database/IbatisPagingItemReaderAsyncTests.java new file mode 100644 index 000000000..7a486de68 --- /dev/null +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/database/IbatisPagingItemReaderAsyncTests.java @@ -0,0 +1,153 @@ +package org.springframework.batch.item.database; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; + +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.Callable; +import java.util.concurrent.CompletionService; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorCompletionService; +import java.util.concurrent.Executors; + +import javax.sql.DataSource; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.springframework.batch.item.sample.Foo; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.core.io.ClassPathResource; +import org.springframework.jdbc.core.simple.SimpleJdbcTemplate; +import org.springframework.orm.ibatis.SqlMapClientFactoryBean; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; +import org.springframework.test.jdbc.SimpleJdbcTestUtils; + +import com.ibatis.sqlmap.client.SqlMapClient; + +@RunWith(SpringJUnit4ClassRunner.class) +@ContextConfiguration(locations = "JdbcPagingItemReaderCommonTests-context.xml") +public class IbatisPagingItemReaderAsyncTests { + + /** + * The number of items to read + */ + private static final int ITEM_COUNT = 100; + + /** + * The number of threads to create + */ + private static final int THREAD_COUNT = 5; + + private static Log logger = LogFactory.getLog(IbatisPagingItemReaderAsyncTests.class); + + @Autowired + private DataSource dataSource; + + private int maxId; + + @Before + public void init() { + SimpleJdbcTemplate jdbcTemplate = new SimpleJdbcTemplate(dataSource); + maxId = jdbcTemplate.queryForInt("SELECT MAX(ID) from T_FOOS"); + for (int i = maxId + 1; i <= ITEM_COUNT; i++) { + jdbcTemplate.update("INSERT into T_FOOS (ID,NAME,VALUE) values (?, ?, ?)", i, "foo" + i, i); + } + assertEquals(ITEM_COUNT, SimpleJdbcTestUtils.countRowsInTable(jdbcTemplate, "T_FOOS")); + } + + @After + public void destroy() { + SimpleJdbcTemplate jdbcTemplate = new SimpleJdbcTemplate(dataSource); + jdbcTemplate.update("DELETE from T_FOOS where ID>?", maxId); + } + + @Test + public void testAsyncReader() throws Throwable { + List throwables = new ArrayList(); + int max = 10; + for (int i = 0; i < max; i++) { + try { + doTest(); + } + catch (Throwable e) { + throwables.add(e); + } + } + if (!throwables.isEmpty()) { + throw new IllegalStateException(String.format("Failed %d out of %d", throwables.size(), max), throwables + .get(0)); + } + } + + /** + * @throws Exception + * @throws InterruptedException + * @throws ExecutionException + */ + private void doTest() throws Exception, InterruptedException, ExecutionException { + final IbatisPagingItemReader reader = getItemReader(); + CompletionService> completionService = new ExecutorCompletionService>(Executors + .newFixedThreadPool(THREAD_COUNT)); + for (int i = 0; i < THREAD_COUNT; i++) { + completionService.submit(new Callable>() { + public List call() throws Exception { + List list = new ArrayList(); + Foo next = null; + do { + next = reader.read(); + Thread.sleep(10L); + logger.debug("Reading item: " + next); + if (next != null) { + list.add(next); + } + } while (next != null); + return list; + } + }); + } + int count = 0; + for (int i = 0; i < THREAD_COUNT; i++) { + List items = completionService.take().get(); + count += items.size(); + logger.debug("Finished items count: " + items.size()); + logger.debug("Finished items: " + items); + assertNotNull(items); + } + assertEquals(ITEM_COUNT, count); + reader.close(); + } + + private IbatisPagingItemReader getItemReader() throws Exception { + SqlMapClientFactoryBean factory = new SqlMapClientFactoryBean(); + factory.setConfigLocation(new ClassPathResource("ibatis-config.xml", getClass())); + factory.setDataSource(dataSource); + factory.afterPropertiesSet(); + SqlMapClient sqlMapClient = createSqlMapClient(); + + IbatisPagingItemReader reader = new IbatisPagingItemReader(); + reader.setQueryId("getPagedFoos"); + reader.setPageSize(2); + reader.setSqlMapClient(sqlMapClient); + reader.setSaveState(true); + + reader.afterPropertiesSet(); + + return reader; + } + + private SqlMapClient createSqlMapClient() throws Exception { + SqlMapClientFactoryBean factory = new SqlMapClientFactoryBean(); + factory.setConfigLocation(new ClassPathResource("ibatis-config.xml", getClass())); + factory.setDataSource(dataSource); + factory.afterPropertiesSet(); + return (SqlMapClient) factory.getObject(); + } + + +} diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/database/JdbcPagingItemReaderAsyncTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/database/JdbcPagingItemReaderAsyncTests.java new file mode 100644 index 000000000..125336be5 --- /dev/null +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/database/JdbcPagingItemReaderAsyncTests.java @@ -0,0 +1,156 @@ +package org.springframework.batch.item.database; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; + +import java.sql.ResultSet; +import java.sql.SQLException; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.Callable; +import java.util.concurrent.CompletionService; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorCompletionService; +import java.util.concurrent.Executors; + +import javax.sql.DataSource; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.springframework.batch.item.ItemReader; +import org.springframework.batch.item.database.support.HsqlPagingQueryProvider; +import org.springframework.batch.item.sample.Foo; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.jdbc.core.simple.ParameterizedRowMapper; +import org.springframework.jdbc.core.simple.SimpleJdbcTemplate; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; +import org.springframework.test.jdbc.SimpleJdbcTestUtils; + +@RunWith(SpringJUnit4ClassRunner.class) +@ContextConfiguration(locations = "JdbcPagingItemReaderCommonTests-context.xml") +public class JdbcPagingItemReaderAsyncTests { + + /** + * The page size + */ + private static final int PAGE_SIZE = 5; + + /** + * The number of items to read + */ + private static final int ITEM_COUNT = 100; + + /** + * The number of threads to create + */ + private static final int THREAD_COUNT = 5; + + private static Log logger = LogFactory.getLog(JdbcPagingItemReaderAsyncTests.class); + + @Autowired + private DataSource dataSource; + + private int maxId; + + @Before + public void init() { + SimpleJdbcTemplate jdbcTemplate = new SimpleJdbcTemplate(dataSource); + maxId = jdbcTemplate.queryForInt("SELECT MAX(ID) from T_FOOS"); + for (int i = maxId + 1; i <= ITEM_COUNT; i++) { + jdbcTemplate.update("INSERT into T_FOOS (ID,NAME,VALUE) values (?, ?, ?)", i, "foo" + i, i); + } + assertEquals(ITEM_COUNT, SimpleJdbcTestUtils.countRowsInTable(jdbcTemplate, "T_FOOS")); + } + + @After + public void destroy() { + SimpleJdbcTemplate jdbcTemplate = new SimpleJdbcTemplate(dataSource); + jdbcTemplate.update("DELETE from T_FOOS where ID>?", maxId); + } + + @Test + public void testAsyncReader() throws Throwable { + List throwables = new ArrayList(); + int max = 10; + for (int i = 0; i < max; i++) { + try { + doTest(); + } + catch (Throwable e) { + throwables.add(e); + } + } + if (!throwables.isEmpty()) { + throw new IllegalStateException(String.format("Failed %d out of %d", throwables.size(), max), throwables + .get(0)); + } + } + + /** + * @throws Exception + * @throws InterruptedException + * @throws ExecutionException + */ + private void doTest() throws Exception, InterruptedException, ExecutionException { + final ItemReader reader = getItemReader(); + CompletionService> completionService = new ExecutorCompletionService>(Executors + .newFixedThreadPool(THREAD_COUNT)); + for (int i = 0; i < THREAD_COUNT; i++) { + completionService.submit(new Callable>() { + public List call() throws Exception { + List list = new ArrayList(); + Foo next = null; + do { + next = reader.read(); + Thread.sleep(10L); + logger.debug("Reading item: " + next); + if (next != null) { + list.add(next); + } + } while (next != null); + return list; + } + }); + } + int count = 0; + for (int i = 0; i < THREAD_COUNT; i++) { + List items = completionService.take().get(); + count += items.size(); + logger.debug("Finished items count: " + items.size()); + logger.debug("Finished items: " + items); + assertNotNull(items); + } + assertEquals(ITEM_COUNT, count); + } + + protected ItemReader getItemReader() throws Exception { + + JdbcPagingItemReader reader = new JdbcPagingItemReader(); + reader.setDataSource(dataSource); + HsqlPagingQueryProvider queryProvider = new HsqlPagingQueryProvider(); + queryProvider.setSelectClause("select ID, NAME, VALUE"); + queryProvider.setFromClause("from T_FOOS"); + queryProvider.setSortKey("ID"); + reader.setQueryProvider(queryProvider); + reader.setRowMapper(new ParameterizedRowMapper() { + public Foo mapRow(ResultSet rs, int i) throws SQLException { + Foo foo = new Foo(); + foo.setId(rs.getInt(1)); + foo.setName(rs.getString(2)); + foo.setValue(rs.getInt(3)); + return foo; + } + }); + reader.setPageSize(PAGE_SIZE); + reader.afterPropertiesSet(); + reader.setSaveState(false); + + return reader; + } + +} diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/database/JpaPagingItemReaderAsyncTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/database/JpaPagingItemReaderAsyncTests.java new file mode 100644 index 000000000..006c05018 --- /dev/null +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/database/JpaPagingItemReaderAsyncTests.java @@ -0,0 +1,142 @@ +package org.springframework.batch.item.database; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; + +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.Callable; +import java.util.concurrent.CompletionService; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorCompletionService; +import java.util.concurrent.Executors; + +import javax.persistence.EntityManagerFactory; +import javax.sql.DataSource; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.springframework.batch.item.ExecutionContext; +import org.springframework.batch.item.sample.Foo; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.jdbc.core.simple.SimpleJdbcTemplate; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; +import org.springframework.test.jdbc.SimpleJdbcTestUtils; + +@RunWith(SpringJUnit4ClassRunner.class) +@ContextConfiguration(locations = "JpaPagingItemReaderCommonTests-context.xml") +public class JpaPagingItemReaderAsyncTests { + + /** + * The number of items to read + */ + private static final int ITEM_COUNT = 100; + + /** + * The number of threads to create + */ + private static final int THREAD_COUNT = 5; + + private static Log logger = LogFactory.getLog(JpaPagingItemReaderAsyncTests.class); + + @Autowired + private DataSource dataSource; + + @Autowired + private EntityManagerFactory entityManagerFactory; + + private int maxId; + + @Before + public void init() { + SimpleJdbcTemplate jdbcTemplate = new SimpleJdbcTemplate(dataSource); + maxId = jdbcTemplate.queryForInt("SELECT MAX(ID) from T_FOOS"); + for (int i = maxId + 1; i <= ITEM_COUNT; i++) { + jdbcTemplate.update("INSERT into T_FOOS (ID,NAME,VALUE) values (?, ?, ?)", i, "foo" + i, i); + } + assertEquals(ITEM_COUNT, SimpleJdbcTestUtils.countRowsInTable(jdbcTemplate, "T_FOOS")); + } + + @After + public void destroy() { + SimpleJdbcTemplate jdbcTemplate = new SimpleJdbcTemplate(dataSource); + jdbcTemplate.update("DELETE from T_FOOS where ID>?", maxId); + } + + @Test + public void testAsyncReader() throws Throwable { + List throwables = new ArrayList(); + int max = 10; + for (int i = 0; i < max; i++) { + try { + doTest(); + } + catch (Throwable e) { + throwables.add(e); + } + } + if (!throwables.isEmpty()) { + throw new IllegalStateException(String.format("Failed %d out of %d", throwables.size(), max), throwables + .get(0)); + } + } + + /** + * @throws Exception + * @throws InterruptedException + * @throws ExecutionException + */ + private void doTest() throws Exception, InterruptedException, ExecutionException { + final JpaPagingItemReader reader = getItemReader(); + CompletionService> completionService = new ExecutorCompletionService>(Executors + .newFixedThreadPool(THREAD_COUNT)); + for (int i = 0; i < THREAD_COUNT; i++) { + completionService.submit(new Callable>() { + public List call() throws Exception { + List list = new ArrayList(); + Foo next = null; + do { + next = reader.read(); + Thread.sleep(10L); + logger.debug("Reading item: " + next); + if (next != null) { + list.add(next); + } + } while (next != null); + return list; + } + }); + } + int count = 0; + for (int i = 0; i < THREAD_COUNT; i++) { + List items = completionService.take().get(); + count += items.size(); + logger.debug("Finished items count: " + items.size()); + logger.debug("Finished items: " + items); + assertNotNull(items); + } + assertEquals(ITEM_COUNT, count); + reader.close(); + } + + private JpaPagingItemReader getItemReader() throws Exception { + + String jpqlQuery = "select f from Foo f"; + + JpaPagingItemReader reader = new JpaPagingItemReader(); + reader.setQueryString(jpqlQuery); + reader.setEntityManagerFactory(entityManagerFactory); + reader.setPageSize(3); + reader.afterPropertiesSet(); + reader.setSaveState(false); + reader.open(new ExecutionContext()); + + return reader; + } + +}