程序包 org.springframework.batch.integration.partition

类 MessageChannelPartitionHandler

  • 所有已实现的接口:
    PartitionHandler, org.springframework.beans.factory.InitializingBean
    @MessageEndpoint
public class MessageChannelPartitionHandler
extends java.lang.Object
implements PartitionHandler, org.springframework.beans.factory.InitializingBean
    A PartitionHandler that uses MessageChannel instances to send instructions to remote workers and receive their responses. The 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, the job will fail and can be restarted to pick up missing messages and processing. The remote workers need access to the Spring Batch JobRepository so that the shared state across those restarts can be managed centrally. While a MessageChannel is used for sending the requests to the workers, the worker's responses can be obtained in one of two ways:
    • A reply channel - Slaves will respond with messages that will be aggregated via this component.
    • Polling the job repository - Since the state of each slave is maintained independently within the job repository, we can poll the store to determine the state without the need of the slaves to formally respond.
    Note: The reply channel for this is instance based. Sharing this component across multiple step instances may result in the crossing of messages. It's recommended that this component be step or job scoped.
    作者:
    Dave Syer, Will Schipp, Michael Minella, Mahmoud Ben Hassine

    • 方法详细资料

      • afterPropertiesSet

        public void afterPropertiesSet()
                        throws java.lang.Exception
        指定者:
        afterPropertiesSet 在接口中 org.springframework.beans.factory.InitializingBean
        抛出:
        java.lang.Exception

      • setTimeout

        public void setTimeout​(long timeout)
        When using job repository polling, the time limit to wait.
        参数:
        timeout - milliseconds to wait, defaults to -1 (no timeout).

      • setJobExplorer

        public void setJobExplorer​(JobExplorer jobExplorer)
        JobExplorer to use to query the job repository. Either this or a DataSource is required when using job repository polling.
        参数:
        jobExplorer - JobExplorer to use for lookups

      • setPollInterval

        public void setPollInterval​(long pollInterval)
        How often to poll the job repository for the status of the slaves.
        参数:
        pollInterval - milliseconds between polls, defaults to 10000 (10 seconds).

      • setDataSource

        public void setDataSource​(javax.sql.DataSource dataSource)
        DataSource pointing to the job repository
        参数:
        dataSource - DataSource that points to the job repository's store

      • setMessagingOperations

        public void setMessagingOperations​(org.springframework.integration.core.MessagingTemplate messagingGateway)
        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: The timeout for the reply should be set sufficiently long that the remote steps have time to complete.
        参数:
        messagingGateway - the MessagingTemplate to set

      • setStepName

        public void setStepName​(java.lang.String stepName)
        The name of the Step that will be used to execute the partitioned 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 StepExecution context. The name will be translated into a Step instance by the remote worker.
        参数:
        stepName - the name of the Step instance to execute business logic

      • aggregate

        @Aggregator(sendPartialResultsOnExpiry="true")
public java.util.List<?> aggregate​(@Payloads
                                   java.util.List<?> messages)
        参数:
        messages - the messages to be aggregated
        返回:
        the list as it was passed in

      • setReplyChannel

        public void setReplyChannel​(org.springframework.messaging.PollableChannel replyChannel)

      • handle

        public java.util.Collection<StepExecution> handle​(StepExecutionSplitter stepExecutionSplitter,
                                                  StepExecution masterStepExecution)
                                           throws java.lang.Exception
        Sends StepExecutionRequest objects to the request channel of the MessagingTemplate, and then receives the result back as a list of StepExecution on a reply channel. Use the aggregate(List) method as an aggregator of the individual remote replies. The receive timeout needs to be set realistically in the MessagingTemplateand the aggregator, so that there is a good chance of all work being done.
        指定者:
        handle 在接口中 PartitionHandler
        参数:
        stepExecutionSplitter - a strategy for generating a collection of StepExecution instances
        masterStepExecution - the master step execution for the whole partition
        返回:
        a collection of completed StepExecution instances
        抛出:
        java.lang.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.
        另请参阅:
        PartitionHandler.handle(StepExecutionSplitter, StepExecution)