Spring Batch (8) JobLauncher

📎 JobLauncher

지난 시간에는 JobRepository에 대해 살펴보았습니다.

JobRepository는 Spring Batch 작업을 진행하면서 JobInstance, JobExecution, StepExeuction 등 Batch와 관련된 도메인의 CRUD 처리를 하는 역할을 합니다.

 

이번 시간에는 Spring Batch Job의 실행 주체인 JobLauncher에 대해 살펴보겠습니다.

 

 

📌 JobLauncher

기본 개념

• Spring Batch Job을 실행시키는 역할
• Job & JobParameters를 인자로 받으며 요청된 배치 작업을 수행한 후 JobExecution을 반환
• Spring Batch에서는 BatchAutoConfiguration 클래스 내에 존재하는 JobLauncherApplicationRunner 클래스가 자동으로 JobLauncher을 실행합니다.

76 라인에서 JobLauncherApplicationRunner 객체 생성

JobLauncherApplicationRunner 클래스 - 199라인의 run() 메서드를 통해 Job 실행

Spring Batch Job은 SyncTaskExecutor(동기), SipmleAsyncTaskExecutor(비동기), 커스텀 구현체 등의 설정을 통해 동기 및 비동기 실행이 가능합니다.

 

 

 

📌 JobLauncher 동작 과정

@RequiredArgsConstructor
@Configuration
public class JobLauncherConfiguration {
 
    private final JobBuilderFactory jobBuilderFactory;
    private final StepBuilderFactory stepBuilderFactory;
 
    @Bean
    public Job jobLauncherJob() {
        return this.jobBuilderFactory.get("jobLauncherJob")
                .start(jobLauncherStep1())
                .next(jobLauncherStep2())
                .incrementer(new RunIdIncrementer())
                .build();
    }
 
    @Bean
    public Step jobLauncherStep1() {
        return stepBuilderFactory.get("jobLauncherStep1")
                .tasklet(new Tasklet() {
                    @Override
                    public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception {
                        Thread.sleep(3000);
                        return RepeatStatus.FINISHED;
                    }
                })
                .build();
    }
 
    @Bean
    public Step jobLauncherStep2() {
        return stepBuilderFactory.get("jobLauncherStep2")
                .tasklet((contribution, chunkContext) -> null)
                .build();
    }
}
 

위의 간단한 Job, Step 코드를 통해 JobLauncher에 대해 살펴보겠습니다.

 

 

Job을 실행시키는 JobLauncherApplicationRunner 객체는 BatchAutoConfiguration 클래스에서 생성이 됩니다.

위 사진은 JobLauncherApplicationRunner 클래스에서 Job이 실행되는 과정입니다.

run() 메서드에서 args 인자는 파라미터로 받은 변수가 존재하면 해당 값을 설정하는 역할을 합니다.

 

 

launchJobFromProperties() 메서드가 호출하는 executeLocalJobs() 메서드의 로직입니다.

 

위 코드에서 for 문 내부에 존재하는 this.jobs는 애플리케이션 코드 전체에 존재하는 Job을 나타내고 있습니다.

저는 위와 같이 이전 예제 코드도 전부 포함되어 있기에 this.jobs의 size가 8이고, 해당 job 중에서 현재 실행한 job의 name과 동일한지 비교하여 동일할 경우 execute(job, jobParameters) 메서드를 호출합니다.

 

 

execute() 메서드 내에서는 JobParameters 객체를 생성하고, jobLauncher.run() 메서드를 호출하여 Job을 실행시킵니다.

 

 

위 run() 메서드를 호출할 때 JobLauncher 인터페이스의 구현체인 SimpleJobLauncher 클래스에서 run() 메서드를 오버라이드 하고 있습니다.

 

위 코드에서 JobRepository에서 jobName + jobKey 조합으로 JobInstance, JobExecution을 조회하기 때문에 lastExecution은 null로 리턴하게 됩니다.

 

 

 

