Listener

Spring Batch에서는 메인 로직 외에 구간 사이사이에 어떤 일을 처리하고자 할 때 Listener를 사용한다. 예를 들면, Job, Step, Chunk, ItemReader, ItemProcessor, ItemWriter의 실행 직전과 직후에 어떤 행위를 할지 정의할 수 있다.

 

Listener의 종류

• JobExecutionListener
• StepExecutionListener
• ChunkListener
• ItemReadListener
• ItemProcessListener
• ItemWriteListener

 

JobExecutionListener

public interface JobExecutionListener {
    default void beforeJob(JobExecution jobExecution) {
    }

    default void afterJob(JobExecution jobExecution) {
    }
}

• Job의 실행 전에 실행하는 beforeJob.
• Job의 실행 후에 실행하는 afterJob.
• 두 메서드 모두 인자로 JobExecution을 넘겨준다.
• 구현한 JobExecutionListener는 아래처럼 JobBuilder에서 Job을 만들 때 listener()에 넣어주면 된다.

 

나만의 JobListener

package cwchoiit.springbatchmaster.tasklet;

import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.JobExecutionListener;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.stereotype.Component;

import java.util.HashMap;
import java.util.Map;

@Component
public class JobListener implements JobExecutionListener {

    @Override
    public void beforeJob(JobExecution jobExecution) {
        ExecutionContext executionContext = jobExecution.getExecutionContext();
        Map<String, Object> executionContextMap = new HashMap<>();
        executionContextMap.put("name", "홍길동");
        executionContextMap.put("age", 30);
        jobExecution.setExecutionContext(new ExecutionContext(executionContextMap));
    }

    @Override
    public void afterJob(JobExecution jobExecution) {
        System.out.println(jobExecution.getJobId());
    }
}

listener 등록

@Bean
public Job reserveRestaurantJob(JobRepository jobRepository,
                                JobListener jobListener,
                                Step searchAvailableKoreanRestaurantStep,
                                Step reserveRestaurantStep,
                                Step sendDepositStep) {
    return new JobBuilder("reserveRestaurantJob", jobRepository)
            .listener(jobListener)
            .start(searchAvailableKoreanRestaurantStep)
            .next(reserveRestaurantStep)
            .next(sendDepositStep)
            .build();
}

 

StepExecutionListener

public interface StepExecutionListener extends StepListener {
    default void beforeStep(StepExecution stepExecution) {
    }

    @Nullable
    default ExitStatus afterStep(StepExecution stepExecution) {
        return null;
    }
}

• Step의 실행 전 실행하는 beforeStep
• Step의 실행 후 실행하는 afterStep
• 두 메서드 모두 인자로 StepExecution을 넘겨준다.
• 구현된 StepExecutionListener는 아래처럼 StepBuilder에서 Step을 만들 때 listener()에 넣어주면 된다.

나만의 StepListener

package cwchoiit.springbatchmaster.tasklet;

import org.springframework.batch.core.ExitStatus;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.StepExecutionListener;
import org.springframework.stereotype.Component;

@Component
public class StepListener implements StepExecutionListener {

    @Override
    public void beforeStep(StepExecution stepExecution) {
        StepExecutionListener.super.beforeStep(stepExecution);
    }

    @Override
    public ExitStatus afterStep(StepExecution stepExecution) {
        return StepExecutionListener.super.afterStep(stepExecution);
    }
}

listener 등록

@Bean
@JobScope
public Step sampleStep(JobRepository jobRepository,
                       StepListener stepListener,
                       PlatformTransactionManager platformTransactionManager) {
    return new StepBuilder("sampleStep", jobRepository)
            .tasklet(sampleTasklet(), platformTransactionManager)
            .listener(stepListener)
            .build();
}

 

 

ChunkListener

public interface ChunkListener extends StepListener {
    String ROLLBACK_EXCEPTION_KEY = "sb_rollback_exception";

    default void beforeChunk(ChunkContext context) {
    }

    default void afterChunk(ChunkContext context) {
    }

    default void afterChunkError(ChunkContext context) {
    }
}

• 맥락은 위 Listener들과 똑같다. 다만, 여기서는 Chunk를 했을 때 에러가 발생하면 호출되는 메서드도 있다. 

ItemReadListener

public interface ItemReadListener<T> extends StepListener {
    default void beforeRead() {
    }

    default void afterRead(T item) {
    }

    default void onReadError(Exception ex) {
    }
}

• 마찬가지다.

ItemProcessListener

public interface ItemProcessListener<T, S> extends StepListener {
    default void beforeProcess(T item) {
    }

    default void afterProcess(T item, @Nullable S result) {
    }

    default void onProcessError(T item, Exception e) {
    }
}

ItemWriteListener

public interface ItemWriteListener<S> extends StepListener {
    default void beforeWrite(Chunk<? extends S> items) {
    }

    default void afterWrite(Chunk<? extends S> items) {
    }

    default void onWriteError(Exception exception, Chunk<? extends S> items) {
    }
}

 

 

정리를 하자면

이번 포스팅에서는, 리스너들을 알아보았다. 각 작업 전후로 어떤 행위가 필요할때, 또는 Chunk, Read, Process, Write의 경우 에러가 발생했을 때 어떤 행위가 필요하면 정의하면 좋을듯하다.