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

Merge remote-tracking branch 'snicoll/BATCH-1912'

Browse Source
This commit is contained in:
Michael Minella
2012-11-28 14:14:04 -06:00
parent 09bf67859d 182419162a
commit 064f39feb8
2 changed files with 136 additions and 61 deletions
Download Patch File Download Diff File Expand all files Collapse all files

84
spring-batch-core/src/main/java/org/springframework/batch/core/partition/support/AbstractPartitionHandler.java Normal file
View File

@@ -0,0 +1,84 @@
/*
* Copyright 2006-2007 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.batch.core.partition.support;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.partition.PartitionHandler;
import org.springframework.batch.core.partition.StepExecutionSplitter;
import java.util.Collection;
import java.util.Set;
/**
* Base {@link PartitionHandler} implementation providing common base
* features. Subclasses are expected to implement only the
* {@link #doHandle(org.springframework.batch.core.StepExecution, java.util.Set)}
* method which returns with the result of the execution(s) or an exception if
* the step failed to process.
*
* @author Sebastien Gerard
* @author Dave Syer
*/
public abstract class AbstractPartitionHandler implements PartitionHandler {
private int gridSize = 1;
/**
* Executes the specified {@link StepExecution} instances and returns an updated
* view of them. Throws an {@link Exception} if anything goes wrong.
*
* @param masterStepExecution the whole partition execution
* @param partitionStepExecutions the {@link StepExecution} instances to execute
* @return an updated view of these completed {@link StepExecution} instances
* @throws Exception if anything goes wrong. This allows implementations to
* be liberal and rely on the caller to translate an exception into a step
* failure as necessary.
*/
protected abstract Set<StepExecution> doHandle(StepExecution masterStepExecution,
Set<StepExecution> partitionStepExecutions) throws Exception;
public Collection<StepExecution> handle(final StepExecutionSplitter stepSplitter,
final StepExecution masterStepExecution) throws Exception {
final Set<StepExecution> stepExecutions = stepSplitter.split(masterStepExecution, gridSize);
return doHandle(masterStepExecution, stepExecutions);
}
/**
* Returns the number of step executions.
*
* @return the number of step executions
*/
public int getGridSize() {
return gridSize;
}
/**
* Passed to the {@link StepExecutionSplitter} in the
* {@link #handle(StepExecutionSplitter, StepExecution)} method, instructing
* it how many {@link StepExecution} instances are required, ideally. The
* {@link StepExecutionSplitter} is allowed to ignore the grid size in the
* case of a restart, since the input data partitions must be preserved.
*
* @param gridSize the number of step executions that will be created
*/
public void setGridSize(int gridSize) {
this.gridSize = gridSize;
}
}

113
spring-batch-core/src/main/java/org/springframework/batch/core/partition/support/TaskExecutorPartitionHandler.java
View File

