Spring-Batch 02. 코드 실행해보기

 

이번에는 Tasklet을 직접 생성하여 스프링 배치를 실행해보겠습니다.

 

Lombok 사용을 위한 build.gradle Dependencies 종속성 추가

    // Lombok 의존성 추가
    compileOnly 'org.projectlombok:lombok:1.18.30'
    annotationProcessor 'org.projectlombok:lombok:1.18.30'

 

Tasklet 구현체 생성

package org.schooldevops.springbatch.sample;

import lombok.extern.slf4j.Slf4j;
import org.springframework.batch.core.StepContribution;
import org.springframework.batch.core.scope.context.ChunkContext;
import org.springframework.batch.core.step.tasklet.Tasklet;
import org.springframework.batch.repeat.RepeatStatus;
import org.springframework.beans.factory.InitializingBean;

@Slf4j
public class GreetingTask implements Tasklet, InitializingBean {

    @Override
    public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception {
        log.info("---------------Task Execute -------------");
        log.info("GreetingTask: {}, {}", contribution, chunkContext);
        return RepeatStatus.FINISHED;
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        log.info("--------------After properties Sets() --------------");
    }

}

 

- Tasklet과 InitializeBean 인터페이스를 구현

？ 구현한다 에 대한 표현이 헷갈립니다.

그래서 검색해보았습니다.

 

부모 인터페이스가 자식 클래스에게 implements되는 것을 구현 한다 라고 하는데, 자식 클래스에서 재정의하여 구현하기 때문입니다.

상속과 달리 반드시 인터페이스의 메서드를 재정의하여 구현해야 합니다.

 

- Tasklet의 execute 메소드 구현

- InitializeBean의 afterPropertiesSet 메소드 구현

 

execute

- StepContribution, ChunkContext 를 파라미터로 받음

- RepeatStatus를 반환

  - FINISHED : Tasklet이 종료됨

  - CONTINUABLE : 계속해서 태스크를 수행할꺼임

  - CONTINUEIF(condition) : 조건에 따라 종료할지 지속할지 결정, ？ enum 구현체 내에는 작성이 안되어있다??

public enum RepeatStatus {
    CONTINUABLE(true),
    FINISHED(false);

    private final boolean continuable;

    private RepeatStatus(boolean continuable) {
        this.continuable = continuable;
    }

    public static RepeatStatus continueIf(boolean continuable) {
        return continuable ? CONTINUABLE : FINISHED;
    }

    public boolean isContinuable() {
        return this == CONTINUABLE;
    }

    public RepeatStatus and(boolean value) {
        return value && this.continuable ? CONTINUABLE : FINISHED;
    }
}

 

afterPropertiesSet

	@Override
	public void afterPropertiesSet() throws Exception {
		initializeTransitionsIfNotInitialized();
	}

 

- 태스크를 수행할때 프로퍼티를 설정 후 수행하는 메소드

- 없어도 됨

실행되는 동작 메소드의 네이밍이 멋지다. 바로 의미를 알아볼 수 있도록 되어있습니다.

 

	private synchronized void initializeTransitionsIfNotInitialized() {
		if (startState == null) {
			initializeTransitions();
		}
	}

그런데 메소드에 Synchronized 키워드가 입력되어 있는데, 여러개의 스레드가 한개의 자원을 사용하고자 할 때, 현재 데이터를 사용하고 있는 해당 스레드를 제외하고 나머지 스레드들은 데이터에 접근 할 수 없도록 막는 개념입니다.

 

@Configuration 으로 생성할 배치 빈을 스프링에 등록

package org.schooldevops.springbatch.sample;

import lombok.extern.slf4j.Slf4j;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.job.builder.JobBuilder;
import org.springframework.batch.core.launch.support.RunIdIncrementer;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.step.builder.StepBuilder;
import org.springframework.batch.core.step.tasklet.Tasklet;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.transaction.PlatformTransactionManager;

@Slf4j
@Configuration
public class BasicTaskJobConfiguration {
    @Autowired
    PlatformTransactionManager transactionManager;

    @Bean
    public Tasklet greetingTasklet(){
        return new GreetingTask();
    }

    @Bean
    public Step step(JobRepository jobRepository, PlatformTransactionManager transactionManager){
        log.info("--------------- Init myStep ------------------");

        return new StepBuilder("myStep", jobRepository)
                .tasklet(greetingTasklet(), transactionManager)
                .build();
    }

    @Bean
    public Job myJob(Step step, JobRepository jobRepository){
        log.info("------------- Init myJob ---------------");
        return new JobBuilder("myJob", jobRepository)
                .incrementer(new RunIdIncrementer())
                .start(step)
                .build();
    }
}

 

- greetingTasklet 메소드를 통해서 Tasklet을 빈에 등록

- step 메소드를 통해서 step 빈을 등록, JobRepository와 PlatformTransactionManager 을 파라미터로 받음

