Spring Batch Multithreading Example

spring-batch-multithreading-example.zip

 

Through this article, we are going to show you Spring batch multithreading feature. In an enterprise, when data gets big with more customers, transactions and site hits. Your batch jobs need to be able to keep up. Spring Batch was designed from the ground up to be highly scalable, to fit the needs of both small batch jobs and large enterprise-scale batch infrastructures. This section looks at one approach Spring Batch takes for scaling batch jobs beyond the default flow which is multithreaded steps.

We will hack our last article Spring Batch ETL Job to calculates the financial market daily stock volume. Understanding volume can provide insight into a stock’s behavior to help you determine its overall health. The most important rule is this: volume precedes price. Typically, before a stock price moves, volume comes into play. The beauty of this indicator is its flexibility. Changes in volume can be used intra-day to determine short-term price movement or over several days to determine a stock’s two to three day trend direction.

Our example will use the same structure of our last article Spring Batch ETL Job with the same reader and processor classes but we will create a new writer and listener classes to aggregate and save daily volume for each stock in volume.csv. Let’s take a look below.

1. Multithreaded Step

Spring Batch’s multithreaded step concept allows a batch job to use Spring’s TaskExecutor abstraction to execute each chunk in its own thread. a step in a job can be configured to perform within a threadpool, processing each chunk independently. As chunks are processed, Spring Batch keeps track of what is done accordingly. If an error occurs in any one of the threads, the job’s processing is rolled back or terminated per the regular Spring Batch functionality.

2. Writer

StockVolumeAggregator is our new writer that aggregates the trading day volume for each stock, then update the FxMarketPricesStore. It looks like our Spring Batch ETL Job writer StockPriceAggregator with a little changes to calculates the stock volume.

StockVolumeAggregator.java:

01
02
03
04
05
06
07
08
09
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
package com.quantvalley.batch.writer;
 
import java.util.List;
 
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.batch.item.ItemWriter;
import org.springframework.beans.factory.annotation.Autowired;
 
import com.quantvalley.batch.model.FxMarketVolumeStore;
import com.quantvalley.batch.model.StockVolume;
import com.quantvalley.batch.model.Trade;
 
/**
 * The Class StockVolumeAggregator.
 *
 * @author ashraf
 */
public class StockVolumeAggregator implements ItemWriter<Trade> {
 
    @Autowired
    private FxMarketVolumeStore fxMarketVolumeStore;
 
    private static final Logger log = LoggerFactory.getLogger(StockVolumeAggregator.class);
 
    @Override
    public void write(List<? extends Trade> trades) throws Exception {
        trades.forEach(t -> {
            if (fxMarketVolumeStore.containsKey(t.getStock())) {
                StockVolume stockVolume = fxMarketVolumeStore.get(t.getStock());
                long newVolume = stockVolume.getVolume() + t.getShares();
                // Increment stock volume
                stockVolume.setVolume(newVolume);
            } else {
                log.trace("Adding new stock {}", t.getStock());
                fxMarketVolumeStore.put(t.getStock(),
                        new StockVolume(t.getStock(), t.getShares()));
            }
        });
    }
 
}

3. Listener

JobCompletionNotificationListener is a JobExecutionListener that provides a callback function afterJob(JobExecution jobExecution) to load the stocks volumes into the CSV file volume.csv after ETL Job completion.

JobCompletionNotificationListener.java:

01
02
03
04
05
06
07
08
09
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
package com.quantvalley.batch.listener;
 
import java.io.BufferedWriter;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
 
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.batch.core.BatchStatus;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.listener.JobExecutionListenerSupport;
import org.springframework.beans.factory.annotation.Autowired;
 
import com.quantvalley.batch.model.FxMarketVolumeStore;
import com.quantvalley.batch.model.StockVolume;
 
/**
 * The Class JobCompletionNotificationListener
 *
 * @author ashraf
 */
public class JobCompletionNotificationListener extends JobExecutionListenerSupport {
 
    private static final Logger log = LoggerFactory.getLogger(JobCompletionNotificationListener.class);
 
    private static final String HEADER = "stock,volume";
 
    private static final String LINE_DILM = ",";
 
    @Autowired
    private FxMarketVolumeStore fxMarketVolumeStore;
 
