Job 실습해보기

1. HelloWorld 띄우기

2-1. Job Parameter ChunkContext

2-2. Job Parameter 늦은 바인딩

3-1. Job Parameter 유효성 검증하기 1

3-2. Job Parameter 유효성 검증하기 2

4-1. Job Parameter 증가시키기

4-2. Job Parameter 자동 타임스탬프 사용하기

5-1. Job Listener - JobExecutionListener 구현

5-2. Job Listener - 애너테이션 사용하기

6-1. Job ExecutionContext 조작하기 - Job 의 ExecutionContext 에 foo 데이터 추가하기

6-2. Job ExecutionContext 조작하기 - Step 의 ExecutionContext 에 foo 데이터 추가하기

6-3. Job ExecutionContext 에 foo 키 승격하기

7-1 Step-Flow

1. HelloWorld 띄우기

/**
 * Hello World Job
 */
@EnableBatchProcessing
@Configuration
@RequiredArgsConstructor
public class Ch01 {

    private final JobBuilderFactory jobBuilderFactory;
    private final StepBuilderFactory stepBuilderFactory;

    @Bean
    public Job job() {
        return this.jobBuilderFactory.get("basicJob")
                .start(step1())
                .build();
    }

    @Bean
    public Step step1() {
        return this.stepBuilderFactory.get("step1")
                .tasklet((contribution, chunkContext) -> {
                    System.out.println("contribution = " + contribution);
                    System.out.println("chunkContext = " + chunkContext);
                    System.out.println("Hello World");
                    return RepeatStatus.FINISHED;
                })
                .build();
    }

}

2-1. Job Parameter ChunkContext

/**
 * Job Parameter ChunkContext
 */
@EnableBatchProcessing
@Configuration
@RequiredArgsConstructor
public class Ch02_1 {

    private final JobBuilderFactory jobBuilderFactory;
    private final StepBuilderFactory stepBuilderFactory;

    @Bean
    public Job job() {
        return this.jobBuilderFactory.get("basicJob")
                .start(step1())
                .build();
    }

    @Bean
    public Step step1() {
        return this.stepBuilderFactory.get("step1")
                .tasklet(helloWorldTasklet())
                .build();
    }

    private Tasklet helloWorldTasklet() {
        return (contribution, chunkContext) -> {
            String foo = (String) chunkContext.getStepContext()
                    .getJobParameters()
                    .get("foo");
            System.out.println("foo = " + foo);
            return RepeatStatus.FINISHED;
        };
    }

}

2-2. Job Parameter 늦은 바인딩

/**
 * Job Parameter 늦은 바인딩
 */
@EnableBatchProcessing
@Configuration
@RequiredArgsConstructor
public class Ch02_2 {

    private final JobBuilderFactory jobBuilderFactory;
    private final StepBuilderFactory stepBuilderFactory;

    @Bean
    public Job job() {
        return this.jobBuilderFactory.get("basicJob")
                .start(step1())
                .build();
    }

    @Bean
    public Step step1() {
        return this.stepBuilderFactory.get("step1")
                .tasklet(helloWorldTasklet(null))
                .build();
    }

    /**
     * 실행 명령어
     * // 스프링 배치의 JobParameters 는 스프링 부트의 명령행 기능을 사용해 프로퍼티를 구성하는 것과 다르다
     * // 따라서 -- 접두사 또는 -D 아규먼트를 사용하면 안된다.
     * java -jar ./build/libs/demo-0.0.1-SNAPSHOT.jar foo=bar
     * java -jar ./build/libs/demo-0.0.1-SNAPSHOT.jar -foo=bar
     * // - 를 붙이면  특정 잡파라미터는 식별에는 사용되지 않도록 함(중복실행가능하다는 뜻)
     * @param foo
     * @return
     */
    @StepScope
    @Bean
    public Tasklet helloWorldTasklet(@Value("#{jobParameters['foo']}") String foo) {
        return (contribution, chunkContext) -> {
            System.out.println("foo = " + foo);
            return RepeatStatus.FINISHED;
        };
    }

}