- 스프링 배치는 데이터 소스와 함께 작업하므로 PlatformTransactionManager가 필요합니다.

- StepBuilder 를 생성

- 해당 스텝은 JobRepository에 등록됨

- build() 를 통해서 스텝을 생성하고 빈으로 등록하도록 return

 

- JobBuilder를 통해서 잡을 생성했고 , Job 은 Step과 JobRepository가 필요합니다.

- incrementer는 잡이 지속적으로 실행될때, 유니크성을 구분할 수 있는 방법을 설정

- RunIdIncrementer는 잡의 아이디를 실행할때 지속적으로 증가 시키면서 유니크한 잡을 실행

- start(step) 으로 잡의 시작포인트 설정, 파라미터로 받은 step을 등록 해주면서

- builder()로 잡을 생성하고 빈에 등록되도록 return

 

gradle :bootRun 실행 결과

2024-10-14T21:58:19.803+09:00  INFO 11691 --- [           main] com.zaxxer.hikari.pool.HikariPool        : HikariPool-1 - Added connection com.mysql.cj.jdbc.ConnectionImpl@776802b0
2024-10-14T21:58:19.807+09:00  INFO 11691 --- [           main] com.zaxxer.hikari.HikariDataSource       : HikariPool-1 - Start completed.
2024-10-14T21:58:20.203+09:00  INFO 11691 --- [           main] o.s.s.sample.BasicTaskJobConfiguration   : --------------- Init myStep ------------------
2024-10-14T21:58:20.221+09:00  INFO 11691 --- [           main] o.s.s.sample.BasicTaskJobConfiguration   : ------------- Init myJob ---------------
2024-10-14T21:58:20.294+09:00  INFO 11691 --- [           main] o.s.s.sample.SampleApplication           : Started SampleApplication in 11.152 seconds (process running for 11.295)
2024-10-14T21:58:20.296+09:00  INFO 11691 --- [           main] o.s.b.a.b.JobLauncherApplicationRunner   : Running default command line with: []
2024-10-14T21:58:26.816+09:00  INFO 11691 --- [           main] o.s.b.c.l.support.SimpleJobLauncher      : Job: [SimpleJob: [name=myJob]] launched with the following parameters: [{'run.id':'{value=1, type=class java.lang.Long, identifying=true}'}]
2024-10-14T21:58:27.181+09:00  INFO 11691 --- [           main] o.s.batch.core.job.SimpleStepHandler     : Executing step: [myStep]
2024-10-14T21:58:27.330+09:00  INFO 11691 --- [           main] o.s.springbatch.sample.GreetingTask      : ---------------Task Execute -------------
2024-10-14T21:58:27.331+09:00  INFO 11691 --- [           main] o.s.springbatch.sample.GreetingTask      : GreetingTask: [StepContribution: read=0, written=0, filtered=0, readSkips=0, writeSkips=0, processSkips=0, exitStatus=EXECUTING], ChunkContext: attributes=[], complete=false, stepContext=SynchronizedAttributeAccessor: [], stepExecutionContext={batch.version=5.1.2, batch.taskletType=org.schooldevops.springbatch.sample.GreetingTask, batch.stepType=org.springframework.batch.core.step.tasklet.TaskletStep}, jobExecutionContext={batch.version=5.1.2}, jobParameters={run.id=1}
2024-10-14T21:58:27.458+09:00  INFO 11691 --- [           main] o.s.batch.core.step.AbstractStep         : Step: [myStep] executed in 276ms
2024-10-14T21:58:27.661+09:00  INFO 11691 --- [           main] o.s.b.c.l.support.SimpleJobLauncher      : Job: [SimpleJob: [name=myJob]] completed with the following parameters: [{'run.id':'{value=1, type=class java.lang.Long, identifying=true}'}] and the following status: [COMPLETED] in 741ms
2024-10-14T21:58:27.680+09:00  INFO 11691 --- [ionShutdownHook] com.zaxxer.hikari.HikariDataSource       : HikariPool-1 - Shutdown initiated...
2024-10-14T21:58:32.086+09:00  INFO 11691 --- [ionShutdownHook] com.zaxxer.hikari.HikariDataSource       : HikariPool-1 - Shutdown completed.

BUILD SUCCESSFUL in 24s
4 actionable tasks: 4 executed

 

After Properties Sets() -> SimpleJobLauncher -> SimpleStepHandler -> GreetingTask 순으로 실행됨을 볼 수 있습니다.

 

Reference

 

https://devocean.sk.com/experts/techBoardDetail.do?ID=166690&boardType=experts&page=&searchData=&subIndex=&idList=&searchText=&techType=&searchDataSub=&searchDataMain=&writerID=kido

[SpringBatch 연재 02] SpringBatch 코드 설명 및 아키텍처 알아보기