- 전체
- JAVA 일반
- JAVA 수학
- JAVA 그래픽
- JAVA 자료구조
- JAVA 인공지능
- JAVA 인터넷
- Java Framework
- Java GUI (AWT,SWING,SWT,JFACE)
- SWT and RCP (web RAP/RWT)[eclipse], EMF
Java Framework Spring Batch Multithreading Example
2019.01.31 11:22
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 . Let’s take a look below.
1. Multithreaded Step
Spring Batch’s multithreaded step concept allows a batch job to use Spring’s 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
is our new writer that aggregates the trading day volume for each stock, then update the . It looks like our Spring Batch ETL Job writer 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
is a that provides a callback function to load the stocks volumes into the CSV file 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 implementation (we use 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:
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@Beanpublic Job fxMarketPricesETLJob() { return jobBuilderFactory.get("FxMarket Volume ETL Job").incrementer(new RunIdIncrementer()).listener(listener()) .flow(etlStep()).end().build();}@Beanpublic TaskExecutor taskExecutor(){ SimpleAsyncTaskExecutor asyncTaskExecutor=new SimpleAsyncTaskExecutor("spring_batch"); asyncTaskExecutor.setConcurrencyLimit(5); return asyncTaskExecutor;}@Beanpublic 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 file, then it aggregates and save daily volume for each stock in .
4.2.1. Input
trades.csv:
|
01
02
03
04
05
06
07
08
09
10
|
stock,time,price,sharesJHX,09:30:00.00,57,95JNJ,09:30:00.00,91.14,548OPK,09:30:00.00,8.3,300OPK,09:30:00.00,8.3,63OMC,09:30:00.00,74.53,100OMC,09:30:00.00,74.53,24TWTR,09:30:00.00,64.89,100TWTR,09:30:00.00,64.89,25TWTR,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,volumeELAY,8563079EEM,9220571FTR,12444516AEGY,12869499ERBB,19696299MJNA,8263325PVEC,10083433FITX,14781867BRGO,11458750BAC,10860160 |
5. Conclusion
It was noticeable that there is a significant time different when we enable the step multithreading feature where the time () is almost is about 63.5 % of the total time () 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: default2016-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 hierarchy2016-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 startup2016-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: HSQL2016-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 hierarchy2016-06-04 21:37:21.018 INFO 3312 --- [ Thread-1] o.s.j.e.a.AnnotationMBeanExporter : Unregistering JMX-exposed beans on shutdown2016-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: default2016-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 hierarchy2016-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 startup2016-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: HSQL2016-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 hierarchy2016-06-04 21:39:06.324 INFO 3484 --- [ Thread-1] o.s.j.e.a.AnnotationMBeanExporter : Unregistering JMX-exposed beans on shutdown2016-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.
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/
광고 클릭에서 발생하는 수익금은 모두 웹사이트 서버의 유지 및 관리, 그리고 기술 콘텐츠 향상을 위해 쓰여집니다.
댓글 0
| 번호 | 제목 | 글쓴이 | 날짜 | 조회 수 |
|---|---|---|---|---|
| 6 |
java4autocad Java for Autocad
| 졸리운_곰 | 2017.04.26 | 387 |
| 5 | Java3D and Eclipse | 졸리운_곰 | 2015.08.12 | 317 |
| 4 |
Draw2d Intro
| 졸리운_곰 | 2015.07.29 | 514 |
| 3 |
FreeLayout: A New Java Layout
| 졸리운_곰 | 2015.05.14 | 315 |
| 2 |
XML file to Treeviewer
| 졸리운_곰 | 2015.03.19 | 398 |
| 1 |
Java GUI(AWT) 프로그래밍
| 졸리운_곰 | 2015.03.09 | 938 |

