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

RESOLVED - BATCH-699: Document thread safety

Browse Source 
Added javadoc mostly to ItemReaders and ItemWriters adn fixed the misleading javadoc of ItemReader interface.
This commit is contained in:
robokaso
2008-07-02 11:11:16 +00:00
parent 5cd1efd248
commit fa54917a0a
21 changed files with 169 additions and 92 deletions
Download Patch File Download Diff File Expand all files Collapse all files

56
spring-batch-samples/src/main/java/org/springframework/batch/sample/item/reader/StagingItemReader.java
View File

@@ -25,6 +25,10 @@ import org.springframework.jdbc.support.lob.LobHandler;
import org.springframework.transaction.support.TransactionSynchronizationManager;
import org.springframework.util.Assert;
/**
* Thread-safe database {@link ItemReader} implementing the process indicator
* pattern.
*/
public class StagingItemReader extends JdbcDaoSupport implements ItemStream, ItemReader, StepExecutionListener {
// Key for buffer in transaction synchronization manager
@@ -45,7 +49,8 @@ public class StagingItemReader extends JdbcDaoSupport implements ItemStream, Ite
/**
* Public setter for the {@link LobHandler}.
*
* @param lobHandler the {@link LobHandler} to set (defaults to {@link DefaultLobHandler}).
* @param lobHandler the {@link LobHandler} to set (defaults to
* {@link DefaultLobHandler}).
*/
public void setLobHandler(LobHandler lobHandler) {
this.lobHandler = lobHandler;
@@ -107,18 +112,18 @@ public class StagingItemReader extends JdbcDaoSupport implements ItemStream, Ite
return null;
}
Object result = getJdbcTemplate().queryForObject("SELECT VALUE FROM BATCH_STAGING WHERE ID=?",
new Object[] { id }, new RowMapper() {
public Object mapRow(ResultSet rs, int rowNum) throws SQLException {
byte[] blob = lobHandler.getBlobAsBytes(rs, 1);
return SerializationUtils.deserialize(blob);
}
});
new Object[] { id }, new RowMapper() {
public Object mapRow(ResultSet rs, int rowNum) throws SQLException {
byte[] blob = lobHandler.getBlobAsBytes(rs, 1);
return SerializationUtils.deserialize(blob);
}
});
// Update now - changes will rollback if there is a problem later.
int count = getJdbcTemplate().update("UPDATE BATCH_STAGING SET PROCESSED=? WHERE ID=? AND PROCESSED=?",
new Object[] { StagingItemWriter.DONE, id, StagingItemWriter.NEW });
new Object[] { StagingItemWriter.DONE, id, StagingItemWriter.NEW });
if (count != 1) {
throw new OptimisticLockingFailureException("The staging record with ID=" + id
+ " was updated concurrently when trying to mark as complete (updated " + count + " records.");
+ " was updated concurrently when trying to mark as complete (updated " + count + " records.");
}
return result;
}
@@ -133,14 +138,15 @@ public class StagingItemReader extends JdbcDaoSupport implements ItemStream, Ite
synchronized (lock) {
if (keys.hasNext()) {
Assert.state(TransactionSynchronizationManager.isActualTransactionActive(),
"Transaction not active for this thread.");
"Transaction not active for this thread.");
Long next = (Long) keys.next();
getBuffer().add(next);
key = next;
logger.debug("Retrieved key from list: " + key);
}
}
} else {
}
else {
logger.debug("Retrieved key from buffer: " + key);
}
return key;
@@ -192,9 +198,12 @@ public class StagingItemReader extends JdbcDaoSupport implements ItemStream, Ite
}
/**
* Mark is supported in a multi- as well as a single-threaded environment. The state backing the mark is a buffer,
* and access is synchronized, so multiple threads can be accommodated. Buffers are stored as transaction resources
* (using {@link TransactionSynchronizationManager#bindResource(Object, Object)}), so they are thread bound.
* Mark is supported in a multi- as well as a single-threaded environment.
* The state backing the mark is a buffer, and access is synchronized, so
* multiple threads can be accommodated. Buffers are stored as transaction
* resources (using
* {@link TransactionSynchronizationManager#bindResource(Object, Object)}),
* so they are thread bound.
*
* @see org.springframework.batch.item.ItemReader#mark()
*/
@@ -205,7 +214,9 @@ public class StagingItemReader extends JdbcDaoSupport implements ItemStream, Ite
/*
* (non-Javadoc)
*
* @see org.springframework.batch.item.ItemStream#reset(org.springframework.batch.item.ExecutionContext)
* @see
* org.springframework.batch.item.ItemStream#reset(org.springframework.batch
* .item.ExecutionContext)
*/
public void reset() {
getBuffer().rollback();
@@ -214,7 +225,9 @@ public class StagingItemReader extends JdbcDaoSupport implements ItemStream, Ite
/*
* (non-Javadoc)
*
* @see org.springframework.batch.item.ExecutionContextProvider#getExecutionContext()
* @see
* org.springframework.batch.item.ExecutionContextProvider#getExecutionContext
* ()
*/
public void update(ExecutionContext executionContext) {
}
@@ -222,7 +235,9 @@ public class StagingItemReader extends JdbcDaoSupport implements ItemStream, Ite
/*
* (non-Javadoc)
*
* @see org.springframework.batch.core.domain.StepListener#afterStep(StepExecution)
* @see
* org.springframework.batch.core.domain.StepListener#afterStep(StepExecution
* )
*/
public ExitStatus afterStep(StepExecution stepExecution) {
return null;
@@ -231,7 +246,8 @@ public class StagingItemReader extends JdbcDaoSupport implements ItemStream, Ite
/*
* (non-Javadoc)
*
* @see org.springframework.batch.core.domain.StepListener#beforeStep(org.springframework.batch.core.domain.StepExecution)
* @seeorg.springframework.batch.core.domain.StepListener#beforeStep(org.
* springframework.batch.core.domain.StepExecution)
*/
public void beforeStep(StepExecution stepExecution) {
this.stepExecution = stepExecution;
@@ -240,7 +256,9 @@ public class StagingItemReader extends JdbcDaoSupport implements ItemStream, Ite
/*
* (non-Javadoc)
*
* @see org.springframework.batch.core.domain.StepListener#onErrorInStep(java.lang.Throwable)
* @see
* org.springframework.batch.core.domain.StepListener#onErrorInStep(java
* .lang.Throwable)
*/
public ExitStatus onErrorInStep(StepExecution stepExecution, Throwable e) {
return null;

45
spring-batch-samples/src/main/java/org/springframework/batch/sample/item/writer/StagingItemWriter.java
View File

@@ -19,6 +19,9 @@ import org.springframework.jdbc.support.lob.LobHandler;
import org.springframework.util.Assert;
import org.springframework.util.ClassUtils;
/**
* Database {@link ItemWriter} implementing the process indicator pattern.
*/
public class StagingItemWriter extends JdbcDaoSupport implements StepExecutionListener, ItemWriter {
public static final String NEW = "N";
@@ -71,16 +74,15 @@ public class StagingItemWriter extends JdbcDaoSupport implements StepExecutionLi
final long id = incrementer.nextLongValue();
final long jobId = stepExecution.getJobExecution().getJobId().longValue();
final byte[] blob = SerializationUtils.serialize((Serializable) data);
getJdbcTemplate()
.update("INSERT into BATCH_STAGING (ID, JOB_ID, VALUE, PROCESSED) values (?,?,?,?)",
new PreparedStatementSetter() {
public void setValues(PreparedStatement ps) throws SQLException {
ps.setLong(1, id);
ps.setLong(2, jobId);
lobHandler.getLobCreator().setBlobAsBytes(ps, 3, blob);
ps.setString(4, NEW);
}
getJdbcTemplate().update("INSERT into BATCH_STAGING (ID, JOB_ID, VALUE, PROCESSED) values (?,?,?,?)",
new PreparedStatementSetter() {
public void setValues(PreparedStatement ps) throws SQLException {
ps.setLong(1, id);
ps.setLong(2, jobId);
lobHandler.getLobCreator().setBlobAsBytes(ps, 3, blob);
ps.setString(4, NEW);
}
});
}
@@ -90,22 +92,33 @@ public class StagingItemWriter extends JdbcDaoSupport implements StepExecutionLi
public void flush() throws FlushFailedException {
}
/* (non-Javadoc)
* @see org.springframework.batch.core.domain.StepListener#afterStep(StepExecution)
/*
* (non-Javadoc)
*
* @see
* org.springframework.batch.core.domain.StepListener#afterStep(StepExecution
* )
*/
public ExitStatus afterStep(StepExecution stepExecution) {
return null;
}
/* (non-Javadoc)
* @see org.springframework.batch.core.domain.StepListener#beforeStep(org.springframework.batch.core.domain.StepExecution)
/*
* (non-Javadoc)
*
* @seeorg.springframework.batch.core.domain.StepListener#beforeStep(org.
* springframework.batch.core.domain.StepExecution)
*/
public void beforeStep(StepExecution stepExecution) {
this.stepExecution = stepExecution;
}
/* (non-Javadoc)
* @see org.springframework.batch.core.domain.StepListener#onErrorInStep(java.lang.Throwable)
/*
* (non-Javadoc)
*
* @see
* org.springframework.batch.core.domain.StepListener#onErrorInStep(java
* .lang.Throwable)
*/
public ExitStatus onErrorInStep(StepExecution stepExecution, Throwable e) {
return null;