[Spring Batch] Job 과 Step

 

* 참고 사이트 : https://docs.spring.io/spring-batch/reference/index.html

Overview :: Spring Batch

 

 

 1. Job 이란?

 

 

 

 -  JobInstance, JobExecution 으로 구성

 

•  JobInstance : 논리적 작업 실행의 개념 ( JobParameter를 포함 ) 
•  JobExecution : 실제 Job 의 첫 번째 실행
•  JobExecution 은 다음과 같은 속성값을 가짐

 

 

 -  실제 관련 코드 (Job 실행)

@Scheduled(cron = "0 0 18 * * *")
    public void runJob() {
        if (!isUpdate) return;

        try {
            jobLauncher.run(jobRegistry.getJob(TRIP_RECOMMEND_JOB_NAME), new JobParametersBuilder()
                            .addString("time", LocalDateTime.now().toString())
                            .toJobParameters());
        } catch (NoSuchJobException | JobInstanceAlreadyCompleteException | JobExecutionAlreadyRunningException |
                     JobParametersInvalidException | JobRestartException e) {
                throw new RuntimeException(e);
        }
    }

 

 

 

 

 -  실제 관련 코드 (Job 구현)

    @Bean
    public Job jdbcJob() {
        return new JobBuilder(TRIP_RECOMMEND_JOB_NAME, jobRepository)
                .incrementer(new RunIdIncrementer())
                .start(tripLikeRecommendStep())
                .build();
    }

 

 

2. Step 이란?

 -  실제 Job 의 실행 단계를 의미

 

 - chunk 기반

 

 

• ItemReader : 작업을 처리할 데이터를 Load
• ItemProcessor : 작업의 실제 처리 과정
• ItemWriter : 작업한 데이터를 저장하는 과정 

 

- tasklet 기반

 

 

• 비교적 간단한 단일 작업으로 구성

 

- step execution property

 

 

 

 

 

 3. Step 실행 방법

 

 - Chunk

 

• 한 번에 하나씩 데이터를 읽고 지정된 청크 단위(커밋 단위) 로 묶어서 처리
• multi-thread step 을 이용하여 성능을 향상 할 수 있음

 

 

 - Tasklet

 

• 작업을 독립적으로(단건) 처리

 

 

 - Step 병렬 처리

 

• Step 의 처리 속도를 향상 시키기 위해서 작업을 병렬로 실행

 

 

 

 -  실제 관련 코드 (Step 구현)

 

/**
     * DB 에서 tripId를 Paging 형식으로 가져오기
     * Paging: multi - thread 에 대해 안전
     */
    @Bean
    public JdbcPagingItemReader<Long> jdbcPagingItemReader() {
        return new JdbcPagingItemReaderBuilder<Long>()
                .name("jdbcPageItemReader")
                .dataSource(dataSource)
                .pageSize(chunkSize)
                .queryProvider(createQueryProvider())
                .rowMapper(((rs, rowNum) -> rs.getLong(1)))
                .build();
    }

 

 

 

 -  실제 관련 코드 ( chunk multi - thread)

 

    /**
     * chunk 단위 멀티스레드 설정
     */
@Bean
public TaskExecutor taskExecutor(){
    ThreadPoolTaskExecutor threadPoolTaskExecutor = new ThreadPoolTaskExecutor();
    threadPoolTaskExecutor.setCorePoolSize(WORKER_SIZE);
    threadPoolTaskExecutor.setMaxPoolSize(WORKER_SIZE);
    threadPoolTaskExecutor.setThreadNamePrefix("exe-");
    threadPoolTaskExecutor.setWaitForTasksToCompleteOnShutdown(true);
    threadPoolTaskExecutor.setAllowCoreThreadTimeOut(true);
    threadPoolTaskExecutor.setKeepAliveSeconds(1);
    threadPoolTaskExecutor.initialize();
    return threadPoolTaskExecutor;
}