@@ -16,8 +16,6 @@
package org.springframework.batch.core.partition.support;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.Callable;
@@ -29,9 +27,9 @@ import org.springframework.batch.core.ExitStatus;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.partition.PartitionHandler;
import org.springframework.batch.core.partition.StepExecutionSplitter;
import org.springframework.batch.core.step.StepHolder;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Required;
import org.springframework.core.task.SyncTaskExecutor;
import org.springframework.core.task.TaskExecutor;
import org.springframework.core.task.TaskRejectedException;
@@ -42,13 +40,14 @@ import org.springframework.util.Assert;
* partitioned {@link Step} locally in multiple threads. This can be an
* effective approach for scaling batch steps that are IO intensive, like
* directory and filesystem scanning and copying.
*
* <p/>
* By default, the thread pool is synchronous.
*
* @author Sebastien Gerard
* @author Dave Syer
* @since 2.0
*/
public class TaskExecutorPartitionHandler implements PartitionHandler, StepHolder, InitializingBean {
private int gridSize = 1;
public class TaskExecutorPartitionHandler extends AbstractPartitionHandler implements StepHolder, InitializingBean {
private TaskExecutor taskExecutor = new SyncTaskExecutor();
@@ -57,19 +56,6 @@ public class TaskExecutorPartitionHandler implements PartitionHandler, StepHolde
public void afterPropertiesSet() throws Exception {
}
/**
* Passed to the {@link StepExecutionSplitter} in the
* {@link #handle(StepExecutionSplitter, StepExecution)} method, instructing
* it how many {@link StepExecution} instances are required, ideally. The
* {@link StepExecutionSplitter} is allowed to ignore the grid size in the
* case of a restart, since the input data partitions must be preserved.
*
* @param gridSize the number of step executions that will be created
*/
public void setGridSize(int gridSize) {
this.gridSize = gridSize;
}
/**
* Setter for the {@link TaskExecutor} that is used to farm out step
* executions to multiple threads.
@@ -87,6 +73,7 @@ public class TaskExecutorPartitionHandler implements PartitionHandler, StepHolde
*
* @param step the {@link Step} instance to use to execute business logic
*/
@Required
public void setStep(Step step) {
this.step = step;
}
@@ -101,51 +88,55 @@ public class TaskExecutorPartitionHandler implements PartitionHandler, StepHolde
return this.step;
}
/**
* @see PartitionHandler#handle(StepExecutionSplitter, StepExecution)
*/
public Collection<StepExecution> handle(StepExecutionSplitter stepExecutionSplitter,
StepExecution masterStepExecution) throws Exception {
@Override
protected Set<StepExecution> doHandle(StepExecution masterStepExecution,
Set<StepExecution> partitionStepExecutions) throws Exception {
Assert.notNull(step, "A Step must be provided.");
final Set<Future<StepExecution>> tasks = new HashSet<Future<StepExecution>>(getGridSize());
final Set<StepExecution> result = new HashSet<StepExecution>();
Assert.notNull(step, "A Step must be provided.");
Set<Future<StepExecution>> tasks = new HashSet<Future<StepExecution>>(gridSize);
for (final StepExecution stepExecution : partitionStepExecutions) {
final FutureTask<StepExecution> task = createTask(step, stepExecution);
Collection<StepExecution> result = new ArrayList<StepExecution>();
try {
taskExecutor.execute(task);
tasks.add(task);
} catch (TaskRejectedException e) {
// couldn't execute one of the tasks
ExitStatus exitStatus = ExitStatus.FAILED
.addExitDescription("TaskExecutor rejected the task for this step.");
/*
* Set the status in case the caller is tracking it through the
* JobExecution.
*/
stepExecution.setStatus(BatchStatus.FAILED);
stepExecution.setExitStatus(exitStatus);
result.add(stepExecution);
}
}
for (final StepExecution stepExecution : stepExecutionSplitter.split(masterStepExecution, gridSize)) {
final FutureTask<StepExecution> task = new FutureTask<StepExecution>(new Callable<StepExecution>() {
public StepExecution call() throws Exception {
step.execute(stepExecution);
return stepExecution;
}
});
try {
taskExecutor.execute(task);
tasks.add(task);
}
catch (TaskRejectedException e) {
// couldn't execute one of the tasks
ExitStatus exitStatus = ExitStatus.FAILED
.addExitDescription("TaskExecutor rejected the task for this step.");
/*
* Set the status in case the caller is tracking it through the
* JobExecution.
*/
stepExecution.setStatus(BatchStatus.FAILED);
stepExecution.setExitStatus(exitStatus);
result.add(stepExecution);
}
}
for (Future<StepExecution> task : tasks) {
result.add(task.get());
}
return result;
for (Future<StepExecution> task : tasks) {
result.add(task.get());
}
return result;
}
/**
* Creates the task executing the given step in the context of the given execution.
*
* @param step the step to execute
* @param stepExecution the given execution
* @return the task executing the given step
*/
protected FutureTask<StepExecution> createTask(final Step step,
final StepExecution stepExecution) {
return new FutureTask<StepExecution>(new Callable<StepExecution>() {
public StepExecution call() throws Exception {
step.execute(stepExecution);
return stepExecution;
}
});
}
}