3-1. Job Parameter 유효성 검증하기 1

/**
 * Job Parameter 유효성 검증하기 한개
 */
@EnableBatchProcessing
@Configuration
@RequiredArgsConstructor
public class Ch03_1 {

    public static class ParameterValidator implements JobParametersValidator{

        @Override
        public void validate(JobParameters parameters) throws JobParametersInvalidException {
            System.out.println("ParameterValidator.validate");
            System.out.println("parameters = " + parameters);
            System.out.println("parameters foo = " + parameters.getString("foo"));
            String foo = parameters.getString("foo");
            if (!StringUtils.hasText(foo))
                throw new JobParametersInvalidException("foo must not empty");
            else if (!StringUtils.endsWithIgnoreCase(foo, "bar"))
                throw new JobParametersInvalidException("foo must is bar");

        }
    }

    @Bean
    public JobParametersValidator validator() {
        DefaultJobParametersValidator validator = new DefaultJobParametersValidator();
        validator.setRequiredKeys(new String[]{"foo"});
        validator.setOptionalKeys(new String[]{"name"});
        return validator;
    }

    private final JobBuilderFactory jobBuilderFactory;
    private final StepBuilderFactory stepBuilderFactory;

    @Bean
    public Job job() {
        return this.jobBuilderFactory.get("basicJob")
                .start(step1())
                .validator(validator())
                .build();
    }

    @Bean
    public Step step1() {
        return this.stepBuilderFactory.get("step1")
                .tasklet(helloWorldTasklet(null, null))
                .build();
    }

    @StepScope
    @Bean
    public Tasklet helloWorldTasklet(@Value("#{jobParameters['foo']}")String foo,
                                     @Value("#{jobParameters['name']}")String name) {
        return (contribution, chunkContext) -> {
            System.out.println("foo = " + foo);
            System.out.println("name = " + name);
            return RepeatStatus.FINISHED;
        };
    }

}

3-2. Job Parameter 유효성 검증하기 2

/**
 * Job Parameter 유효성 검증하기 두개
 */
@EnableBatchProcessing
@Configuration
@RequiredArgsConstructor
public class Ch03_2 {

    public static class ParameterValidator implements JobParametersValidator{

        @Override
        public void validate(JobParameters parameters) throws JobParametersInvalidException {
            System.out.println("ParameterValidator.validate");
            System.out.println("parameters = " + parameters);
            System.out.println("parameters foo = " + parameters.getString("foo"));
            String foo = parameters.getString("foo");
            if (!StringUtils.hasText(foo))
                throw new JobParametersInvalidException("foo must not empty");
            else if (!StringUtils.endsWithIgnoreCase(foo, "bar"))
                throw new JobParametersInvalidException("foo must is bar");
        }
    }

    @Bean
    public CompositeJobParametersValidator validator() {
        CompositeJobParametersValidator validator = new CompositeJobParametersValidator();

        DefaultJobParametersValidator defaultJobParametersValidator = new DefaultJobParametersValidator();
        defaultJobParametersValidator.setRequiredKeys(new String[]{"foo"});
        defaultJobParametersValidator.setOptionalKeys(new String[]{"name"});

        defaultJobParametersValidator.afterPropertiesSet();

        validator.setValidators(
                Arrays.asList(new ParameterValidator(), defaultJobParametersValidator)
        );

        return validator;
    }

    private final JobBuilderFactory jobBuilderFactory;
    private final StepBuilderFactory stepBuilderFactory;

    @Bean
    public Job job() {
        return this.jobBuilderFactory.get("basicJob")
                .start(step1())
                .validator(validator())
                .build();
    }

    @Bean
    public Step step1() {
        return this.stepBuilderFactory.get("step1")
                .tasklet(helloWorldTasklet(null, null))
                .build();
    }

    @StepScope
    @Bean
    public Tasklet helloWorldTasklet(@Value("#{jobParameters['foo']}")String foo,
                                     @Value("#{jobParameters['name']}")String name) {
        return (contribution, chunkContext) -> {
            System.out.println("foo = " + foo);
            System.out.println("name = " + name);
            return RepeatStatus.FINISHED;
        };
    }

}