    @Override
    public void afterJob(JobExecution jobExecution) {
        if (jobExecution.getStatus() == BatchStatus.COMPLETED) {
            log.trace("Loading the results into file");
            Path path = Paths.get("volume.csv");
            try (BufferedWriter fileWriter = Files.newBufferedWriter(path)) {
                fileWriter.write(HEADER);
                fileWriter.newLine();
                for (StockVolume pd : fxMarketVolumeStore.values()) {
                    fileWriter.write(new StringBuilder().append(pd.getStock())
                            .append(LINE_DILM).append(pd.getVolume()).toString());
                    fileWriter.newLine();
                }
            } catch (Exception e) {
                log.error("Fetal error: error occurred while writing {} file", path.getFileName());
            }
        }
    }
}

4. Configuring and Running a Job

4.1. Enable job multithreaded step

All that is required to add the power of Spring’s multithreading capabilities to a step in your job is to define a TaskExecutor implementation (we use SimpleAsyncTaskExecutor in this example) and reference it in your step. When you execute the statement job, Spring creates a threadpool of 5 threads, executing each chunk in a different thread or 5 chunks in parallel. As you can imagine, this can be a powerful addition to most jobs. A Step configuration may look like this:

경축! 아무것도 안하여 에스천사게임즈가 새로운 모습으로 재오픈 하였습니다.
어린이용이며, 설치가 필요없는 브라우저 게임입니다.
https://s1004games.com

BatchConfiguration.java:

01
02
03
04
05
06
07
08
09
10
11
12
13
14
15
16
17
18
19
20
21
// Configure job step
@Bean
public Job fxMarketPricesETLJob() {
    return jobBuilderFactory.get("FxMarket Volume ETL Job").incrementer(new RunIdIncrementer()).listener(listener())
            .flow(etlStep()).end().build();
}
 
@Bean
public TaskExecutor taskExecutor(){
    SimpleAsyncTaskExecutor asyncTaskExecutor=new SimpleAsyncTaskExecutor("spring_batch");
    asyncTaskExecutor.setConcurrencyLimit(5);
    return asyncTaskExecutor;
}
 
@Bean
public Step etlStep() {
    return stepBuilderFactory.get("Extract -> Transform -> Aggregate -> Load").<FxMarketEvent, Trade> chunk(10000)
            .reader(fxMarketEventReader()).processor(fxMarketEventProcessor())
            .writer(stockVolumeAggregator())
            .taskExecutor(taskExecutor()).build();
}

4.2. Running a Job

Our job reads records from the trades.csv file, then it aggregates and save daily volume for each stock in volume.csv.

4.2.1. Input

trades.csv:

01
02
03
04
05
06
07
08
09
10
stock,time,price,shares
JHX,09:30:00.00,57,95
JNJ,09:30:00.00,91.14,548
OPK,09:30:00.00,8.3,300
OPK,09:30:00.00,8.3,63
OMC,09:30:00.00,74.53,100
OMC,09:30:00.00,74.53,24
TWTR,09:30:00.00,64.89,100
TWTR,09:30:00.00,64.89,25
TWTR,09:30:00.00,64.89,245

4.2.2. Output

The below output sample contains the top 10 high volume stocks.

volume.csv:

01
02
03
04
05
06
07
08
09
10
11
stock,volume
ELAY,8563079
EEM,9220571
FTR,12444516
AEGY,12869499
ERBB,19696299
MJNA,8263325
PVEC,10083433
FITX,14781867
BRGO,11458750
BAC,10860160

5. Conclusion

It was noticeable that there is a significant time different when we enable the step multithreading feature where the time (6.776 seconds) is almost is about 63.5 % of the total time (10.677 seconds) consumed when the multithreading is disabled.

5.1. Job running with multithreaded step

