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

BATCH-1537: use startAfterValue to make restart more efficient in JdbcPagingItemReader

Browse Source
This commit is contained in:
dsyer
2010-03-26 11:00:25 +00:00
parent 0c4162296f
commit 43b72a8150
3 changed files with 75 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -29,6 +29,7 @@ import java.util.concurrent.CopyOnWriteArrayList;
import javax.sql.DataSource;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.batch.item.ItemStreamException;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.core.RowCallbackHandler;
@@ -48,17 +49,16 @@ import org.springframework.util.ClassUtils;
* 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.
* to current position. On restart it uses the last sort key value to locate the
* first page to read (so it doesn't matter if the successfully processed itmes
* have been removed or modified).
* </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.
* 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.
* </p>
*
* <p>
@@ -74,6 +74,11 @@ import org.springframework.util.ClassUtils;
*/
public class JdbcPagingItemReader<T> extends AbstractPagingItemReader<T> implements InitializingBean {
/**
*
*/
private static final String START_AFTER_VALUE = "start.after";
public static final int VALUE_NOT_SET = -1;
private DataSource dataSource;
@@ -126,7 +131,9 @@ public class JdbcPagingItemReader<T> extends AbstractPagingItemReader<T> impleme
}
/**
* The row mapper implementation to be used by this reader
* The row mapper implementation to be used by this reader. The row mapper
* is used to convert result set rows into objects, which are then returned
* by the reader.
*
* @param rowMapper a
* {@link org.springframework.jdbc.core.simple.ParameterizedRowMapper}
@@ -217,9 +224,28 @@ public class JdbcPagingItemReader<T> extends AbstractPagingItemReader<T> impleme
}
@Override
protected void doJumpToPage(int itemIndex) {
public void update(ExecutionContext executionContext) throws ItemStreamException {
super.update(executionContext);
if (isSaveState() && startAfterValue != null) {
executionContext.put(getExecutionContextUserSupport().getKey(START_AFTER_VALUE), startAfterValue);
}
}
if (getPage() > 0) {
@Override
public void open(ExecutionContext executionContext) {
super.open(executionContext);
if (isSaveState()) {
startAfterValue = executionContext.get(getExecutionContextUserSupport().getKey(START_AFTER_VALUE));
}
}
@Override
protected void doJumpToPage(int itemIndex) {
/*
* Normally this would be false (the startAfterValue is enough
* information to restart from.
*/
if (startAfterValue == null && getPage() > 0) {
String jumpToItemSql;
jumpToItemSql = queryProvider.generateJumpToItemQuery(itemIndex, getPageSize());
@@ -243,7 +269,6 @@ public class JdbcPagingItemReader<T> extends AbstractPagingItemReader<T> impleme
}
}
}
private Map<String, Object> getParameterMap(Map<String, Object> values, Object sortKeyValue) {

72
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/AbstractItemCountingItemStreamItemReader.java
View File

@@ -34,8 +34,7 @@ import org.springframework.util.Assert;
*
* @author Robert Kasanicky
*/
public abstract class AbstractItemCountingItemStreamItemReader<T> implements
ItemReader<T>, ItemStream {
public abstract class AbstractItemCountingItemStreamItemReader<T> implements ItemReader<T>, ItemStream {
private static final String READ_COUNT = "read.count";
@@ -78,8 +77,7 @@ public abstract class AbstractItemCountingItemStreamItemReader<T> implements
}
}
public final T read() throws Exception, UnexpectedInputException,
ParseException {
public final T read() throws Exception, UnexpectedInputException, ParseException {
if (currentItemCount >= maxItemCount) {
return null;
}
@@ -99,8 +97,7 @@ public abstract class AbstractItemCountingItemStreamItemReader<T> implements
*
* @see #setName(String)
*
* @param count
* the value of the current item count
* @param count the value of the current item count
*/
public void setCurrentItemCount(int count) {
this.currentItemCount = count;
@@ -108,14 +105,14 @@ public abstract class AbstractItemCountingItemStreamItemReader<T> implements
/**
* The maximum index of the items to be read. If the
* {@link ExecutionContext} contains a key <code>[name].read.count.max</code>
* (where <code>[name]</code> is the name of this component) the value from
* the {@link ExecutionContext} will be used in preference.
* {@link ExecutionContext} contains a key
* <code>[name].read.count.max</code> (where <code>[name]</code> is the name
* of this component) the value from the {@link ExecutionContext} will be
* used in preference.
*
* @see #setName(String)
*
* @param count
* the value of the maximum item count
* @param count the value of the maximum item count
*/
public void setMaxItemCount(int count) {
this.maxItemCount = count;
@@ -125,35 +122,34 @@ public abstract class AbstractItemCountingItemStreamItemReader<T> implements
currentItemCount = 0;
try {
doClose();
} catch (Exception e) {
}
catch (Exception e) {
throw new ItemStreamException("Error while closing item reader", e);
}
}
public void open(ExecutionContext executionContext)
throws ItemStreamException {
public void open(ExecutionContext executionContext) throws ItemStreamException {
try {
doOpen();
} catch (Exception e) {
}
catch (Exception e) {
throw new ItemStreamException("Failed to initialize the reader", e);
}
if (executionContext.containsKey(ecSupport.getKey(READ_COUNT_MAX))) {
maxItemCount = executionContext.getInt(ecSupport
.getKey(READ_COUNT_MAX));
maxItemCount = executionContext.getInt(ecSupport.getKey(READ_COUNT_MAX));
}
if (executionContext.containsKey(ecSupport.getKey(READ_COUNT))) {
int itemCount = executionContext.getInt(ecSupport
.getKey(READ_COUNT));
int itemCount = executionContext.getInt(ecSupport.getKey(READ_COUNT));
if (itemCount < maxItemCount) {
try {
jumpToItem(itemCount);
} catch (Exception e) {
throw new ItemStreamException(
"Could not move to stored position on restart", e);
}
catch (Exception e) {
throw new ItemStreamException("Could not move to stored position on restart", e);
}
}
currentItemCount = itemCount;
@@ -162,28 +158,27 @@ public abstract class AbstractItemCountingItemStreamItemReader<T> implements
}
public void update(ExecutionContext executionContext)
throws ItemStreamException {
public void update(ExecutionContext executionContext) throws ItemStreamException {
if (saveState) {
Assert.notNull(executionContext,
"ExecutionContext must not be null");
executionContext.putInt(ecSupport.getKey(READ_COUNT),
currentItemCount);
Assert.notNull(executionContext, "ExecutionContext must not be null");
executionContext.putInt(ecSupport.getKey(READ_COUNT), currentItemCount);
if (maxItemCount < Integer.MAX_VALUE) {
executionContext.putInt(ecSupport.getKey(READ_COUNT_MAX),
maxItemCount);
executionContext.putInt(ecSupport.getKey(READ_COUNT_MAX), maxItemCount);
}
}
}
protected ExecutionContextUserSupport getExecutionContextUserSupport() {
return ecSupport;
}
/**
* The name of the component which will be used as a stem for keys in the
* {@link ExecutionContext}. Subclasses should provide a default value, e.g.
* the short form of the class name.
*
* @param name
* the name for the component
* @param name the name for the component
*/
public void setName(String name) {
ecSupport.setName(name);
@@ -193,12 +188,21 @@ public abstract class AbstractItemCountingItemStreamItemReader<T> implements
* Set the flag that determines whether to save internal data for
* {@link ExecutionContext}. Only switch this to false if you don't want to
* save any state from this stream, and you don't need it to be restartable.
* Always set it to false if the reader is being used in a concurrent
* environment.
*
* @param saveState
* flag value (default true).
* @param saveState flag value (default true).
*/
public void setSaveState(boolean saveState) {
this.saveState = saveState;
}
/**
* The flag that determines whether to save internal state for restarts.
* @return true if the flag was set
*/
public boolean isSaveState() {
return saveState;
}
}