4-1. Job Parameter 증가시키기

/**
 * Job Parameter 증가시키기
 */
@EnableBatchProcessing
@Configuration
@RequiredArgsConstructor
public class Ch04_1 {

    public static class ParameterValidator implements JobParametersValidator{

        @Override
        public void validate(JobParameters parameters) throws JobParametersInvalidException {
            System.out.println("ParameterValidator.validate");
            System.out.println("parameters = " + parameters);
            System.out.println("parameters foo = " + parameters.getString("foo"));
            String foo = parameters.getString("foo");
            if (!StringUtils.hasText(foo))
                throw new JobParametersInvalidException("foo must not empty");
            else if (!StringUtils.endsWithIgnoreCase(foo, "bar"))
                throw new JobParametersInvalidException("foo must is bar");
        }
    }

    @Bean
    public CompositeJobParametersValidator validator() {
        CompositeJobParametersValidator validator = new CompositeJobParametersValidator();

        DefaultJobParametersValidator defaultJobParametersValidator = new DefaultJobParametersValidator();
        defaultJobParametersValidator.setRequiredKeys(new String[]{"foo"});
        defaultJobParametersValidator.setOptionalKeys(new String[]{"name", "run.id"});

        defaultJobParametersValidator.afterPropertiesSet();

        validator.setValidators(
                Arrays.asList(new ParameterValidator(), defaultJobParametersValidator)
        );

        return validator;
    }

    private final JobBuilderFactory jobBuilderFactory;
    private final StepBuilderFactory stepBuilderFactory;

    @Bean
    public Job job() {
        return this.jobBuilderFactory.get("basicJob")
                .start(step1())
                .validator(validator())
                .incrementer(new RunIdIncrementer())
                .build();
    }

    @Bean
    public Step step1() {
        return this.stepBuilderFactory.get("step1")
                .tasklet(helloWorldTasklet(null, null, null))
                .build();
    }

    @StepScope
    @Bean
    public Tasklet helloWorldTasklet(@Value("#{jobParameters['foo']}")String foo,
                                     @Value("#{jobParameters['name']}")String name,
                                     @Value("#{jobParameters['run.id']}")Long id) {
        return (contribution, chunkContext) -> {
            System.out.println("foo = " + foo);
            System.out.println("name = " + name);
            System.out.println("id = " + id);
            return RepeatStatus.FINISHED;
        };
    }

}

4-2. Job Parameter 자동 타임스탬프 사용하기

/**
 * Job Parameter 자동 타임스템프 사용하기
 */
@EnableBatchProcessing
@Configuration
@RequiredArgsConstructor
public class Ch04_2 {

    public static class ParameterValidator implements JobParametersValidator{

        @Override
        public void validate(JobParameters parameters) throws JobParametersInvalidException {
            System.out.println("ParameterValidator.validate");
            System.out.println("parameters = " + parameters);
            System.out.println("parameters foo = " + parameters.getString("foo"));
            String foo = parameters.getString("foo");
            if (!StringUtils.hasText(foo))
                throw new JobParametersInvalidException("foo must not empty");
            else if (!StringUtils.endsWithIgnoreCase(foo, "bar"))
                throw new JobParametersInvalidException("foo must is bar");
        }
    }

    public static class DailyJobTimeStamper implements JobParametersIncrementer{

        @Override
        public JobParameters getNext(JobParameters parameters) {
            return new JobParametersBuilder(parameters)
                    .addDate("currentDate", new Date())
                    .toJobParameters();
        }
    }

    @Bean
    public CompositeJobParametersValidator validator() {
        CompositeJobParametersValidator validator = new CompositeJobParametersValidator();

        DefaultJobParametersValidator defaultJobParametersValidator = new DefaultJobParametersValidator();
        defaultJobParametersValidator.setRequiredKeys(new String[]{"foo"});
        defaultJobParametersValidator.setOptionalKeys(new String[]{"name", "currentDate"});

        defaultJobParametersValidator.afterPropertiesSet();

        validator.setValidators(
                Arrays.asList(new ParameterValidator(), defaultJobParametersValidator)
        );

        return validator;
    }

