10. 스프링배치 플로우 컨트롤 하기

이전까지는 데이터를 읽고 저장하는 순서대로 처리하는 배치 작업에 대해 알아봤다. 이번에는 배치 작업에서 여러 step을 정의하고 조건에 따라 순서대로 실행하거나 특정 step을 건너 뛸 수 있는 Spring Batch의 Flow Controller에 대해 알아보자

 

Flow 컨트롤 방법

• next: 현재 Step이 성공적으로 종료되면 다음 Step으로 이동한다.
• from: 특정 Step에서 현재 Step으로 이동한다.
• on: 특정 ExitStatus에 따라 다음 Step을 결정한다.
• to: 특정 Step으로 이동한다.
• stop: 현재 Flow를 종료한다.
• end: FlowBuilder를 종료한다.

Flow 컨트롤

next

• next는 Start 스텝이 수행하고 난 뒤, next 스텝으로 이동하게 된다.
• next는 계속해서 추가 될 수 있으며, start --> next --> next ... 순으로 진행되도록 한다.

@Slf4j
@Configuration
@RequiredArgsConstructor
public class NextStepTaskJobConfiguration {

	public static final String NEXT_STEP_TASK = "NEXT_STEP_TASK";

	private final PlatformTransactionManager transactionManager;

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

		return new StepBuilder("step01", jobRepository)
			.tasklet((contribution, chunkContext) -> {
				log.info("Execute Step 01 Tasklet ...");
				return RepeatStatus.FINISHED;
			}, transactionManager)
			.build();
	}

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

		return new StepBuilder("step02", jobRepository)
			.tasklet((contribution, chunkContext) -> {
				log.info("Execute Step 02 Tasklet ...");
				return RepeatStatus.FINISHED;
			}, transactionManager)
			.build();
	}

	@Bean
	public Job nextStepJob(@Qualifier("step01") Step step01, @Qualifier("step02") Step step02, JobRepository jobRepository) {
		log.info("------------------ Init myJob -----------------");
		return new JobBuilder(NEXT_STEP_TASK, jobRepository)
			.incrementer(new RunIdIncrementer())
			.start(step01)
			.next(step02)
			.build();
	}
}

 

on

• on 은 특정 스텝의 종료 조건에 따라 어떠한 스텝으로 이동할지 결정할 수 있도록 한다.
• 아래 예는 Step01을 먼저 수행하고, 해당 결과에 따라서 다음 스텝으로 이동하는 플로우를 보여준다.
  • on("FAILED") 인경우 step03을 수행하도록 한다.
  • from(step01).on("COMPLETED") 인경우 step01의 결과 완료인경우라면 step02를 수행하도록 한다.
• 이처럼 on과 from을 통해서 스텝의 종료 조건에 따라 원하는 플로우를 처리할 수 있다.

@Slf4j
@Configuration
@RequiredArgsConstructor
public class OnStepTaskJobConfiguration {

	public static final String ON_STEP_TASK = "ON_STEP_TASK";

	private final PlatformTransactionManager transactionManager;

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

		return new StepBuilder("stepOn01", jobRepository)
			.tasklet((contribution, chunkContext) -> {
				log.info("Execute Step 01 Tasklet ...");

				Random random = new Random();
				int randomValue = random.nextInt(1000);

				if (randomValue % 2 == 0) {
					return RepeatStatus.FINISHED;
				} else {
					throw new RuntimeException("Error This value is Odd: " + randomValue);
				}
			}, transactionManager)
			.build();
	}

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

		return new StepBuilder("stepOn02", jobRepository)
			.tasklet(new Tasklet() {
				@Override
				public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception {
					log.info("Execute Step 02 Tasklet ...");
					return RepeatStatus.FINISHED;
				}
			}, transactionManager)
			.build();
	}

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

		return new StepBuilder("stepOn03", jobRepository)
			.tasklet((contribution, chunkContext) -> {
				log.info("Execute Step 03 Tasklet ...");
				return RepeatStatus.FINISHED;
			}, transactionManager)
			.build();
	}

	@Bean
	public Job onStepJob(@Qualifier("stepOn01") Step stepOn01, @Qualifier("stepOn02") Step stepOn02, @Qualifier("stepOn03") Step stepOn03, JobRepository jobRepository) {
		log.info("------------------ Init myJob -----------------");
		return new JobBuilder(ON_STEP_TASK, jobRepository)
			.incrementer(new RunIdIncrementer())
			.start(stepOn01)
			.on("FAILED").to(stepOn03)
			.from(stepOn01).on("COMPLETED").to(stepOn02)
			.end()
			.build();
	}
}

 

stop

• stop은 특정 step의 작업 결과의 상태를 보고 정지할지 결정한다.
• 아래 예제는 step1의 결과가 실패인경우 stop을 통해 배치 작업을 정지하고 있다.

@Slf4j
@Configuration
@RequiredArgsConstructor
public class StopStepTaskJobConfiguration {

	public static final String STOP_STEP_TASK = "STOP_STEP_TASK";

	private final PlatformTransactionManager transactionManager;

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

		return new StepBuilder("stepStop01", jobRepository)
			.tasklet(new Tasklet() {
				@Override
				public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception {
					log.info("Execute Step 01 Tasklet ...");

					Random random = new Random();
					int randomValue = random.nextInt(1000);

					if (randomValue % 2 == 0) {
						return RepeatStatus.FINISHED;
					} else {
						throw new RuntimeException("Error This value is Odd: " + randomValue);
					}
				}
			}, transactionManager)
			.build();
	}

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

		return new StepBuilder("stepStop02", jobRepository)
			.tasklet(new Tasklet() {
				@Override
				public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception {
					log.info("Execute Step 02 Tasklet ...");
					return RepeatStatus.FINISHED;
				}
			}, transactionManager)
			.build();
	}

	@Bean
	public Job stopStepJob(@Qualifier("stepStop01") Step stepOn01, @Qualifier("stepStop02") Step stepOn02, JobRepository jobRepository) {
		log.info("------------------ Init myJob -----------------");
		return new JobBuilder(STOP_STEP_TASK, jobRepository)
			.incrementer(new RunIdIncrementer())
			.start(stepOn01)
			.on("FAILED").stop()
			.from(stepOn01).on("COMPLETED").to(stepOn02)
			.end()
			.build();
	}
}

 

• Flow Controller를 이용하면 Job의 여러 step을 조건에 따라 수행 할 수 있다.