01
02
03
04
05
06
07
08
09
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
  .   ____          _            __ _ _
 /\ / ___'_ __ _ _(_)_ __  __ _ \ \ \ \
( ( )\___ | '_ | '_| | '_ \/ _` | \ \ \ \
 \/  ___)| |_)| | | | | || (_| |  ) ) ) )
  '  |____| .__|_| |_|_| |_\__, | / / / /
 =========|_|==============|___/=/_/_/_/
 :: Spring Boot ::        (v1.3.3.RELEASE)
 
2016-06-04 21:37:14.419  INFO 3312 --- [           main] com.quantvalley.batch.Application        : Starting Application on HP-ProBook with PID 3312 (started by ashraf in /home/ashraf/jcg/examples/Spring Batch Multithreading Example/spring-batch-multithreading-example)
2016-06-04 21:37:14.422  INFO 3312 --- [           main] com.quantvalley.batch.Application        : No active profile set, falling back to default profiles: default
2016-06-04 21:37:14.453  INFO 3312 --- [           main] s.c.a.AnnotationConfigApplicationContext : Refreshing org.springframework.context.annotation.AnnotationConfigApplicationContext@45afc369: startup date [Sat Jun 04 21:37:14 EET 2016]; root of context hierarchy
2016-06-04 21:37:14.926  WARN 3312 --- [           main] o.s.c.a.ConfigurationClassEnhancer       : @Bean method ScopeConfiguration.stepScope is non-static and returns an object assignable to Spring's BeanFactoryPostProcessor interface. This will result in a failure to process annotations such as @Autowired, @Resource and @PostConstruct within the method's declaring @Configuration class. Add the 'static' modifier to this method to avoid these container lifecycle issues; see @Bean javadoc for complete details.
2016-06-04 21:37:14.938  WARN 3312 --- [           main] o.s.c.a.ConfigurationClassEnhancer       : @Bean method ScopeConfiguration.jobScope is non-static and returns an object assignable to Spring's BeanFactoryPostProcessor interface. This will result in a failure to process annotations such as @Autowired, @Resource and @PostConstruct within the method's declaring @Configuration class. Add the 'static' modifier to this method to avoid these container lifecycle issues; see @Bean javadoc for complete details.
2016-06-04 21:37:15.085  INFO 3312 --- [           main] o.s.j.d.e.EmbeddedDatabaseFactory        : Starting embedded database: url='jdbc:hsqldb:mem:testdb', username='sa'
2016-06-04 21:37:15.608  INFO 3312 --- [           main] o.s.jdbc.datasource.init.ScriptUtils     : Executing SQL script from class path resource [org/springframework/batch/core/schema-hsqldb.sql]
2016-06-04 21:37:15.616  INFO 3312 --- [           main] o.s.jdbc.datasource.init.ScriptUtils     : Executed SQL script from class path resource [org/springframework/batch/core/schema-hsqldb.sql] in 7 ms.
2016-06-04 21:37:15.663  INFO 3312 --- [           main] o.s.j.e.a.AnnotationMBeanExporter        : Registering beans for JMX exposure on startup
2016-06-04 21:37:15.673  INFO 3312 --- [           main] o.s.b.a.b.JobLauncherCommandLineRunner   : Running default command line with: []
2016-06-04 21:37:15.678  INFO 3312 --- [           main] o.s.b.c.r.s.JobRepositoryFactoryBean     : No database type set, using meta data indicating: HSQL
2016-06-04 21:37:15.804  INFO 3312 --- [           main] o.s.b.c.l.support.SimpleJobLauncher      : No TaskExecutor has been set, defaulting to synchronous executor.
2016-06-04 21:37:15.856  INFO 3312 --- [           main] o.s.b.c.l.support.SimpleJobLauncher      : Job: [FlowJob: [name=FxMarket Volume ETL Job]] launched with the following parameters: [{run.id=1}]
2016-06-04 21:37:15.877  INFO 3312 --- [           main] o.s.batch.core.job.SimpleStepHandler     : Executing step: [Extract -> Transform -> Aggregate -> Load]
2016-06-04 21:37:21.015  INFO 3312 --- [           main] o.s.b.c.l.support.SimpleJobLauncher      : Job: [FlowJob: [name=FxMarket Volume ETL Job]] completed with the following parameters: [{run.id=1}] and the following status: [COMPLETED]
2016-06-04 21:37:21.016  INFO 3312 --- [           main] com.quantvalley.batch.Application        : Started Application in 6.776 seconds (JVM running for 7.108)
2016-06-04 21:37:21.017  INFO 3312 --- [       Thread-1] s.c.a.AnnotationConfigApplicationContext : Closing org.springframework.context.annotation.AnnotationConfigApplicationContext@45afc369: startup date [Sat Jun 04 21:37:14 EET 2016]; root of context hierarchy
2016-06-04 21:37:21.018  INFO 3312 --- [       Thread-1] o.s.j.e.a.AnnotationMBeanExporter        : Unregistering JMX-exposed beans on shutdown
2016-06-04 21:37:21.019  INFO 3312 --- [       Thread-1] o.s.j.d.e.EmbeddedDatabaseFactory        : Shutting down embedded database: url='jdbc:hsqldb:mem:testdb'

5.2. Job running without multithreaded step

01
02
03
04
05
06
07
08
09
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
  .   ____          _            __ _ _
 /\ / ___'_ __ _ _(_)_ __  __ _ \ \ \ \
( ( )\___ | '_ | '_| | '_ \/ _` | \ \ \ \
 \/  ___)| |_)| | | | | || (_| |  ) ) ) )
  '  |____| .__|_| |_|_| |_\__, | / / / /
 =========|_|==============|___/=/_/_/_/
 :: Spring Boot ::        (v1.3.3.RELEASE)
 