    private final JobBuilderFactory jobBuilderFactory;
    private final StepBuilderFactory stepBuilderFactory;

    @Bean
    public Job job() {
        return this.jobBuilderFactory.get("basicJob")
                .start(step1())
                .validator(validator())
                .incrementer(new DailyJobTimeStamper())
                .build();
    }

    @Bean
    public Step step1() {
        return this.stepBuilderFactory.get("step1")
                .tasklet(helloWorldTasklet(null, null, null))
                .build();
    }

    @StepScope
    @Bean
    public Tasklet helloWorldTasklet(@Value("#{jobParameters['foo']}")String foo,
                                     @Value("#{jobParameters['name']}")String name,
                                     @Value("#{jobParameters['currentDate']}")Long currentDate) {
        return (contribution, chunkContext) -> {
            System.out.println("foo = " + foo);
            System.out.println("name = " + name);
            System.out.println("currentDate = " + new Date(currentDate));
            return RepeatStatus.FINISHED;
        };
    }

}

5-1. Job Listener - JobExecutionListener 구현

/**
 * Job Listener 잡 리스너 사용하기 JobExecutionListener 구현
 */
@EnableBatchProcessing
@Configuration
@RequiredArgsConstructor
public class Ch05_1 {

    public static class JobLoggerListener implements JobExecutionListener {
        private static String START_MESSAGE = "%s is beginning execution";
        private static String END_MESSAGE = "%s has completed with the status %s";

        @Override
        public void beforeJob(JobExecution jobExecution) {
            System.out.println(String.format(START_MESSAGE, jobExecution.getJobInstance().getJobName()));
        }

        @Override
        public void afterJob(JobExecution jobExecution) {
            System.out.println(String.format(END_MESSAGE,
                    jobExecution.getJobInstance().getJobName(),
                    jobExecution.getStatus()));
        }
    }


    private final JobBuilderFactory jobBuilderFactory;
    private final StepBuilderFactory stepBuilderFactory;

    @Bean
    public Job job() {
        return this.jobBuilderFactory.get("basicJob")
                .start(step1())
                .listener(new JobLoggerListener())
                .build();
    }

    @Bean
    public Step step1() {
        return this.stepBuilderFactory.get("step1")
                .tasklet(helloWorldTasklet(null))
                .build();
    }

    @StepScope
    @Bean
    public Tasklet helloWorldTasklet(@Value("#{jobParameters['foo']}")String foo) {
        return (contribution, chunkContext) -> {
            System.out.println("foo = " + foo);

            return RepeatStatus.FINISHED;
        };
    }

}

5-2. Job Listener - 애너테이션 사용하기

/**
 * Job Listener 잡 리스너 사용하기 애너테이션 사용
 */
@EnableBatchProcessing
@Configuration
@RequiredArgsConstructor
public class Ch05_2 {

    public static class JobLoggerListener {
        private static String START_MESSAGE = "%s is beginning execution";
        private static String END_MESSAGE = "%s has completed with the status %s";

        @BeforeJob
        public void beforeJob(JobExecution jobExecution) {
            System.out.println(String.format(START_MESSAGE, jobExecution.getJobInstance().getJobName()));
        }

        @AfterJob
        public void afterJob(JobExecution jobExecution) {
            System.out.println(String.format(END_MESSAGE,
                    jobExecution.getJobInstance().getJobName(),
                    jobExecution.getStatus()));
        }
    }

    private final JobBuilderFactory jobBuilderFactory;
    private final StepBuilderFactory stepBuilderFactory;

    @Bean
    public Job job() {
        return this.jobBuilderFactory.get("basicJob")
                .start(step1())
                .listener(JobListenerFactoryBean.getListener(new JobLoggerListener()))
                .build();
    }

    @Bean
    public Step step1() {
        return this.stepBuilderFactory.get("step1")
                .tasklet(helloWorldTasklet(null))
                .build();
    }