SimpleJobLauncher 클래스의 run() 전체 메서드를 살펴보면 다음과 같습니다.

    /**
     * Run the provided job with the given {@link JobParameters}. The
     * {@link JobParameters} will be used to determine if this is an execution
     * of an existing job instance, or if a new one should be created.
     *
     * @param job the job to be run.
     * @param jobParameters the {@link JobParameters} for this particular
     * execution.
     * @return the {@link JobExecution} if it returns synchronously. If the
     * implementation is asynchronous, the status might well be unknown.
     * @throws JobExecutionAlreadyRunningException if the JobInstance already
     * exists and has an execution already running.
     * @throws JobRestartException if the execution would be a re-start, but a
     * re-start is either not allowed or not needed.
     * @throws JobInstanceAlreadyCompleteException if this instance has already
     * completed successfully
     * @throws JobParametersInvalidException thrown if jobParameters is invalid.
     */
    @Override
    public JobExecution run(final Job job, final JobParameters jobParameters)
            throws JobExecutionAlreadyRunningException, JobRestartException, JobInstanceAlreadyCompleteException,
            JobParametersInvalidException {
 
        Assert.notNull(job, "The Job must not be null.");
        Assert.notNull(jobParameters, "The JobParameters must not be null.");
 
        final JobExecution jobExecution;
        JobExecution lastExecution = jobRepository.getLastJobExecution(job.getName(), jobParameters);
        if (lastExecution != null) {
            if (!job.isRestartable()) {
                throw new JobRestartException("JobInstance already exists and is not restartable");
            }
            /*
             * validate here if it has stepExecutions that are UNKNOWN, STARTING, STARTED and STOPPING
             * retrieve the previous execution and check
             */
            for (StepExecution execution : lastExecution.getStepExecutions()) {
                BatchStatus status = execution.getStatus();
                if (status.isRunning() || status == BatchStatus.STOPPING) {
                    throw new JobExecutionAlreadyRunningException("A job execution for this job is already running: "
                            + lastExecution);
                } else if (status == BatchStatus.UNKNOWN) {
                    throw new JobRestartException(
                            "Cannot restart step [" + execution.getStepName() + "] from UNKNOWN status. "
                                + "The last execution ended with a failure that could not be rolled back, "
                                + "so it may be dangerous to proceed. Manual intervention is probably necessary.");
                }
            }
        }
 
        // Check the validity of the parameters before doing creating anything
        // in the repository...
        job.getJobParametersValidator().validate(jobParameters);
 
        /*
         * There is a very small probability that a non-restartable job can be
         * restarted, but only if another process or thread manages to launch
         * <i>and</i> fail a job execution for this instance between the last
         * assertion and the next method returning successfully.
         */
        jobExecution = jobRepository.createJobExecution(job.getName(), jobParameters);
 
        try {
            taskExecutor.execute(new Runnable() {
 
                @Override
                public void run() {
                    try {
                        if (logger.isInfoEnabled()) {
                            logger.info("Job: [" + job + "] launched with the following parameters: [" + jobParameters
                                    + "]");
                        }
                        job.execute(jobExecution);
                        if (logger.isInfoEnabled()) {
                            Duration jobExecutionDuration = BatchMetrics.calculateDuration(jobExecution.getStartTime(), jobExecution.getEndTime());
                            logger.info("Job: [" + job + "] completed with the following parameters: [" + jobParameters
                                    + "] and the following status: [" + jobExecution.getStatus() + "]"
                                    + (jobExecutionDuration == null ? "" : " in " + BatchMetrics.formatDuration(jobExecutionDuration)));
                        }
                    }
                    catch (Throwable t) {
                        if (logger.isInfoEnabled()) {
                            logger.info("Job: [" + job
                                    + "] failed unexpectedly and fatally with the following parameters: [" + jobParameters
                                    + "]", t);
                        }
                        rethrow(t);
                    }
                }
 
                private void rethrow(Throwable t) {
                    if (t instanceof RuntimeException) {
                        throw (RuntimeException) t;
                    }
                    else if (t instanceof Error) {
                        throw (Error) t;
                    }
                    throw new IllegalStateException(t);
                }
            });
        }
        catch (TaskRejectedException e) {
            jobExecution.upgradeStatus(BatchStatus.FAILED);
            if (jobExecution.getExitStatus().equals(ExitStatus.UNKNOWN)) {
                jobExecution.setExitStatus(ExitStatus.FAILED.addExitDescription(e));
            }
            jobRepository.update(jobExecution);
        }
 
        return jobExecution;
    }

 

 

실제 Job을 실행하기 위해 사용되는 인터페이스는 TaskExecutor 이고, 익명클래스로 구현하여 동기 실행을 하고 있습니다.

 

만약 Job을 비동기로 실행하려면 setTaskExecutor() 메서드를 통해 직접 설정해주면 비동기 실행이 가능합니다.

 

 

참고 문서

• https://docs.spring.io/spring-batch/docs/current/reference/html/index-single.html
• https://www.inflearn.com/course/%EC%8A%A4%ED%94%84%EB%A7%81-%EB%B0%B0%EC%B9%98/dashboard