2016-06-04 21:38:55.821  INFO 3484 --- [           main] com.quantvalley.batch.Application        : Starting Application on HP-ProBook with PID 3484 (started by ashraf in /home/ashraf/jcg/examples/Spring Batch Multithreading Example/spring-batch-multithreading-example)
2016-06-04 21:38:55.823  INFO 3484 --- [           main] com.quantvalley.batch.Application        : No active profile set, falling back to default profiles: default
2016-06-04 21:38:55.861  INFO 3484 --- [           main] s.c.a.AnnotationConfigApplicationContext : Refreshing org.springframework.context.annotation.AnnotationConfigApplicationContext@45afc369: startup date [Sat Jun 04 21:38:55 EET 2016]; root of context hierarchy
2016-06-04 21:38:56.348  WARN 3484 --- [           main] o.s.c.a.ConfigurationClassEnhancer       : @Bean method ScopeConfiguration.stepScope is non-static and returns an object assignable to Spring's BeanFactoryPostProcessor interface. This will result in a failure to process annotations such as @Autowired, @Resource and @PostConstruct within the method's declaring @Configuration class. Add the 'static' modifier to this method to avoid these container lifecycle issues; see @Bean javadoc for complete details.
2016-06-04 21:38:56.360  WARN 3484 --- [           main] o.s.c.a.ConfigurationClassEnhancer       : @Bean method ScopeConfiguration.jobScope is non-static and returns an object assignable to Spring's BeanFactoryPostProcessor interface. This will result in a failure to process annotations such as @Autowired, @Resource and @PostConstruct within the method's declaring @Configuration class. Add the 'static' modifier to this method to avoid these container lifecycle issues; see @Bean javadoc for complete details.
2016-06-04 21:38:56.498  INFO 3484 --- [           main] o.s.j.d.e.EmbeddedDatabaseFactory        : Starting embedded database: url='jdbc:hsqldb:mem:testdb', username='sa'
2016-06-04 21:38:57.018  INFO 3484 --- [           main] o.s.jdbc.datasource.init.ScriptUtils     : Executing SQL script from class path resource [org/springframework/batch/core/schema-hsqldb.sql]
2016-06-04 21:38:57.026  INFO 3484 --- [           main] o.s.jdbc.datasource.init.ScriptUtils     : Executed SQL script from class path resource [org/springframework/batch/core/schema-hsqldb.sql] in 7 ms.
2016-06-04 21:38:57.081  INFO 3484 --- [           main] o.s.j.e.a.AnnotationMBeanExporter        : Registering beans for JMX exposure on startup
2016-06-04 21:38:57.096  INFO 3484 --- [           main] o.s.b.a.b.JobLauncherCommandLineRunner   : Running default command line with: []
2016-06-04 21:38:57.104  INFO 3484 --- [           main] o.s.b.c.r.s.JobRepositoryFactoryBean     : No database type set, using meta data indicating: HSQL
2016-06-04 21:38:57.253  INFO 3484 --- [           main] o.s.b.c.l.support.SimpleJobLauncher      : No TaskExecutor has been set, defaulting to synchronous executor.
2016-06-04 21:38:57.307  INFO 3484 --- [           main] o.s.b.c.l.support.SimpleJobLauncher      : Job: [FlowJob: [name=FxMarket Volume ETL Job]] launched with the following parameters: [{run.id=1}]
2016-06-04 21:38:57.327  INFO 3484 --- [           main] o.s.batch.core.job.SimpleStepHandler     : Executing step: [Extract -> Transform -> Aggregate -> Load]
2016-06-04 21:39:06.319  INFO 3484 --- [           main] o.s.b.c.l.support.SimpleJobLauncher      : Job: [FlowJob: [name=FxMarket Volume ETL Job]] completed with the following parameters: [{run.id=1}] and the following status: [COMPLETED]
2016-06-04 21:39:06.321  INFO 3484 --- [           main] com.quantvalley.batch.Application        : Started Application in 10.677 seconds (JVM running for 11.016)
2016-06-04 21:39:06.322  INFO 3484 --- [       Thread-1] s.c.a.AnnotationConfigApplicationContext : Closing org.springframework.context.annotation.AnnotationConfigApplicationContext@45afc369: startup date [Sat Jun 04 21:38:55 EET 2016]; root of context hierarchy
2016-06-04 21:39:06.324  INFO 3484 --- [       Thread-1] o.s.j.e.a.AnnotationMBeanExporter        : Unregistering JMX-exposed beans on shutdown
2016-06-04 21:39:06.324  INFO 3484 --- [       Thread-1] o.s.j.d.e.EmbeddedDatabaseFactory        : Shutting down embedded database: url='jdbc:hsqldb:mem:testdb'