    @StepScope
    @Bean
    public Tasklet helloWorldTasklet(@Value("#{jobParameters['foo']}")String foo) {
        return (contribution, chunkContext) -> {
            System.out.println("foo = " + foo);
            return RepeatStatus.FINISHED;
        };
    }

}

6-1. Job ExecutionContext 조작하기 - Job 의 ExecutionContext 에 foo.value 추가하기

/**
 * Job ExecutionContext 조작하기
 * Job 의 ExecutionContext 에 foo.value 추가하기
 */
@EnableBatchProcessing
@Configuration
@RequiredArgsConstructor
public class Ch06_1 {

    public static class HelloWorld implements Tasklet {
        private final static String HELLO_WORLD = "Hello, %s";
        @Override
        public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception {
            String foo = (String) chunkContext.getStepContext()
                    .getJobParameters()
                    .get("foo");
            ExecutionContext jobContext = chunkContext.getStepContext()
                    .getStepExecution()
                    .getJobExecution()
                    .getExecutionContext();
            jobContext.put("foo.value", foo);
            System.out.println(String.format(HELLO_WORLD, foo));
            return RepeatStatus.FINISHED;
        }
    }

    private final JobBuilderFactory jobBuilderFactory;
    private final StepBuilderFactory stepBuilderFactory;

    @Bean
    public Job job() {
        return this.jobBuilderFactory.get("basicJob")
                .start(step1())
                .listener(JobListenerFactoryBean.getListener(new Ch05_1.JobLoggerListener()))
                .build();
    }

    @Bean
    public Step step1() {
        return this.stepBuilderFactory.get("step1")
                .tasklet(new HelloWorld())
                .build();
    }

}

6-2. Job ExecutionContext 조작하기 - Step 의 ExecutionContext 에 foo.value 추가하기

/**
 * Job ExecutionContext 조작하기
 * Step 의 ExecutionContext 에 foo 추가하기
 */
@EnableBatchProcessing
@Configuration
@RequiredArgsConstructor
public class Ch06_2 {

    public static class HelloWorld implements Tasklet {
        private final static String HELLO_WORLD = "Hello, %s";
        @Override
        public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception {
            String foo = (String) chunkContext.getStepContext()
                    .getJobParameters()
                    .get("foo");
            ExecutionContext stepContext = chunkContext.getStepContext()
                    .getStepExecution()
                    .getExecutionContext();
            stepContext.put("foo.value", foo);
            System.out.println(String.format(HELLO_WORLD, foo));
            return RepeatStatus.FINISHED;
        }
    }

    private final JobBuilderFactory jobBuilderFactory;
    private final StepBuilderFactory stepBuilderFactory;

    @Bean
    public Job job() {
        return this.jobBuilderFactory.get("basicJob")
                .start(step1())
                .listener(JobListenerFactoryBean.getListener(new Ch05_1.JobLoggerListener()))
                .build();
    }

    @Bean
    public Step step1() {
        return this.stepBuilderFactory.get("step1")
                .tasklet(new HelloWorld())
                .build();
    }

}

6-3. 잡의 ExecutionContext에 name 키 승격하기

/**
 * Job ExecutionContext 조작하기
 * Job 의 ExecutionContext 에 foo 승격시키기
 * =================> java -jar ./build/libs/demo-0.0.1-SNAPSHOT.jar foo=bar 실행
 */
@EnableBatchProcessing
@Configuration
@RequiredArgsConstructor
public class Ch06_3 {

    public static class HelloWorld implements Tasklet {
        private final static String HELLO_WORLD = "Hello, %s";
        @Override
        public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception {
            String foo = (String) chunkContext.getStepContext()
                    .getJobParameters()
                    .get("foo");
            ExecutionContext stepContext = chunkContext.getStepContext()
                    .getStepExecution()
                    .getExecutionContext();
            stepContext.put("foo.value", foo);
            System.out.println(String.format(HELLO_WORLD, foo));
            return RepeatStatus.FINISHED;
        }
    }

    private final JobBuilderFactory jobBuilderFactory;
    private final StepBuilderFactory stepBuilderFactory;

