Converted adoc to use both XML and Java config
This commit is contained in:
Glenn Renfro
committed by
Michael Minella
Michael Minella
parent
f61679b55f
commit
31c7e8f83f
@@ -6,6 +6,8 @@
|
||||
|
||||
== Spring Batch Integration
|
||||
|
||||
include::toggle.adoc[]
|
||||
|
||||
[[spring-batch-integration-introduction]]
|
||||
|
||||
=== Spring Batch Integration Introduction
|
||||
@@ -38,10 +40,10 @@ provide methods to distribute workloads over a number of workers.
|
||||
|
||||
This section covers the following key concepts:
|
||||
|
||||
[role="xmlContent"]
|
||||
* <<spring-batch-integration.adoc#namespace-support,Namespace Support>>
|
||||
|
||||
|
||||
|
||||
[[continue-section-list]]
|
||||
* <<spring-batch-integration.adoc#launching-batch-jobs-through-messages,Launching Batch Jobs through Messages>>
|
||||
|
||||
|
||||
@@ -59,7 +61,7 @@ Batch Process Execution>>
|
||||
|
||||
|
||||
[[namespace-support]]
|
||||
|
||||
[role="xmlContent"]
|
||||
==== Namespace Support
|
||||
|
||||
Since Spring Batch Integration 1.3, dedicated XML Namespace
|
||||
@@ -265,7 +267,8 @@ launch the job via the __Job Launching Gateway__, and then log the output of the
|
||||
`JobExecution` with the
|
||||
`logging-channel-adapter`.
|
||||
|
||||
[source, xml]
|
||||
.XML Configuration
|
||||
[source, xml, role="xmlContent"]
|
||||
----
|
||||
<int:channel id="inboundFileChannel"/>
|
||||
<int:channel id="outboundJobRequestChannel"/>
|
||||
@@ -292,6 +295,38 @@ launch the job via the __Job Launching Gateway__, and then log the output of the
|
||||
<int:logging-channel-adapter channel="jobLaunchReplyChannel"/>
|
||||
----
|
||||
|
||||
.Java Configuration
|
||||
[source, java, role="javaContent"]
|
||||
----
|
||||
@Bean
|
||||
public FileMessageToJobRequest fileMessageToJobRequest() {
|
||||
FileMessageToJobRequest fileMessageToJobRequest = new FileMessageToJobRequest();
|
||||
fileMessageToJobRequest.setFileParameterName("input.file.name");
|
||||
fileMessageToJobRequest.setJob(personJob());
|
||||
return fileMessageToJobRequest;
|
||||
}
|
||||
|
||||
@Bean
|
||||
public JobLaunchingGateway jobLaunchingGateway() {
|
||||
SimpleJobLauncher simpleJobLauncher = new SimpleJobLauncher();
|
||||
simpleJobLauncher.setJobRepository(jobRepository);
|
||||
simpleJobLauncher.setTaskExecutor(new SyncTaskExecutor());
|
||||
JobLaunchingGateway jobLaunchingGateway = new JobLaunchingGateway(simpleJobLauncher);
|
||||
|
||||
return jobLaunchingGateway;
|
||||
}
|
||||
|
||||
@Bean
|
||||
public IntegrationFlow integrationFlow(JobLaunchingGateway jobLaunchingGateway) {
|
||||
return IntegrationFlows.from(Files.inboundAdapter(new File("/tmp/myfiles")).
|
||||
filter(new SimplePatternFileListFilter("*.csv")),
|
||||
c -> c.poller(Pollers.fixedRate(1000).maxMessagesPerPoll(1))).
|
||||
handle(fileMessageToJobRequest()).
|
||||
handle(jobLaunchingGateway).
|
||||
log(LoggingHandler.Level.WARN, "headers.id + ': ' + payload").
|
||||
get();
|
||||
}
|
||||
----
|
||||
|
||||
|
||||
[[example-itemreader-configuration]]
|
||||
@@ -303,7 +338,8 @@ configure our Spring Batch
|
||||
`ItemReader` (for example) to use the files found at the location defined
|
||||
by the job parameter called "input.file.name", as shown in the following bean configuration:
|
||||
|
||||
[source, xml]
|
||||
.XML Configuration
|
||||
[source, xml, role="xmlContent"]
|
||||
----
|
||||
<bean id="itemReader" class="org.springframework.batch.item.file.FlatFileItemReader"
|
||||
scope="step">
|
||||
@@ -312,6 +348,19 @@ by the job parameter called "input.file.name", as shown in the following bean co
|
||||
</bean>
|
||||
----
|
||||
|
||||
.Java Configuration
|
||||
[source, java, role="javaContent"]
|
||||
----
|
||||
@Bean
|
||||
@StepScope
|
||||
public ItemReader sampleReader(@Value("#{jobParameters[input.file.name]}") String resource) {
|
||||
...
|
||||
FlatFileItemReader flatFileItemReader = new FlatFileItemReader();
|
||||
flatFileItemReader.setResource(new FileSystemResource(resource));
|
||||
...
|
||||
return flatFileItemReader;
|
||||
}
|
||||
----
|
||||
|
||||
The main points of interest in the preceding example are injecting the value of
|
||||
`#{jobParameters['input.file.name']}`
|
||||
@@ -364,7 +413,9 @@ When this `Gateway` is receiving messages from a
|
||||
`PollableChannel`, you must either provide
|
||||
a global default `Poller` or provide a `Poller` sub-element to the
|
||||
`Job Launching Gateway`, as shown in the following example:
|
||||
[source, xml]
|
||||
|
||||
.XML Configuration
|
||||
[source, xml, role="xmlContent"]
|
||||
----
|
||||
<batch-int:job-launching-gateway request-channel="queueChannel"
|
||||
reply-channel="replyChannel" job-launcher="jobLauncher">
|
||||
@@ -372,6 +423,18 @@ a global default `Poller` or provide a `Poller` sub-element to the
|
||||
</batch-int:job-launching-gateway>
|
||||
----
|
||||
|
||||
.Java Configuration
|
||||
[source, java, role="javaContent"]
|
||||
----
|
||||
@Bean
|
||||
@ServiceActivator(inputChannel = "queueChannel", poller = @Poller(fixedRate="1000"))
|
||||
public JobLaunchingGateway sampleJobLaunchingGateway() {
|
||||
JobLaunchingGateway jobLaunchingGateway = new JobLaunchingGateway(jobLauncher());
|
||||
jobLaunchingGateway.setOutputChannel(replyChannel());
|
||||
return jobLaunchingGateway;
|
||||
}
|
||||
----
|
||||
|
||||
[[providing-feedback-with-informational-messages]]
|
||||
|
||||
==== Providing Feedback with Informational Messages
|
||||
@@ -428,7 +491,8 @@ message to a `Gateway` for a
|
||||
|
||||
First, create the notification integration beans:
|
||||
|
||||
[source, xml]
|
||||
.XML Configuration
|
||||
[source, xml, role="xmlContent"]
|
||||
----
|
||||
<int:channel id="stepExecutionsChannel"/>
|
||||
|
||||
@@ -439,10 +503,31 @@ First, create the notification integration beans:
|
||||
<int:logging-channel-adapter channel="stepExecutionsChannel"/>
|
||||
----
|
||||
|
||||
.Java Configuration
|
||||
[source, java, role="javaContent"]
|
||||
----
|
||||
@Bean
|
||||
@ServiceActivator(inputChannel = "stepExecutionsChannel")
|
||||
public LoggingHandler loggingHandler() {
|
||||
LoggingHandler adapter = new LoggingHandler(LoggingHandler.Level.WARN);
|
||||
adapter.setLoggerName("TEST_LOGGER");
|
||||
adapter.setLogExpressionString("headers.id + ': ' + payload");
|
||||
return adapter;
|
||||
}
|
||||
|
||||
@MessagingGateway(name = "notificationExecutionsListener", defaultRequestChannel = "stepExecutionsChannel")
|
||||
public interface NotificationExecutionListener extends StepExecutionListener {}
|
||||
----
|
||||
|
||||
[role="javaContent"]
|
||||
NOTE: You will need to add the `@IntegrationComponentScan` annotation to your configuration.
|
||||
|
||||
[[message-gateway-entry-list]]
|
||||
|
||||
Second, modify your job to add a step-level listener:
|
||||
|
||||
[source, xml]
|
||||
.XML Configuration
|
||||
[source, xml, role="xmlContent"]
|
||||
----
|
||||
<job id="importPayments">
|
||||
<step id="step1">
|
||||
@@ -457,6 +542,18 @@ Second, modify your job to add a step-level listener:
|
||||
</job>
|
||||
----
|
||||
|
||||
.Java Configuration
|
||||
[source, java, role="javaContent"]
|
||||
----
|
||||
public Job importPaymentsJob() {
|
||||
return jobBuilderFactory.get("importPayments")
|
||||
.start(stepBuilderFactory.get("step1")
|
||||
.chunk(200)
|
||||
.listener(notificationExecutionsListener())
|
||||
...
|
||||
}
|
||||
----
|
||||
|
||||
[[asynchronous-processors]]
|
||||
|
||||
==== Asynchronous Processors
|
||||
@@ -482,8 +579,8 @@ writes back the chunk as soon as all the results become available.
|
||||
The following example shows how to configuration the `AsyncItemProcessor`:
|
||||
|
||||
|
||||
|
||||
[source, xml]
|
||||
.XML Configuration
|
||||
[source, xml, role="xmlContent"]
|
||||
----
|
||||
<bean id="processor"
|
||||
class="org.springframework.batch.integration.async.AsyncItemProcessor">
|
||||
@@ -496,6 +593,17 @@ The following example shows how to configuration the `AsyncItemProcessor`:
|
||||
</bean>
|
||||
----
|
||||
|
||||
.Java Configuration
|
||||
[source, java, role="javaContent"]
|
||||
----
|
||||
@Bean
|
||||
public AsyncItemProcessor processor(ItemProcessor itemProcessor, TaskExecutor taskExecutor) {
|
||||
AsyncItemProcessor asyncItemProcessor = new AsyncItemProcessor();
|
||||
asyncItemProcessor.setTaskExecutor(taskExecutor);
|
||||
asyncItemProcessor.setDelegate(itemProcessor);
|
||||
return asyncItemProcessor;
|
||||
}
|
||||
----
|
||||
|
||||
The `delegate` property refers
|
||||
to your `ItemProcessor` bean, and
|
||||
@@ -504,7 +612,8 @@ refers to the `TaskExecutor` of your choice.
|
||||
|
||||
The following example shows how to configure the `AsyncItemWriter`:
|
||||
|
||||
[source, xml]
|
||||
.XML Configuration
|
||||
[source, xml, role="xmlContent"]
|
||||
----
|
||||
<bean id="itemWriter"
|
||||
class="org.springframework.batch.integration.async.AsyncItemWriter">
|
||||
@@ -514,6 +623,16 @@ The following example shows how to configure the `AsyncItemWriter`:
|
||||
</bean>
|
||||
----
|
||||
|
||||
.Java Configuration
|
||||
[source, java, role="javaContent"]
|
||||
----
|
||||
@Bean
|
||||
public AsyncItemWriter processor(ItemWriter itemWriter) {
|
||||
AsyncItemWriter asyncItemWriter = new AsyncItemWriter();
|
||||
asyncItemWriter.setDelegate(itemWriter);
|
||||
return asyncItemWriter;
|
||||
}
|
||||
----
|
||||
|
||||
Again, the `delegate` property is
|
||||
actually a reference to your `ItemWriter` bean.
|
||||
@@ -569,7 +688,8 @@ external systems for processing.
|
||||
A simple job with a step to be remotely chunked might have a
|
||||
configuration similar to the following:
|
||||
|
||||
[source, xml]
|
||||
.XML Configuration
|
||||
[source, xml, role="xmlContent"]
|
||||
----
|
||||
<job id="personJob">
|
||||
<step id="step1">
|
||||
@@ -581,6 +701,20 @@ configuration similar to the following:
|
||||
</job>
|
||||
----
|
||||
|
||||
.Java Configuration
|
||||
[source, java, role="javaContent"]
|
||||
----
|
||||
public Job chunkJob(ItemReader itemReader) {
|
||||
return jobBuilderFactory.get("personJob")
|
||||
.start(stepBuilderFactory.get("step1")
|
||||
.chunk(200)
|
||||
.reader(itemReader())
|
||||
.writer(itemWriter())
|
||||
.build())
|
||||
.build();
|
||||
}
|
||||
----
|
||||
|
||||
The `ItemReader` reference points to the bean you want
|
||||
to use for reading data on the master. The `ItemWriter` reference
|
||||
points to a special `ItemWriter`
|
||||
@@ -591,7 +725,8 @@ following configuration provides a basic master setup. You
|
||||
should check any additional component properties, such as
|
||||
throttle limits and so on, when implementing your use case.
|
||||
|
||||
[source, xml]
|
||||
.XML Configuration
|
||||
[source, xml, role="xmlContent"]
|
||||
----
|
||||
<bean id="connectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
|
||||
<property name="brokerURL" value="tcp://localhost:61616"/>
|
||||
@@ -627,6 +762,69 @@ throttle limits and so on, when implementing your use case.
|
||||
channel="replies"/>
|
||||
----
|
||||
|
||||
.Java Configuration
|
||||
[source, java, role="javaContent"]
|
||||
----
|
||||
@Bean
|
||||
public org.apache.activemq.ActiveMQConnectionFactory connectionFactory() {
|
||||
ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory();
|
||||
factory.setBrokerURL("tcp://localhost:61616");
|
||||
return factory;
|
||||
}
|
||||
|
||||
@Bean
|
||||
public DirectChannel requests() {
|
||||
return new DirectChannel();
|
||||
}
|
||||
|
||||
@Bean
|
||||
public IntegrationFlow jmsOutboundFlow() {
|
||||
return IntegrationFlows.from("requests")
|
||||
.handle(Jms.outboundGateway(connectionFactory())
|
||||
.requestDestination("requests"))
|
||||
.get();
|
||||
}
|
||||
|
||||
@Bean
|
||||
public MessagingTemplate messagingTemplate() {
|
||||
MessagingTemplate template = new MessagingTemplate();
|
||||
template.setDefaultChannel(requests());
|
||||
template.setReceiveTimeout(2000);
|
||||
return template;
|
||||
}
|
||||
|
||||
@Bean
|
||||
@StepScope()
|
||||
public ChunkMessageChannelItemWriter itemWriter() {
|
||||
ChunkMessageChannelItemWriter chunkMessageChannelItemWriter = new ChunkMessageChannelItemWriter();
|
||||
chunkMessageChannelItemWriter.setMessagingOperations(messagingTemplate());
|
||||
chunkMessageChannelItemWriter.setReplyChannel(replies());
|
||||
return chunkMessageChannelItemWriter;
|
||||
}
|
||||
|
||||
@Bean
|
||||
public RemoteChunkHandlerFactoryBean chunkHandler() {
|
||||
RemoteChunkHandlerFactoryBean remoteChunkHandlerFactoryBean = new RemoteChunkHandlerFactoryBean();
|
||||
remoteChunkHandlerFactoryBean.setChunkWriter(itemWriter());
|
||||
remoteChunkHandlerFactoryBean.setStep(step1());
|
||||
return remoteChunkHandlerFactoryBean;
|
||||
}
|
||||
|
||||
@Bean
|
||||
public QueueChannel replies() {
|
||||
return new QueueChannel();
|
||||
}
|
||||
|
||||
@Bean
|
||||
public IntegrationFlow jmsReplies() {
|
||||
return IntegrationFlows
|
||||
.from(Jms.messageDrivenChannelAdapter(connectionFactory())
|
||||
.configureListenerContainer(c -> c.subscriptionDurable(false))
|
||||
.destination("replies"))
|
||||
.channel(replies())
|
||||
.get();
|
||||
}
|
||||
----
|
||||
|
||||
The preceding configuration provides us with a number of beans. We
|
||||
configure our messaging middleware using ActiveMQ and the
|
||||
@@ -639,8 +837,8 @@ configured middleware.
|
||||
Now we can move on to the slave configuration, as shown in the following example:
|
||||
|
||||
|
||||
|
||||
[source, xml]
|
||||
.XML Configuration
|
||||
[source, xml, role="xmlContent"]
|
||||
----
|
||||
<bean id="connectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
|
||||
<property name="brokerURL" value="tcp://localhost:61616"/>
|
||||
@@ -679,6 +877,60 @@ Now we can move on to the slave configuration, as shown in the following example
|
||||
</bean>
|
||||
----
|
||||
|
||||
.Java Configuration
|
||||
[source, java, role="javaContent"]
|
||||
----
|
||||
@Bean
|
||||
public org.apache.activemq.ActiveMQConnectionFactory connectionFactory() {
|
||||
ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory();
|
||||
factory.setBrokerURL("tcp://localhost:61616");
|
||||
return factory;
|
||||
}
|
||||
|
||||
@Bean
|
||||
public DirectChannel requests() {
|
||||
return new DirectChannel();
|
||||
}
|
||||
@Bean
|
||||
public DirectChannel replies() {
|
||||
return new DirectChannel();
|
||||
}
|
||||
|
||||
@Bean
|
||||
public IntegrationFlow jmsIn() {
|
||||
return IntegrationFlows
|
||||
.from(Jms.messageDrivenChannelAdapter(connectionFactory())
|
||||
.configureListenerContainer(c -> c.subscriptionDurable(false))
|
||||
.destination("requests"))
|
||||
.channel(requests())
|
||||
.get();
|
||||
}
|
||||
|
||||
@Bean
|
||||
public IntegrationFlow outgoingReplies() {
|
||||
return IntegrationFlows.from("replies")
|
||||
.handle(Jms.outboundGateway(connectionFactory())
|
||||
.requestDestination("replies"))
|
||||
.get();
|
||||
}
|
||||
|
||||
@Bean
|
||||
@ServiceActivator(inputChannel = "requests")
|
||||
public AggregatorFactoryBean serviceActivator() throws Exception{
|
||||
AggregatorFactoryBean aggregatorFactoryBean = new AggregatorFactoryBean();
|
||||
aggregatorFactoryBean.setProcessorBean(chunkProcessorChunkHandler());
|
||||
aggregatorFactoryBean.setOutputChannel(replies());
|
||||
...
|
||||
return aggregatorFactoryBean;
|
||||
}
|
||||
|
||||
@Bean
|
||||
public ChunkProcessorChunkHandler chunkProcessorChunkHandler() {
|
||||
ChunkProcessorChunkHandler chunkProcessorChunkHandler = new ChunkProcessorChunkHandler();
|
||||
chunkProcessorChunkHandler.setChunkProcessor(new SimpleChunkProcessor(personItemProcessor(), personItemWriter()));
|
||||
return chunkProcessorChunkHandler;
|
||||
}
|
||||
----
|
||||
|
||||
Most of these configuration items should look familiar from the
|
||||
master configuration. Slaves do not need access to
|
||||
@@ -744,8 +996,8 @@ the `MessageChannelPartitionHandler` and JMS
|
||||
configuration:
|
||||
|
||||
|
||||
|
||||
[source, xml]
|
||||
.XML Configuration
|
||||
[source, xml, role="xmlContent"]
|
||||
----
|
||||
<bean id="partitionHandler"
|
||||
class="org.springframework.batch.integration.partition.MessageChannelPartitionHandler">
|
||||
@@ -796,9 +1048,111 @@ configuration:
|
||||
class="org.springframework.batch.integration.partition.BeanFactoryStepLocator" />
|
||||
----
|
||||
|
||||
.Java Configuration
|
||||
[source, java, role="javaContent"]
|
||||
----
|
||||
@Bean
|
||||
public PartitionHandler partitionHandler() {
|
||||
MessageChannelPartitionHandler partitionHandler = new MessageChannelPartitionHandler();
|
||||
partitionHandler.setStepName("step1");
|
||||
partitionHandler.setGridSize(3);
|
||||
partitionHandler.setReplyChannel(outboundReplies());
|
||||
MessagingTemplate template = new MessagingTemplate();
|
||||
template.setDefaultChannel(outboundRequests());
|
||||
template.setReceiveTimeout(100000);
|
||||
partitionHandler.setMessagingOperations(template);
|
||||
return partitionHandler;
|
||||
}
|
||||
|
||||
@Bean
|
||||
public DirectChannel outboundRequests() {
|
||||
return new DirectChannel();
|
||||
}
|
||||
|
||||
@Bean
|
||||
public IntegrationFlow outboundJmsRequests() {
|
||||
return IntegrationFlows.from("outboundRequests")
|
||||
.handle(Jms.outboundGateway(connectionFactory())
|
||||
.requestDestination("requestsQueue"))
|
||||
.get();
|
||||
}
|
||||
|
||||
@Bean
|
||||
public DirectChannel inboundRequests() {
|
||||
return new DirectChannel();
|
||||
}
|
||||
|
||||
public IntegrationFlow inboundJmsRequests() {
|
||||
return IntegrationFlows
|
||||
.from(Jms.messageDrivenChannelAdapter(connectionFactory())
|
||||
.configureListenerContainer(c -> c.subscriptionDurable(false))
|
||||
.destination("requestsQueue"))
|
||||
.channel(inboundRequests())
|
||||
.get();
|
||||
}
|
||||
|
||||
@Bean
|
||||
public StepExecutionRequestHandler stepExecutionRequestHandler() {
|
||||
StepExecutionRequestHandler stepExecutionRequestHandler = new StepExecutionRequestHandler();
|
||||
stepExecutionRequestHandler.setJobExplorer(jobExplorer);
|
||||
stepExecutionRequestHandler.setStepLocator(stepLocator());
|
||||
return stepExecutionRequestHandler;
|
||||
}
|
||||
|
||||
@Bean
|
||||
@ServiceActivator(inputChannel = "inboundRequests", outputChannel = "outboundStaging")
|
||||
public StepExecutionRequestHandler serviceActivator() throws Exception {
|
||||
return stepExecutionRequestHandler();
|
||||
}
|
||||
|
||||
@Bean
|
||||
public DirectChannel outboundStaging() {
|
||||
return new DirectChannel();
|
||||
}
|
||||
|
||||
@Bean
|
||||
public IntegrationFlow outboundJmsStaging() {
|
||||
return IntegrationFlows.from("outboundStaging")
|
||||
.handle(Jms.outboundGateway(connectionFactory())
|
||||
.requestDestination("stagingQueue"))
|
||||
.get();
|
||||
}
|
||||
|
||||
@Bean
|
||||
public DirectChannel inboundStaging() {
|
||||
return new DirectChannel();
|
||||
}
|
||||
|
||||
@Bean
|
||||
public IntegrationFlow inboundJmsStaging() {
|
||||
return IntegrationFlows
|
||||
.from(Jms.messageDrivenChannelAdapter(connectionFactory())
|
||||
.configureListenerContainer(c -> c.subscriptionDurable(false))
|
||||
.destination("stagingQueue"))
|
||||
.channel(inboundStaging())
|
||||
.get();
|
||||
}
|
||||
|
||||
@Bean
|
||||
@ServiceActivator(inputChannel = "inboundStaging")
|
||||
public AggregatorFactoryBean partitioningMessageHandler() throws Exception {
|
||||
AggregatorFactoryBean aggregatorFactoryBean = new AggregatorFactoryBean();
|
||||
aggregatorFactoryBean.setProcessorBean(partitionHandler());
|
||||
aggregatorFactoryBean.setOutputChannel(outboundReplies());
|
||||
...
|
||||
return aggregatorFactoryBean;
|
||||
}
|
||||
|
||||
@Bean
|
||||
public QueueChannel outboundReplies() {
|
||||
return new QueueChannel();
|
||||
}
|
||||
----
|
||||
|
||||
You must also ensure that the partition `handler` attribute maps to the `partitionHandler` bean, as shown in the following example:
|
||||
|
||||
[source, xml]
|
||||
.XML Configuration
|
||||
[source, xml, role="xmlContent"]
|
||||
----
|
||||
<job id="personJob">
|
||||
<step id="step1.master">
|
||||
@@ -807,3 +1161,15 @@ You must also ensure that the partition `handler` attribute maps to the `partiti
|
||||
</step>
|
||||
</job>
|
||||
----
|
||||
|
||||
.Java Configuration
|
||||
[source, java, role="javaContent"]
|
||||
----
|
||||
public Job personJob() {
|
||||
return jobBuilderFactory.get("personJob")
|
||||
.start(stepBuilderFactory.get("step1.master")
|
||||
.partitioner(partitionHandler())
|
||||
.build())
|
||||
.build();
|
||||
}
|
||||
----
|
||||
|
||||
Reference in New Issue
Block a user