6. Download the Source Code

This was an example to show how to use Spring Batch Multithreading Job.

Download
You can download the full source code of this example here: SpringBatchMultithreadingExampleCode.zip

[출처] https://examples.javacodegeeks.com/enterprise-java/spring/batch/spring-batch-multithreading-example/

본 웹사이트는 광고를 포함하고 있습니다.
광고 클릭에서 발생하는 수익금은 모두 웹사이트 서버의 유지 및 관리, 그리고 기술 콘텐츠 향상을 위해 쓰여집니다.
번호 제목 글쓴이 날짜 조회 수
35 [EMF] EMF Tutorial EMF 튜터리얼 file 졸리운_곰 2023.08.23 283
34 Eclipse RAP Tutorial for Beginners - Workbench Application (OLD) file 졸리운_곰 2021.01.30 323
33 Learn Eclipse GMF in 15 minutes file 졸리운_곰 2019.11.27 253
32 [Eclipse] GEF entry series (10, an implementation of the form) file 졸리운_곰 2019.11.25 212
31 GEF Programmer Guide 번역 졸리운_곰 2019.11.25 227
30 Learn Eclipse GMF in 15 minutes file 졸리운_곰 2019.11.20 224
29 RCP 에디터 정리 졸리운_곰 2019.11.20 333
28 다른 그림과 관련하여 GEF 편집기 레이아웃에서 그림의 위치 제한 조건을 동적으로 계산 Dynamically calculating the position constraints for a figure in a GEF editor layout in relation to another figure file 졸리운_곰 2019.11.20 277
27 RCP 등에서 .mf 파일로 다른 프로젝트 익스포트 포함시 라이브러리(메소드)를 찾지 못할 때, Eclipse RCP - cant resolve importing libraries in final build file 졸리운_곰 2019.10.15 410
26 ESE2006-EclipseModelingSymposium15_GMF.pdf file 졸리운_곰 2019.09.21 302
25 GMF_Creation_Review.pdf file 졸리운_곰 2019.09.21 366
24 Eclipse EMF and GMF Tutorial file 졸리운_곰 2019.09.21 222
23 GMF Tutorial/ko file 졸리운_곰 2019.09.20 347
22 Model Driven Architecture approach to domain of graphical editors file 졸리운_곰 2019.09.20 268
21 Single_Sourcing_RAP_RCP_en.pdf file 졸리운_곰 2019.05.15 311
20 Rich client platform 설명 및 배우기 참고 졸리운_곰 2019.05.15 309
19 Rich Ajax Platform, Part 1: 소개 file 졸리운_곰 2019.05.15 320
18 또 하나의 크로스 플랫폼: Eclipse RAP file 졸리운_곰 2019.05.15 319
17 Eclipse 4 RCP 튜토리얼(완료) file 졸리운_곰 2019.05.14 977
16 Updating UI in Eclipse RCP 졸리운_곰 2015.11.07 408
대표 김성준 주소 : 경기 용인 분당수지 U타워 등록번호 : 142-07-27414
통신판매업 신고 : 제2012-용인수지-0185호 출판업 신고 : 수지구청 제 123호 개인정보보호최고책임자 : 김성준 sjkim70@stechstar.com
대표전화 : 010-4589-2193 [fax] 02-6280-1294 COPYRIGHT(C) stechstar.com ALL RIGHTS RESERVED