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

BATCHADM-102: Change MessageChannelPartitionHandler to not require a replyChannl

Browse Source
This commit is contained in:
Dave Syer
2011-03-04 09:46:59 +00:00
committed by Michael Minella
parent 4706d61ec2
commit 57993c31af
7 changed files with 225 additions and 74 deletions
Download Patch File Download Diff File Expand all files Collapse all files

99
spring-batch-integration/src/main/java/org/springframework/batch/integration/partition/MessageChannelPartitionHandler.java
View File

@@ -4,6 +4,8 @@ import java.util.Collection;
import java.util.List;
import java.util.Set;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.partition.PartitionHandler;
@@ -14,23 +16,20 @@ import org.springframework.integration.MessageChannel;
import org.springframework.integration.annotation.Aggregator;
import org.springframework.integration.annotation.MessageEndpoint;
import org.springframework.integration.annotation.Payloads;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.core.MessagingOperations;
import org.springframework.integration.core.PollableChannel;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.util.Assert;
/**
* A {@link PartitionHandler} that uses {@link MessageChannel} instances to send
* instructions to remote workers and receive their responses. The
* {@link MessageChannel} provides a nice abstraction so that the location of
* the workers and the transport used to communicate with them can be changed at
* run time. The communication with the remote workers does not need to be
* transactional or have guaranteed delivery, so a local thread pool based
* implementation works as well as a remote web service or JMS implementation.
* If a remote worker fails or doesn't send a reply message, the job will fail
* and can be restarted to pick up missing messages and processing. The remote
* workers need access to the Spring Batch {@link JobRepository} so that the
* shared state across those restarts can be managed centrally.
* A {@link PartitionHandler} that uses {@link MessageChannel} instances to send instructions to remote workers and
* receive their responses. The {@link MessageChannel} provides a nice abstraction so that the location of the workers
* and the transport used to communicate with them can be changed at run time. The communication with the remote workers
* does not need to be transactional or have guaranteed delivery, so a local thread pool based implementation works as
* well as a remote web service or JMS implementation. If a remote worker fails or doesn't send a reply message, the job
* will fail and can be restarted to pick up missing messages and processing. The remote workers need access to the
* Spring Batch {@link JobRepository} so that the shared state across those restarts can be managed centrally.
*
* @author Dave Syer
*
@@ -38,31 +37,25 @@ import org.springframework.util.Assert;
@MessageEndpoint
public class MessageChannelPartitionHandler implements PartitionHandler {
private static Log logger = LogFactory.getLog(MessageChannelPartitionHandler.class);
private int gridSize = 1;
private MessagingOperations messagingGateway;
private String stepName;
private PollableChannel replyChannel;
public void afterPropertiesSet() throws Exception {
Assert.notNull(stepName, "A step name must be provided for the remote workers.");
Assert.state(messagingGateway != null, "The MessagingOperations must be set");
}
/**
* A pre-configured gateway for sending and receiving messages to the remote
* workers. Using this property allows a large degree of control over the
* timeouts and other properties of the send. It should have channels set up
* internally:
* <ul>
* <li>request channel capable of accepting {@link StepExecutionRequest}
* payloads</li>
* <li>reply channel that returns a list of {@link StepExecution} results</li>
* </ul>
* The timeout for the repoy should be set sufficiently long that the remote
* steps have time to complete.
* A pre-configured gateway for sending and receiving messages to the remote workers. Using this property allows a
* large degree of control over the timeouts and other properties of the send. It should have channels set up
* internally: <ul> <li>request channel capable of accepting {@link StepExecutionRequest} payloads</li> <li>reply
* channel that returns a list of {@link StepExecution} results</li> </ul> The timeout for the repoy should be set
* sufficiently long that the remote steps have time to complete.
*
* @param messagingGateway the {@link MessagingOperations} to set
*/
@@ -70,16 +63,10 @@ public class MessageChannelPartitionHandler implements PartitionHandler {
this.messagingGateway = messagingGateway;
}
public void setReplyChannel(PollableChannel replyChannel) {
this.replyChannel = replyChannel;
}
/**
* 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.
* 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
*/
@@ -88,14 +75,12 @@ public class MessageChannelPartitionHandler implements PartitionHandler {
}
/**
* The name of the {@link Step} that will be used to execute the partitioned
* {@link StepExecution}. This is a regular Spring Batch step, with all the
* business logic required to complete an execution based on the input
* parameters in its {@link StepExecution} context. The name will be
* translated into a {@link Step} instance by the remote worker.
* The name of the {@link Step} that will be used to execute the partitioned {@link StepExecution}. This is a
* regular Spring Batch step, with all the business logic required to complete an execution based on the input
* parameters in its {@link StepExecution} context. The name will be translated into a {@link Step} instance by the
* remote worker.
*
* @param stepName the name of the {@link Step} instance to execute business
* logic
* @param stepName the name of the {@link Step} instance to execute business logic
*/
public void setStepName(String stepName) {
this.stepName = stepName;
@@ -111,13 +96,10 @@ public class MessageChannelPartitionHandler implements PartitionHandler {
}
/**
* Sends {@link StepExecutionRequest} objects to the request channel of the
* {@link MessagingOperations}, and then receives the result back as a list of
* {@link StepExecution} on a reply channel. Use the
* {@link #aggregate(List)} method as an aggregator of the individual remote
* replies. The receive timeout needs to be set realistically in the
* {@link MessagingOperations} <b>and</b> the aggregator, so that there is a
* good chance of all work being done.
* Sends {@link StepExecutionRequest} objects to the request channel of the {@link MessagingOperations}, and then
* receives the result back as a list of {@link StepExecution} on a reply channel. Use the {@link #aggregate(List)}
* method as an aggregator of the individual remote replies. The receive timeout needs to be set realistically in
* the {@link MessagingOperations} <b>and</b> the aggregator, so that there is a good chance of all work being done.
*
* @see PartitionHandler#handle(StepExecutionSplitter, StepExecution)
*/
@@ -126,21 +108,32 @@ public class MessageChannelPartitionHandler implements PartitionHandler {
Set<StepExecution> split = stepExecutionSplitter.split(masterStepExecution, gridSize);
int count = 0;
PollableChannel replyChannel = new QueueChannel();
for (StepExecution stepExecution : split) {
messagingGateway.send(createMessage(count++, split.size(), new StepExecutionRequest(stepName, stepExecution
.getJobExecutionId(), stepExecution.getId())));
Message<StepExecutionRequest> request = createMessage(count++, split.size(), new StepExecutionRequest(
stepName, stepExecution.getJobExecutionId(), stepExecution.getId()), replyChannel);
if (logger.isDebugEnabled()) {
logger.debug("Sending request: " + request);
}
messagingGateway.send(request);
}
Message<Collection<StepExecution>> message = messagingGateway.receive(replyChannel);
if (logger.isDebugEnabled()) {
logger.debug("Received replies: " + message);
}
Collection<StepExecution> result = message.getPayload();
return result;
}
private Message<StepExecutionRequest> createMessage(int sequenceNumber, int sequenceSize,
StepExecutionRequest stepExecutionRequest) {
return MessageBuilder.withPayload(stepExecutionRequest).setSequenceNumber(sequenceNumber).setSequenceSize(
sequenceSize).setCorrelationId(
stepExecutionRequest.getJobExecutionId() + ":" + stepExecutionRequest.getStepName()).build();
StepExecutionRequest stepExecutionRequest, PollableChannel replyChannel) {
return MessageBuilder.withPayload(stepExecutionRequest).setSequenceNumber(sequenceNumber)
.setSequenceSize(sequenceSize)
.setCorrelationId(stepExecutionRequest.getJobExecutionId() + ":" + stepExecutionRequest.getStepName())
.setReplyChannel(replyChannel)
.build();
}
}