    @Bean
    public Job job() {
        return this.jobBuilderFactory.get("basicJob")
                .start(step1())
                .next(step2())
                .listener(JobListenerFactoryBean.getListener(new Ch05_1.JobLoggerListener()))
                .build();
    }

    @Bean
    public Step step1() {
        return this.stepBuilderFactory.get("step1")
                .tasklet(new HelloWorld())
                .listener(promotionListener())
                .build();
    }

    @Bean
    public StepExecutionListener promotionListener() {
        ExecutionContextPromotionListener listener = new ExecutionContextPromotionListener();
        listener.setKeys(new String[]{"foo"});
        return listener;
    }

    @Bean
    public Step step2() {
        return this.stepBuilderFactory.get("step2")
                .tasklet((contribution, chunkContext) -> {
                    System.out.println(chunkContext.getStepContext().getJobParameters().get("foo"));
                    return RepeatStatus.FINISHED;
                })
                .build();
    }

}

7-1. Step-Flow

@Configuration
@RequiredArgsConstructor
public class FlowStepConfig {

    private final JobBuilderFactory jobBuilderFactory;
    private final StepBuilderFactory stepBuilderFactory;

    @Bean
    public Job exampleJob() {
        return jobBuilderFactory.get("exampleJob")
                .start(startStep())
                .on("FAILED") //startStep의 ExitStatus가 FAILED일 경우
                .to(failOverStep()) //failOver Step을 실행 시킨다.
                .on("*") //failOver Step의 결과와 상관없이
                .to(writeStep()) //write Step을 실행 시킨다.
                .on("*") //write Step의 결과와 상관없 이
                .end() //Flow를 종료시킨다.

                .from(startStep()) //startStep이 FAILED가 아니고
                .on("COMPLETED") //COMPLETED일 경우
                .to(processStep()) //process Step을 실행 시킨다
                .on("*") //process Step의 결과와 상관없이
                .to(writeStep()) // write Step을 실행 시킨다.
                .on("*") //wrtie Step의 결과와 상관없이
                .end() //Flow를 종료 시킨다.

                .from(startStep()) //startStep의 결과가 FAILED, COMPLETED가 아닌
                .on("*") //모든 경우
                .to(writeStep()) //write Step을 실행시킨다.
                .on("*") //write Step의 결과와 상관없이
                .end() //Flow를 종료시킨다.
                .end()
                .build();
    }

    @Bean
    public Step startStep() {
        return stepBuilderFactory.get("startStep")
                .tasklet((contribution, chunkContext) -> {
                    System.out.println("FlowStepConfig.startStep");

                    String result = "COMPLETED";
                    //String result = "FAIL";
                    //String result = "UNKNOWN";

                    //Flow에서 on은 RepeatStatus가 아닌 ExitStatus를 바라본다.
                    if(result.equals("COMPLETED"))
                        contribution.setExitStatus(ExitStatus.COMPLETED);
                    else if(result.equals("FAIL"))
                        contribution.setExitStatus(ExitStatus.FAILED);
                    else if(result.equals("UNKNOWN"))
                        contribution.setExitStatus(ExitStatus.UNKNOWN);

                    return RepeatStatus.FINISHED;
                })
                .build();
    }

    @Bean
    public Step failOverStep(){
        return stepBuilderFactory.get("nextStep")
                .tasklet((contribution, chunkContext) -> {
                    System.out.println("FlowStepConfig.failOverStep");
                    return RepeatStatus.FINISHED;
                })
                .build();
    }

    @Bean
    public Step processStep(){
        return stepBuilderFactory.get("processStep")
                .tasklet((contribution, chunkContext) -> {
                    System.out.println("FlowStepConfig.processStep");
                    return RepeatStatus.FINISHED;
                })
                .build();
    }


    @Bean
    public Step writeStep(){
        return stepBuilderFactory.get("writeStep")
                .tasklet((contribution, chunkContext) -> {
                    System.out.println("FlowStepConfig.writeStep");
                    return RepeatStatus.FINISHED;
                })
                .build();
    }
}

 

참고 블로그

 

참고 문서 : 스프링 배치 완벽 가이드