Spring Boot provides a lot of utility classes to perform batch jobs and schedule them at regular intervals. This helps developers focus on writing only business logic and hence saves our time and effort.
Spring Batch Framework offers two processing styles.
- TaskletStep Oriented
- Chunk Oriented
TaskletStep is used when either only reading or writing the data item is required.
Chunk is used when both reading and writing the data item is required.
In this example, I will explain how to configure batch jobs with Chunk Oriented Processing Model using spring boot.
Chunk Oriented Processing Model involves three components
1.ItemReader - An input operation, Used for providing the data. It reads the data that will be processed. 2.ItemProcessor - A processing operation, Used for item transformation. It processes input object and transforms to output object. 3.ItemWriter - An Output operation, Used to write the data transformed by ItemProcessor. The data items can be written to database, memory or outputstream.
One item is read from an ItemReader, handed to an ItemProcessor, and transformed into an output object. Once the number of items read equals the commit interval, the entire chunk is written out via the ItemWriter, and then the transaction is committed.
In this sample application, we will read the input data from a csv file and transform into rest service request and call a back-end rest service with 3 data items at a time.
Batch Job involves Job Configuration and Step Configuration.
- Step Configuration defines the reader, processor, writer and chunk items
- Job Configuration defines the incrementer, listener and flow Step that we defined above.
Step 1 : Create a batch configuration class as below.
BatchConfiguration
package com.kswaughs.config;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.JobExecutionListener;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.core.launch.support.RunIdIncrementer;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.file.FlatFileItemReader;
import org.springframework.batch.item.file.mapping.BeanWrapperFieldSetMapper;
import org.springframework.batch.item.file.mapping.DefaultLineMapper;
import org.springframework.batch.item.file.transform.DelimitedLineTokenizer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.io.ClassPathResource;
import com.kswaughs.beans.Order;
import com.kswaughs.beans.SvcReq;
import com.kswaughs.order.OrderSvcInvoker;
import com.kswaughs.processor.JobCompletionNotificationListener;
import com.kswaughs.processor.OrderItemProcessor;
@Configuration
@EnableBatchProcessing
public class BatchConfiguration {
@Autowired
public JobBuilderFactory jobBuilderFactory;
@Autowired
public StepBuilderFactory stepBuilderFactory;
@Bean
public Job processOrderJob() {
return jobBuilderFactory.get("processOrderJob")
.incrementer(new RunIdIncrementer())
.listener(listener())
.flow(orderStep())
.end()
.build();
}
@Bean
public Step orderStep() {
return stepBuilderFactory.get("orderStep")
.<Order, SvcReq> chunk(3)
.reader(reader())
.processor(processor())
.writer(writer())
.build();
}
@Bean
public FlatFileItemReader<Order> reader() {
FlatFileItemReader<Order> reader = new FlatFileItemReader<Order>();
reader.setResource(new ClassPathResource("PhoneData.csv"));
reader.setLineMapper(new DefaultLineMapper<Order>() {{
setLineTokenizer(new DelimitedLineTokenizer() {{
setNames(new String[] { "orderID", "orderName" });
}});
setFieldSetMapper(new BeanWrapperFieldSetMapper<Order>() {{
setTargetType(Order.class);
}});
}});
return reader;
}
@Bean
public OrderItemProcessor processor() {
return new OrderItemProcessor();
}
@Bean
public ItemWriter<SvcReq> writer() {
return new OrderSvcInvoker();
}
@Bean
public JobExecutionListener listener() {
return new JobCompletionNotificationListener();
}
}
Step 2: Below class is the processor that transforms into rest service request.
OrderItemProcessor
package com.kswaughs.processor;
import org.springframework.batch.item.ItemProcessor;
import com.kswaughs.beans.Order;
import com.kswaughs.beans.SvcReq;
public class OrderItemProcessor implements ItemProcessor<Order, SvcReq> {
@Override
public SvcReq process(final Order order) throws Exception {
final SvcReq svcReq = new SvcReq();
svcReq.setId(order.getOrderID());
svcReq.setName(order.getOrderName().toUpperCase());
System.out.println("Converting (" + order + ") into (" + svcReq + ")");
return svcReq;
}
}
Step 3: Below class is the Item Writer that calls the rest service.
OrderSvcInvoker
package com.kswaughs.order;
import java.util.List;
import org.springframework.batch.item.ItemWriter;
import org.springframework.http.ResponseEntity;
import org.springframework.web.client.RestTemplate;
import com.kswaughs.beans.SvcReq;
import com.kswaughs.beans.SvcResp;
public class OrderSvcInvoker implements ItemWriter<SvcReq> {
@Override
public void write(List<? extends SvcReq> svcReqs) throws Exception {
for (SvcReq svcReq : svcReqs) {
RestTemplate restTemplate = new RestTemplate();
ResponseEntity<SvcResp> respEntity = restTemplate
.postForEntity("http://localhost:8080/phone/order", svcReq, SvcResp.class);
SvcResp resp = respEntity.getBody();
System.out.println("calling web service:" + resp);
}
System.out.println("Processed items:" + svcReqs.size());
}
}
Step 4: Below class is the listener class that is executed when the job is finished.
JobCompletionNotificationListener
package com.kswaughs.processor;
import org.springframework.batch.core.BatchStatus;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.listener.JobExecutionListenerSupport;
public class JobCompletionNotificationListener extends JobExecutionListenerSupport {
@Override
public void afterJob(JobExecution jobExecution) {
if(jobExecution.getStatus() == BatchStatus.COMPLETED) {
System.out.println("BATCH JOB FINISHED SUCCESSFULLY");
}
}
}
Step 5: Below are the value bean classes for Item, rest request and rest response.
Order
package com.kswaughs.beans;
public class Order {
private String orderID;
private String orderName;
public String getOrderID() {
return orderID;
}
public void setOrderID(String orderID) {
this.orderID = orderID;
}
public String getOrderName() {
return orderName;
}
public void setOrderName(String orderName) {
this.orderName = orderName;
}
@Override
public String toString() {
StringBuilder builder = new StringBuilder();
builder.append("Order [orderID=");
builder.append(orderID);
builder.append(", orderName=");
builder.append(orderName);
builder.append("]");
return builder.toString();
}
}
SvcReq
package com.kswaughs.beans;
public class SvcReq {
private String id;
private String name;
public String getId() {
return id;
}
public void setId(String id) {
this.id = id;
}
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
@Override
public String toString() {
StringBuilder builder = new StringBuilder();
builder.append("SvcReq [id=");
builder.append(id);
builder.append(", name=");
builder.append(name);
builder.append("]");
return builder.toString();
}
}
SvcResp
package com.kswaughs.beans;
public class SvcResp {
private String id;
private String message;
public String getId() {
return id;
}
public void setId(String id) {
this.id = id;
}
public String getMessage() {
return message;
}
public void setMessage(String message) {
this.message = message;
}
@Override
public String toString() {
StringBuilder builder = new StringBuilder();
builder.append("SvcResp [id=");
builder.append(id);
builder.append(", message=");
builder.append(message);
builder.append("]");
return builder.toString();
}
}
Main class
BootApp
package com.kswaughs.config;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.ComponentScan;
@SpringBootApplication
@ComponentScan({ "com.kswaughs" })
public class BootApp {
public static void main(String[] args) {
SpringApplication.run(new Object[] { BootApp.class }, args);
}
}
Input csv file : PhoneData.csv
101,Apple IPhone 102,Samsung Galaxy Y 103,Moto E 104,Moto X 105,Yuphoria
Output:
2016-03-16 13:58:53.324 INFO 12876 --- [main] o.s.b.c.l.support.SimpleJobLauncher: Job:
[FlowJob: [name=processOrderJob]] launched with the following parameters: [{run.id=1}]
2016-03-16 13:58:53.343 INFO 12876 --- [main] o.s.batch.core.job.SimpleStepHandler: Executing step: [orderStep]
Converting (Order [orderID=101, orderName=Apple IPhone]) into (SvcReq [id=101, name=APPLE IPHONE])
Converting (Order [orderID=102, orderName=Samsung Galaxy Y]) into (SvcReq [id=102, name=SAMSUNG GALAXY Y])
Converting (Order [orderID=103, orderName=Moto E]) into (SvcReq [id=103, name=MOTO E])
calling web service:SvcResp [id=101, message=APPLE IPHONE Processed successfully]
calling web service:SvcResp [id=102, message=SAMSUNG GALAXY Y Processed successfully]
calling web service:SvcResp [id=103, message=MOTO E Processed successfully]
Processed items:3
Converting (Order [orderID=104, orderName=Moto X]) into (SvcReq [id=104, name=MOTO X])
Converting (Order [orderID=105, orderName=Yuphoria]) into (SvcReq [id=105, name=YUPHORIA])
calling web service:SvcResp [id=104, message=MOTO X Processed successfully]
calling web service:SvcResp [id=105, message=YUPHORIA Processed successfully]
Processed items:2
BATCH JOB FINISHED SUCCESSFULLY
2016-03-16 13:58:53.786 INFO 12876 --- [main] o.s.b.c.l.support.SimpleJobLauncher: Job:
[FlowJob: [name=processOrderJob]] completed with the following parameters: [{run.id=1}] and the following status: [COMPLETED]
Maven POM
<dependencies>
<!-- Spring framework related jars -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter</artifactId>
<version>1.3.0.RELEASE</version>
</dependency>
<!-- Batch related jars -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-batch</artifactId>
<version>1.2.7.RELEASE</version>
</dependency>
<!-- Rest service call related jars -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-rest</artifactId>
<version>1.3.0.RELEASE</version>
<exclusions>
<exclusion>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</exclusion>
</exclusions>
</dependency>
</dependencies>