Step은 실질적인 배치 처리를 정의하고 제어하는 데 필요한 모든 정보가 들어있는 도메인 객체로, Job을 처리하는 실질적인 단위로 쓰인다.(Job:Step = 1:M)
Step은 Job을 구성하는 독립된 작업 단위
순차적으로 배치 처리 수행
Step은 모든 단위 작업의 조각으로 자체적으로 입력, 처리기, 출력을 다룬다.
트랜잭션은 Step 내부에서 이루어짐
org.springframework.batch.core.Step
StepExecution
Step의 실행 정보를 담는 객체로, 각각의 Step이 실행될 때마다 StepExecution이 생성된다.
public class StepExecution extends Entity {
private final JobExecution jobExecution; // 현재 JobExecution 정보
private final String stepName; // Step 이름
private volatile BatchStatus status; // Step의 실행 상태(COMPLETED, STARTING, STARTED ...)
private volatile int readCount; // 성공적으로 읽은 레코드 수
private volatile int writeCount; // 성공적으로 쓴 레코드 수
private volatile int commitCount; // Step의 실행에 대해 커밋된 트랜잭션 수
private volatile int rollbackCount; // Step의 실행에 대해 롤백된 트랜잭션 수
private volatile int readSkipCount; // 읽기에 실패해 건너 띈 렉코드 수
private volatile int processSkipCount; // 프로세스가 실패해 건너 띈 렉코드 수
private volatile int writeSkipCount;// 쓰기에 실패해 건너 띈 렉코드 수
private volatile Date startTime; // Step이 실행된 시간(null == 시작되지 않음)
private volatile Date endTime; // Step의 실행 성공 여부와 관계 없이 끝난 시간
private volatile Date lastUpdated; // 마지막으로 수정된 시간
private volatile ExecutionContext executionContext; // Step 실행 사이에 유지해야하는 사용자 데이터
private volatile ExitStatus exitStatus; // Step 실행 결과에 대한 상태 값(UNKOWN, EXECUTING, COMPLETE, ...)
private volatile boolean terminateOnly; // Job 실행 중지 여부
private volatile int filterCount; // 실행에서 필터링된 레코드 수
private transient volatile List<Throwable> failureExceptions; // Step 실행중 발생한 예외 리스트
...
}
Tasklet 기반
Tasklet은 임의의 Step을 실행할 때 하나의 작업을 처리하는 방식
읽기, 처리, 쓰기로 나뉜 방식이 청크 지향 프로세싱이라면 이를 단일 작업으로 만드는 개념이 Tasklet
public interface Tasklet {
// 내부에 원하는 단일 작업을 구현하고 나면, RepeatStatus.FINISHED 반환후 작업이 계속되면 RepeatStatus.CONTINUABLE 반환
@Nullable
RepeatStatus execute(StepContribution var1, ChunkContext var2) throws Exception;
}
@EnableBatchProcessing
@Configuration
public class SystemCommandConfiguration {
@Autowired
private JobBuilderFactory jobBuilderFactory;
@Autowired
private StepBuilderFactory stepBuilderFactory;
@Bean
public Job systemCommandJob() {
return this.jobBuilderFactory.get("systemCommandJob")
.start(systemCommandStep())
.build();
}
@Bean
public Step systemCommandStep() {
return this.stepBuilderFactory.get("systemCommandStep")
.tasklet(systemCommandTasklet())
.build();
}
@Bean
public SystemCommandTasklet systemCommandTasklet() {
SystemCommandTasklet systemCommandTasklet = new SystemCommandTasklet();
// 명령어
systemCommandTasklet.setCommand("touch tmp.txt");
systemCommandTasklet.setTimeout(5000);
// Job이 비정상적으로 종료될 떄 시스템 프로세스와 관련된 스레드를 강제 종료할지 여부
systemCommandTasklet.setInterruptOnCancel(true);
// 작업 디렉토리 설정
systemCommandTasklet.setWorkingDirectory("/Users/dh0023/Develop/gitbook/TIL");
// 시스템 반환 코드를 스프링 배치 상태 값으로 매핑
systemCommandTasklet.setSystemProcessExitCodeMapper(touchCodeMapper());
// 비동기로 실행하는 시스템 명령을 주기적으로 완료 여부 확인, 완료 여부를 확인하는 주기(default=1초)
systemCommandTasklet.setTerminationCheckInterval(5000);
// 시스템 명령을 실행하는 TaskExecutor 구성 가능
// 문제가 발생하면 락이 발생할 수 있으므로, 동기방식으로 구현하지 않는 것이 좋다.
systemCommandTasklet.setTaskExecutor(new SimpleAsyncTaskExecutor());
// 명령을 실행하기 전에 설정하는 환경 파라미터 목록
systemCommandTasklet.setEnvironmentParams(
new String[]{"BATCH_HOME=/Users/dh0023/Develop/spring/spring-practice/batch-practice"});
return systemCommandTasklet;
}
@Bean
public SimpleSystemProcessExitCodeMapper touchCodeMapper() {
// 반환된 시스템 코드가 0이 ExitStatus.FINISHED
// 0이 아니면 ExitStatus.FAILED
return new SimpleSystemProcessExitCodeMapper();
}
@Bean
public ConfigurableSystemProcessExitCodeMapper configurableSystemProcessExitCodeMapper() {
// 일반적인 구성 방법으로 매핑 구성을 할 수 있음.
ConfigurableSystemProcessExitCodeMapper mapper = new ConfigurableSystemProcessExitCodeMapper();
Map<Object, ExitStatus> mappings = new HashMap<Object, ExitStatus>() {
{
put(0, ExitStatus.COMPLETED);
put(1, ExitStatus.FAILED);
put(2, ExitStatus.EXECUTING);
put(3, ExitStatus.NOOP);
put(4, ExitStatus.UNKNOWN);
put(ConfigurableSystemProcessExitCodeMapper.ELSE_KEY, ExitStatus.UNKNOWN);
}
};
mapper.setMappings(mappings);
return mapper;
}
}
SimpleSystemProcessExitCodeMapper
public class SimpleSystemProcessExitCodeMapper implements SystemProcessExitCodeMapper {
@Override
public ExitStatus getExitStatus(int exitCode) {
if (exitCode == 0) {
return ExitStatus.COMPLETED;
} else {
return ExitStatus.FAILED;
}
}
}
Chunk 기반
Chunk란 아이템이 트랜잭션에 commit되는 수를 말한다.
즉, 청크 지향 처리란 한 번에 하나씩 데이터를 읽어 Chunk라는 덩어리를 만든 뒤, Chunk 단위로 트랜잭션을 다루는 것을 의미한다.
Chunk 지향 프로세싱은 1000개의 데이터에 대해 배치 로직을 실행한다고 가정하면, Chunk 단위로 나누지 않았을 경우에는 한개만 실패해도 성공한 999개의 데이터가 롤백된다. Chunk 단위를 10으로 한다면, 작업 중에 다른 Chunk는 영향을 받지 않는다.
이때, Chunk는 커밋 간격(commit interval)에 의해 정의되고 수행하므로, 커밋 간격에 따라 성능이 달라질 수 있다. 최상의 성능을 얻기 위해서는 커밋 간격 설정이 중요하다.
ItemReader, ItemProcessor, ItemWriter 3단계로 비지니스 로직을 분리해 역할을 명확하게 분리할 수 있다.
비즈니스 로직 분리
읽어온 배치 데이터와 쓰여질 데이터 타입이 다른 경우에 대한 대응
각 Chunk는 자체 트랜잭션으로 실행되며, 처리에 실패하면 성공한 트랜잭션 이후부터 다시 시작 가능
그러므로 읽어온 배치의 데이터와 저장할 데이터 타입이 다른 경우에 대응할 수 있다.
ItemReader
Step의 대상이 되는 배치 데이터(File, Xml, DB 등)를 읽어오는 인터페이스
org.springframework.batch.item.ItemReader<T>
public interface ItemReader<T> {
// read 메서드의 반환 타입을 T(제너릭)으로 구현하여 직접 타입을 지정할 수 있음
@Nullable
T read() throws Exception, UnexpectedInputException, ParseException, NonTransientResourceException;
}
ItemProcessor
ItemReader로 읽어 온 배치 데이터를 변환하는 역할을 수행
ItemProcessor는 로직 처리만 수행하여 역할을 분리하고, 명확한 input/output을 ItemProcessor로 구현해놓으면 더 직관적인 코드가 될 것이다.
org.springframework.batch.item.ItemProcessor<T>
public interface ItemProcessor<I, O> {
@Nullable
O process(@NonNull I var1) throws Exception;
}
ItemWriter
배치 데이터(DB, File 등)를 저장한다.
org.springframework.batch.item.ItemWriter<T>
public interface ItemWriter<T> {
// T(제네릭)으로 지정한 타입을 List 매개변수로 받는다.
void write(List<? extends T> var1) throws Exception;
}
리스트의 데이터 수는 설정한 *청크(Chunk) 단위로 불러온다.
청크 기반 Job 예시
@EnableBatchProcessing
@Configuration
public class ChunkBasedConfiguration {
@Autowired
private JobBuilderFactory jobBuilderFactory;
@Autowired
private StepBuilderFactory stepBuilderFactory;
/**
* 실제 스프링 배치 Job 생성
*/
@Bean
public Job chunkBasedJob() {
return this.jobBuilderFactory.get("chunkBasedJob")
.start(chunkStep())
.build();
}
@Bean
public Step chunkStep() {
return this.stepBuilderFactory.get("chunkStep")
.<String, String> chunk(1000) // chunk 기반, 커밋간격 1000
.reader(itemReader())
.writer(itemWriter())
.build();
}
@Bean
public ListItemReader<String> itemReader(){
List<String> items = new ArrayList<>(100000);
for (int i = 0; i < 100000; i++) {
items.add(UUID.randomUUID().toString());
}
return new ListItemReader<>(items);
}
@Bean
public ItemWriter<String> itemWriter() {
return items -> {
for (String item : items) {
System.out.println(">> current item = " + item);
}
System.out.println(">> end itemWriter chunk " + items.size());
};
}
}
일반적으로는 위와 같이 커밋간격을 하드 코딩해 크기를 정의하지만, 크기가 동일하지 않은 청크를 처리해야하는 경우도 있다. 스프링 배치는 org.springframework.batch.repeat.CompletionPolicy 인터페이스를 제공해 청크가 완료되는 시점을 정의할 수 있도록 제공해준다.
CompletionPolicy
청크 완료 여부를 결정할 수 있는 결정로직을 구현할 수 있는 인터페이스로, CompletionPolicy 인터페이스의 구현체에 대해서 알아 볼 것이다.
package org.springframework.batch.repeat;
public interface CompletionPolicy {
// 청크 완료 여부의 상태를 기반으로 결정 로직 수행
boolean isComplete(RepeatContext context, RepeatStatus result);
// 내부 상태를 이용해 청크 완료 여부 판단
boolean isComplete(RepeatContext context);
// 청크의 시작을 알 수 있도록 정책을 초기화
RepeatContext start(RepeatContext parent);
// 각 item이 처리가되면 update 메서드가 호출되면서 내부 상태 갱신
void update(RepeatContext context);
}
직접 구현하는 방법
CompletionPolicy를 구현하여 필수 메서드들을 각각 알맞게 로직을 구성하면된다.
public class RandomChunkSizePolicy implements CompletionPolicy {
private int chunksize;
private int totalProcessed;
private Random random = new Random();
// 청크 완료 여부의 상태를 기반으로 결정 로직 수행
@Override
public boolean isComplete(RepeatContext context, RepeatStatus result) {
if (RepeatStatus.FINISHED == result) {
return true;
} else {
return isComplete(context);
}
}
// 내부 상태를 이용해 청크 완료 여부 판단
@Override
public boolean isComplete(RepeatContext context) {
return this.totalProcessed >= chunksize;
}
// 청크의 시작을 알 수 있도록 정책을 초기화
@Override
public RepeatContext start(RepeatContext parent) {
this.chunksize = random.nextInt(20);
this.totalProcessed = 0;
System.out.println("chunk size has been set to => " + this.chunksize);
return parent;
}
// 각 item이 처리가되면 update 메서드가 호출되면서 내부 상태 갱신
@Override
public void update(RepeatContext context) {
this.totalProcessed++;
}
}
chunk size has been set to => 5
>> current item = b784cda0-961f-4faf-9737-4334e774f0d1
>> current item = 9fc3ec22-5d54-4632-94e3-fcc28cc00260
>> current item = 53452739-0122-4f8b-a52a-667e37cbf908
>> current item = de9ad8ec-1bf7-40d4-b991-72b6699594f9
>> current item = 352104aa-ae63-4928-96e2-460b3b145d43
>> end itemWriter chunk 5
chunk size has been set to => 10
>> current item = 4116f3e9-cad3-4387-8f46-a871d825d7d9
>> current item = 3e1f58b7-a1fa-45bd-b336-2484d1d543f3
>> current item = a58e2401-37e8-402a-8ac3-daddaf2e91a0
>> current item = 01162200-fc7d-4580-8533-d6deab7f3c65
>> current item = 31280b14-49bf-4a03-b1d4-210585e4da40
>> current item = 91b7d269-8105-41dc-ae6a-1d682913c168
>> current item = abeead5f-751b-4862-aa37-4f32eb09439a
>> current item = b6bf1a92-56e8-433a-a0ea-935e101cad6f
>> current item = 84af0778-d6e1-4aef-bd8f-347b047bb276
>> current item = b86477f3-b997-4b04-a3ca-27f517c9170f
>> end itemWriter chunk 10
chunk size has been set to => 11
다음과 같이 Random으로 지정된 수만큼 chunk가 수행되는 것을 확인 할 수 있다.
SimpleCompletionPolicy
가장 기본적인 구현체로, 미리 구성해둔 임곗값에 도달하면 청크 완료로 표시한다.
public class SimpleCompletionPolicy extends DefaultResultCompletionPolicy {
public static final int DEFAULT_CHUNK_SIZE = 5;
int chunkSize = 0;
public SimpleCompletionPolicy() {
this(DEFAULT_CHUNK_SIZE);
}
public SimpleCompletionPolicy(int chunkSize) {
super();
this.chunkSize = chunkSize;
}
@Bean
public Step chunkStep() {
return this.stepBuilderFactory.get("chunkStep")
.<String, String> chunk(completionPolicy()) // completePolicy 호출
.reader(itemReader())
.writer(itemWriter())
.build();
}
@Bean
public CompletionPolicy completionPolicy() {
// 처리된 ITEM 개수를 세어, 이 개수가 임계값에 도달하면 chunk 완료로 표시
SimpleCompletionPolicy simpleCompletionPolicy = new SimpleCompletionPolicy(1000);
return simpleCompletionPolicy;
}
TimeoutTerminationPolicy
타임아웃 값을 구성해, 청크 내에서 처리 시간이 지정한 시간이 넘으면 청크가 완료된 것으로 간주하고, 모든 트랜잭션 처리를 정상적으로 한다는 것이다. TimeoutTerminationPolicy 만으로 청크 완료 시점을 결정하는 경우는 거의 존재하지 않으며, CompositeCompletionPolicy의 일부로 사용하는 경우가 많다.
public class TimeoutTerminationPolicy extends CompletionPolicySupport {
/**
* Default timeout value in milliseconds (the value equivalent to 30 seconds).
*/
public static final long DEFAULT_TIMEOUT = 30000L;
private long timeout = DEFAULT_TIMEOUT;
/**
* Default constructor.
*/
public TimeoutTerminationPolicy() {
super();
}
/**
* Construct a {@link TimeoutTerminationPolicy} with the specified timeout
* value (in milliseconds).
*
* @param timeout duration of the timeout.
*/
public TimeoutTerminationPolicy(long timeout) {
super();
this.timeout = timeout;
}
@Bean
public CompletionPolicy timeoutCompletionPolicy() {
TimeoutTerminationPolicy timeoutTerminationPolicy = new TimeoutTerminationPolicy(3);
return timeoutTerminationPolicy;
}
TimeoutTerminationPolicy로 수행한 경우 각 chunk 단위를 확인해보면 다음과 같이 제각각인 것을 볼 수있다.
>> end itemWriter chunk 795
>> end itemWriter chunk 679
>> end itemWriter chunk 841
>> end itemWriter chunk 1153
>> end itemWriter chunk 1061
>> end itemWriter chunk 1916
>> end itemWriter chunk 1667
>> end itemWriter chunk 1719
>> end itemWriter chunk 931
>> end itemWriter chunk 1634
>> end itemWriter chunk 941
>> end itemWriter chunk 667
>> end itemWriter chunk 665
>> end itemWriter chunk 547
>> end itemWriter chunk 533
>> end itemWriter chunk 973
>> end itemWriter chunk 647
>> end itemWriter chunk 1632
>> end itemWriter chunk 676
CompositeCompletionPolicy
CompositeCompletionPolicy는 청크 완료 여부를 결정하는 여러 정책을 함께 구성할 수 있다. 포함하고 있는 여러 정책 중 하나라도 청크 완료라고 판단되면 해당 청크가 완료된 것으로 표시한다.
@Bean
public CompletionPolicy compositeCompletionPolicy() {
CompositeCompletionPolicy policy = new CompositeCompletionPolicy();
// 여러 정책 설정
policy.setPolicies(
new CompletionPolicy[]{
new TimeoutTerminationPolicy(3),
new SimpleCompletionPolicy(1000)
}
);
return policy;
}
다음과 같이 수행한 경우에는 chunk 단위가 1000개를 넘어선 경우가 없는 것을 확인할 수 있다.
>> end itemWriter chunk 731
>> end itemWriter chunk 1000
>> end itemWriter chunk 690
>> end itemWriter chunk 1000
>> end itemWriter chunk 798
>> end itemWriter chunk 1000
>> end itemWriter chunk 980
>> end itemWriter chunk 838
>> end itemWriter chunk 850
>> end itemWriter chunk 1000
>> end itemWriter chunk 263
>> end itemWriter chunk 556
>> end itemWriter chunk 629
>> end itemWriter chunk 962
>> end itemWriter chunk 960
>> end itemWriter chunk 1000
>> end itemWriter chunk 1000
>> end itemWriter chunk 1000
>> end itemWriter chunk 898
>> end itemWriter chunk 1000
>> end itemWriter chunk 900
>> end itemWriter chunk 798
>> end itemWriter chunk 215
Step Listener
스탭과 청크의 시작과 끝에서 특정 로직을 처리할 수 있게 해준다.
(StepListener는 모든 스탭 리스너가 상속하는 마커 인터페이스이다.)
모든 수준에 리스너를 적용해 Job을 중단할 수 있으며, 일반적으로 전처리를 수행하거나 이후 결과를 평가하거나, 일부 오류처리에도 사용된다.
Job 실행 정보(jobExecution)와 스탭 실행정보( stepExecution)를 인자로 받아 모든 정보를 이용해 다음에 무엇을 수행할지에 대해 결정할 수 있다.
public interface JobExecutionDecider {
/**
* Strategy for branching an execution based on the state of an ongoing
* {@link JobExecution}. The return value will be used as a status to
* determine the next step in the job.
*
* @param jobExecution a job execution
* @param stepExecution the latest step execution (may be {@code null})
* @return the exit status code
*/
FlowExecutionStatus decide(JobExecution jobExecution, @Nullable StepExecution stepExecution);
}
public class RandomDecider implements JobExecutionDecider {
private Random random = new Random();
@Override
public FlowExecutionStatus decide(JobExecution jobExecution, StepExecution stepExecution) {
if (random.nextBoolean()) {
return new FlowExecutionStatus(FlowExecutionStatus.COMPLETED.getName());
} else {
return new FlowExecutionStatus(FlowExecutionStatus.FAILED.getName());
}
}
}
스프링 배치 처리가 성공적으로 종료됐음을 의미
JobInstance가 Completed로 종료되면 동일한 파라미터를 사용해 다시 실행할 수 없다.
Failed(실패)
잡이 성공적으로 완료되지 않았음을 의미
Failed 상태로 종료된 잡은 스프링 배치를 사용해 동일한 파라미터로 다시 실행할 수 있다.
Stopped(중지)
Stopped 상태로 종료된 잡은 다시 수행 가능하다.
Job에 오류가 발생하지 않았어도, 중단된 위치에서 잡을 다시 시작할 수 있다.
사람의 개입이 필요하거나 다른 검사/처리가 필요한 상황에 유용하다.
BatchStatus를 판별할 때, ExitStatus를 평가하면서 식별된다. ExitStatus는 스텝, 청크, 잡에서 반환될 수 있으며, BatchStatus는 StepExecution 이나 JobExecution 내에 보관되며, JobRepository에 저장된다.
