Spring/spring-batch
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

RESOLVED - issue BATCH-1389: Thread safety in JdbcPagingReader

Browse Source 
http://jira.springframework.org/browse/BATCH-1389
This commit is contained in:
dsyer
2009-09-04 13:19:10 +00:00
parent 0cc77cf6ca
commit 29fb16b522
7 changed files with 673 additions and 145 deletions
Download Patch File Download Diff File Expand all files Collapse all files

94
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/AbstractPagingItemReader.java
View File

@@ -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<T> extends AbstractItemCountingItemStreamItemReader<T> implements InitializingBean {
public abstract class AbstractPagingItemReader<T> extends AbstractItemCountingItemStreamItemReader<T> 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<T> results;
protected volatile List<T> 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<T> 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<T> 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<T> 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);
}
}

19
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/IbatisPagingItemReader.java
View File

@@ -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;
* </p>
*
* <p>
* The implementation is *not* thread-safe.
* The implementation is thread-safe, but remember to use
* <code>saveState=false</code> if used in a multi-threaded client (no restart
* available).
* </p>
*
* @author Thomas Risberg
@@ -115,10 +118,16 @@ public class IbatisPagingItemReader<T> extends AbstractPagingItemReader<T> {
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<T>();
}
else {
results.clear();
}
results.addAll(sqlMapClientTemplate.queryForList(queryId, parameters));
}
@Override

158
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/JdbcPagingItemReader.java
View File

@@ -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.
*
* <p>
* {@link org.springframework.batch.item.ItemReader} for reading database
* records using JDBC in a paging fashion.
* </p>
*
* <p>
* 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.
* </p>
*
* <p>
* The performance of the paging depends on the database specific features
* available to limit the number of returned rows.
* </p>
*
* <p>
* Setting a fairly large page size and using a commit interval that matches the
* page size should provide better performance.
* </p>
*
* <p>
* The implementation is thread-safe, but remember to use
* <code>saveState=false</code> if used in a multi-threaded client (no restart
* available).
* </p>
*
* @author Thomas Risberg
* @author Dave Syer
* @since 2.0
*/
public class JdbcPagingItemReader<T> extends AbstractPagingItemReader<T> implements InitializingBean {
@@ -103,20 +119,24 @@ public class JdbcPagingItemReader<T> extends AbstractPagingItemReader<T> 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<String, Object> parameterValues) {
this.parameterValues = parameterValues;
@@ -133,49 +153,34 @@ public class JdbcPagingItemReader<T> extends AbstractPagingItemReader<T> 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<T>();
}
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<T> extends AbstractPagingItemReader<T> 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<T>();
}
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<T> extends AbstractPagingItemReader<T> impleme
}
private class PagingRowCallbackHandler implements RowCallbackHandler {
private final List<T> results = new ArrayList<T>();
public List<T> getResults() {
return results;
}
@SuppressWarnings("unchecked")
public void processRow(ResultSet rs) throws SQLException {
startAfterValue = rs.getObject(1);

96
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/JpaPagingItemReader.java
View File

@@ -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.
*
* <p>
* {@link org.springframework.batch.item.ItemReader} for reading database
* records built on top of JPA.
* </p>
*
* <p>
* 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.
* </p>
*
* <p>
* The performance of the paging depends on the JPA implementation and its use
* of database specific features to limit the number of returned rows.
* </p>
*
* <p>
* Setting a fairly large page size and using a commit interval that matches the
* page size should provide better performance.
* </p>
*
* <p>
* 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.
* </p>
*
* <p>
* 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.
* </p>
*
* <p>
* The implementation is thread-safe, but remember to use
* <code>saveState=false</code> if used in a multi-threaded client (no restart
* available).
* </p>
*
*
* @author Thomas Risberg
* @since 2.0
*/
public class JpaPagingItemReader<T> extends AbstractPagingItemReader<T> {
private EntityManagerFactory entityManagerFactory;
private EntityManager entityManager;
private final Map<String,Object> jpaPropertyMap = new HashMap<String,Object>();
private final Map<String, Object> jpaPropertyMap = new HashMap<String, Object>();
private String queryString;
@@ -76,8 +99,9 @@ public class JpaPagingItemReader<T> extends AbstractPagingItemReader<T> {
/**
* 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<String, Object> parameterValues) {
this.parameterValues = parameterValues;
@@ -99,8 +123,7 @@ public class JpaPagingItemReader<T> extends AbstractPagingItemReader<T> {
@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<T> extends AbstractPagingItemReader<T> {
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<String, Object> me : parameterValues.entrySet()) {
@@ -126,8 +148,14 @@ public class JpaPagingItemReader<T> extends AbstractPagingItemReader<T> {
}
}
results = query.getResultList();
if (results == null) {
results = new CopyOnWriteArrayList<T>();
}
else {
results.clear();
}
results.addAll(query.getResultList());
tx.commit();
}
@@ -140,5 +168,5 @@ public class JpaPagingItemReader<T> extends AbstractPagingItemReader<T> {
entityManager.close();
super.doClose();
}
}