스프링 배치
배치작업이란?
배치작업은 실시간 명령과 반대되는 작업으로 일련의 작업을 모아서 한번에 처리하는것 이라고 설명할 수 있다.
스프링 배치 구조
스프링 배치의 구조는 아래와 같다.
JobRepository : 배치가 수행될때 수행되는 메타 데이터를 관리하고 시작 시간, 종료 시간 Job의 상태 등 배치 수행 관련 데이터를 저장
JobLauncher : Job을 실행시켜주는 역할
Job : 배치 작업
Step : 세부 작업 내역
Job과 step은 일대다 구조로, 1개의 Job은 여러개의 Step을 가질 수 있다.
위 구조가 일반적이며, 자세한 내용은 프로젝트에서 소개한다.
스프링 배치 프로젝트 생성하기
1) https://start.spring.io/ 접속
2) Spring Batch 추가
3) application.yml에 아래와 같이 추가
아래의 설정을 통해서 스프링 배치 프로그램 실행시에 Job의 이름을 입력 받도록 합니다.
spring:
batch:
job:
names: $(job.name:NONE)4) @EnableBatchProcessing 애노테이션 추가
@EnableBatchProcessing 애노테이션이 추가되어야 배치가 정상적으로 구동됨.
@EnableBatchProcessing
@SpringBootApplication
public class SpringBatchTutorialApplication {
public static void main(String[] args) {
SpringApplication.run(SpringBatchTutorialApplication.class, args);
}
}
5) HelloWorld를 출력하는 Job 생성
@Configuration
public class HelloWorldJobConfig {
/**
* JobBuilderFactory를 이용해서 Job 생성
*/
@Autowired
private JobBuilderFactory jobBuilderFactory;
/**
* StepBuildFactory를 이용해서 step 생성
*/
@Autowired
private StepBuilderFactory stepBuilderFactory;
@Bean
public Job helloWorldJob() {
return jobBuilderFactory.get("helloWorldJob")
.incrementer(new RunIdIncrementer())
.start(helloWorldStep())
.build();
}
@Bean
@JobScope
public Step helloWorldStep() {
return stepBuilderFactory.get("helloWorldStep")
.tasklet(helloWorldTasklet())
.build();
}
@Bean
public Tasklet helloWorldTasklet() {
return new Tasklet() {
@Override
public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception {
System.out.println("Hello world SPring Batch");
return RepeatStatus.FINISHED;
}
}
}
}
JobBuilderFactory : Job을 생성하기 위한 Factory로 Spring Batch에서 제공하는 빈
StepBuilderFactory : Step을 생성하기 위한 Factory로 Spring Batch에서 제공하는 빈
jobBuilderFactory.get("helloWorldJob") .incrementer(new RunIdIncrementer()) .start(helloWorldStep()) .build(); :
- 의존성주입받은 JobBuilderFactory를 이용해서 Job을 생성RunIdIncrementer를 이용해서 Job 생성시 자동으로 Id가 증가하도록 설정
- Job은 하위에 여러 Step을 갖을 수 있음.
return stepBuilderFactory.get("helloWorldStep") .tasklet(helloWorldTasklet()) .build(); :
- 의존성주입받은 StepBuilderFactory를 이용해서 step을 생성
이때, Job하위라는 명시를 위해 JobScope 애노테이션을 지정
- Step을 ItemReader,Writer, processor를 갖을 수 있으나, 특별한 작업이 없다면 Tasklet으로만 처리도 가능함. Tasklet은 단순한 배치 작업일때 사용 가능하며 별도로 읽고 쓸게 없다면 쓸 수 있다.
다만, 위 설정을 사용한다면 매개변수로 Job의 이름을 전달받아야 한다.
배치 실행시 파라미터 받기와 검증하기
스프링 배치에서는 Job을 실행하고 나면 스프링 배치에서 관리하고 있는 별도의 테이블이 존재한다.
- batch_job_execution : 스프링 배치에서 실행에 대한 관리를 해주는 테이블(일반적으로는 회사에서 알아서 만들어서 사용함)
@Configuration
public class ValidatedParamJobConfig {
/**
* JobBuilderFactory를 이용해서 Job 생성
*/
@Autowired
private JobBuilderFactory jobBuilderFactory;
/**
* StepBuildFactory를 이용해서 step 생성
*/
@Autowired
private StepBuilderFactory stepBuilderFactory;
@Bean
public Job validatedParamJob(Step validatedParamStep) {
return jobBuilderFactory.get("validatedParamJob")
.incrementer(new RunIdIncrementer())
// 파라미터 Validated 가능
// .validator(new FileParamValidator())
.validator(multipleValidator())
.start(validatedParamStep)
.build();
}
@Bean
@JobScope
public Step validatedParamStep(Tasklet validatedTasklet) {
return stepBuilderFactory.get("helloWorldStep")
.tasklet(validatedTasklet)
.build();
}
@Bean
@StepScope
public Tasklet validatedTasklet(@Value("#{jobParameters['fileName']}" String fileName)) {
return new Tasklet() {
@Override
public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception {
System.out.println("Validated Param Tasklet ");
// 여기에서도 fileName 검증 가능. 그러나 job에서도 가능하다.
return RepeatStatus.FINISHED;
}
}
}
/**
* 여러개의 Validator도 등록이 가능함.
* @return
*/
private CompositeJobParametersValidator multipleValidator() {
CompositeJobParametersValidator compositeJobParametersValidator = new CompositeJobParametersValidator();
compositeJobParametersValidator.setValidators(Arrays.asList(new FileParamValidator()));
return compositeJobParametersValidator;
}
}
@Value("#{jobParameters['fileName']}" String fileName: 이전 코드와 대부분 비슷하지만 위와 같이 Tasklet에서 코딩해주면 fileName이라는 Parameter를 String값으로 가져올 수 있다..validator(new FileParamValidator()): Job 생성시 입력된 파라미터에 대해서 Validator를 추가할 수도 있으며 기본 코드는 아래와 같다.
public class FileParamValidator implements JobParametersValidator {
/**
* 파일이름이 유효한지 검증
* @param parameters some {@link JobParameters} (can be {@code null})
* @throws JobParametersInvalidException
*/
@Override
public void validate(JobParameters parameters) throws JobParametersInvalidException {
String fileName = parameters.getString("fileName");
if (!StringUtils.endsWithIgnoreCase(fileName, "csv")) {
throw new JobParametersInvalidException("This is not csv file");
}
}
}
CompositeJobParametersValidator compositeJobParametersValidator = new CompositeJobParametersValidator();: 배열로서의 여러개의 Validator를 등록할수 수도 있다.
배치 작업 실행전, 후 리스너
기본 코드는 아래와 같다.
@Configuration
public class JobListenerConfig {
/**
* JobBuilderFactory를 이용해서 Job 생성
*/
@Autowired
private JobBuilderFactory jobBuilderFactory;
/**
* StepBuildFactory를 이용해서 step 생성
*/
@Autowired
private StepBuilderFactory stepBuilderFactory;
@Bean
public Job helloWorldJob(Step jobListenerStep) {
return jobBuilderFactory.get("jobListenerJob")
.incrementer(new RunIdIncrementer())
.listener(new JobLoggerListenerImpl())
.start(jobListenerStep)
.build();
}
@Bean
@JobScope
public Step jobListenerStep(Tasklet jobListenerTasklet) {
return stepBuilderFactory.get("jobListenerStep")
.tasklet(jobListenerTasklet)
.build();
}
@Bean
public Tasklet jobListenerTasklet() {
return new Tasklet() {
@Override
public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception {
System.out.println("Job Listener Tasklet");
return RepeatStatus.FINISHED;
}
}
}
}
.listener(new JobLoggerListenerImpl()): Validator를 추가했을때와 유사하게 배치 실행전, 후 Listenr를 위와 같이 등록해줄 수 있다.
실제 Listener는 아래와 같다.
@Slf4j
public class JobLoggerListenerImpl implements JobExecutionListener {
private static final String BEFORE_MESSAGE = "{} Job is Running";
private static final String AFTER_MESSAGE = "{} Job is Done. (Status : {})";
@Override
public void beforeJob(JobExecution jobExecution) {
log.info(BEFORE_MESSAGE, jobExecution.getJobInstance().getJobName());
}
@Override
public void afterJob(JobExecution jobExecution) {
log.info(AFTER_MESSAGE, jobExecution.getJobInstance().getJobName(), jobExecution.getStatus());
if (jobExecution.getStatus() == BatchStatus.FAILED) {
// email
log.info("job is Failed");
}
}
}
