로그인 선택
미리보기
커뮤니티
23.06.15
글 등록
DEVOTEE를 활성화 시키면
지금 작성한 커뮤니티 글에 대해 1개의 댓글을 달아줍니다.
버튼을 누르면 글 수정 시 ChatGPT가 작성한 댓글이 수정됩니다.
임시저장함
최대 20건까지 저장합니다.
| 컨텐츠 유형 | 제목 | 저장일 | 삭제 |
|---|
본인인증
고객님의 소중한 개인정보보호를 위해
본인인증을 진행해주세요. 본인인증 미 진행 시 로그인이 제한됩니다.
본인인증 로그인에 실패하였습니다.
회원이 아니시거나 본인인증 등록이
완료되지 않은 사용자입니다.
회원정보 연결
Spring Batch 는 어떻게 Chunk 지향처리를 하고, Transaction을 언제 걸까
Spring Batch Chunk Oriented & Transaction
맨날 StepBuilder.java 나 JobBuilder.java 로 기계적으로 job 을 만들다 보면, 너무나 많이 추상화 된 프레임워크에 의해 바보가 되는 느낌을 자주 받습니다.
작업을 하다 이 프레임워크는 어느 타이밍에 Transaction 을 걸고, 어떻게 Chunk 지향처리를 하는지 궁금해졌습니다.
Spring Batch 는 chunk 단위로 데이터 처리를 합니다.
Job - Step - Tasklet - Reader + Process + Writer 계층을 이루고 있는 점.
그리고 Tasklet 내 chunk 단위 처리 중 에러가 발생하면 롤백 한다는 내용은 많은이가 알고 있습니다.
일단 jobLauncher 가 Job 을 시작하는건 알겠는데 어떻게 R + P + W 순으로 동작을 하는걸까 (이하 RPW)
그리고 분명히 spring-tx 의 TransactionManager을 이용할거고, 트랜잭션 처리를 할 것 인데 어느 타이밍에 트랜잭션이 동작하는지 모르겠다는 의문이 들었습니다.
writer 에 쓰기 직전에 transaction 을 실시할까? 그러면 커스텀로직 막 넣어서 processor 에서 뭔가 insert 한건 롤백이 안될 것 같은데...
그럼 read 할 때부터 transaction 을 거는건가? 뭣하러 그때부터 걸지?
writer 를 Composite Writer 를 써서 멀티 database 에 writer 를 하는 경우는 무엇을 해야할까
등등의 의문을 가졌습니다.
이걸 확인하기 위해선 크게 두 줄기로 파봤어야합니다.
1. Job 은 어떻게 만들어지는가 - Chunk Oriented 구현 방식에 관한 이야기
뭔가 writer 의 write() 를 호출하기 전후로 트랜잭션 시작, Commit 코드가 있을것 같은 느낌을 받았습니다.
그래서 Job, Step 구성하는 소스를 열어볼 예정입니다.
@Configuration
@RequiredArgsConstructor
public class JobConfig {
private final JobBuilderFactory jobBuilderFactory;
private final StepBuilderFactory stepBuilderFactory;
@Bean
public Job sampleJob(Step sampleStep, JobExecutionListener jobExecutionListener) {
return jobBuilderFactory.get(JOB_NAME)
.start(sampleStep)
.listener(jobExecutionListener)
.preventRestart()
.build();
}
@Bean
@JobScope
public Step sampleStep(ItemReader<BeforeDomain> reader,
ItemProcessor<BeforeDomain, AfterDomain> processor,
CompositeItemWriter<AfterDomain> writer) {
return stepBuilderFactory.get(JOB_NAME.concat("Step"))
.<BeforeDomain, AfterDomain>chunk(CHUNK_SIZE)
.reader(reader)//user_reward
.processor(processor)
.writer(writer)
.build();
}
}흔한 Job 설정이나 Job 을 만들 떄 쓰는
JobBuilderFactory,StepBuilderFactory를 보겠습니다.이 Bean 들은 boot 기준
@EnableBatchProcessing활성화 시, SimpleBatchConfiguration.java 를 통해 자동으로 생성됩니다.StepBuilderFactory.java 를 살펴보면, transactionManager 를 주입받아놓고, get 할 때 같이 넣어줘서 만드는 것을 알 수 있습니다. 원한다면
.transactionManager()빌더 메서드를 통해 기본 말고 원하는 것을 집어넣을 수 있죠.
public StepBuilderFactory(JobRepository jobRepository, PlatformTransactionManager transactionManager) {
this.jobRepository = jobRepository;
this.transactionManager = transactionManager;
}
public StepBuilder get(String name) {
StepBuilder builder = new StepBuilder(name).repository(jobRepository).transactionManager(
transactionManager);
return builder;
}StepBuilder 가 transactionManager 를 넣어준다는건 알았으니, 뭔가 트랜잭션 관리를 잘 해줄거라는건 보이긴하고,
그럼 RPW 가 실제로 어떻게 구성이 되는지 알아보기 위해선 조금 더 StepBuilder 를 파봐야합니다.
StepBuilder.build() 에선, 어느 순간
createTasklet()이라는걸 호출하게됩니다.결국 RPW 도 하나의 Tasklet 으로 구성이 되는거고, Step 은 Tasklet 을 품고 있는 우리가 아는 구조가 되나봅니다.
xml 설정으로 스프링 배치를 개발해보신 분들에겐 당연할 수 있습니다. RPW 는 tasklet 안에 들어있거든요.
<batch:job id="firstBatchJob">
<batch:step id="step1">
<batch:tasklet>
<batch:chunk reader="itemReader" writer="itemWriter" processor="itemProcessor" commit-interval="10">
</batch:chunk>
</batch:tasklet>
</batch:step>
</batch:job> @Override
protected Tasklet createTasklet() {
...
SimpleChunkProvider<I> chunkProvider = new SimpleChunkProvider<>(getReader(), repeatOperations);
SimpleChunkProcessor<I, O> chunkProcessor = new SimpleChunkProcessor<>(getProcessor(), getWriter());
ChunkOrientedTasklet<I> tasklet = new ChunkOrientedTasklet<>(chunkProvider, chunkProcessor);
...
return tasklet;
}다 생략하고 chunkProvider 에 reader 를 넣고, chunkProcessor 에 processor + writer 를 집어넣어
ChunkOrientedTasklet.java를 만들게 됩니다.결국 이 녀석이 핵심인것 같습니다. 내부 execute 코드를 살펴봐야했는데 chunkProvider 의 실제 구현체 SimpleChunkProvider.java 를 보면 우리가 아는 Reader 와 크게 다르지 않습니다.
그런데 SimpleChunkProcessor 는 Processor + Writer 가 합체를 했습니다.
//ChunkOrientedTasklet.java
@Override
public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception {
...
// 처리해야할 데이터
Chunk<I> inputs = (Chunk<I>) chunkContext.getAttribute(INPUTS_KEY);
if (inputs == null) {
inputs = chunkProvider.provide(contribution); //더이상 인풋이 없을때까지 될 때까지 계속 읽는 영역
if (buffering) {
chunkContext.setAttribute(INPUTS_KEY, inputs);
}
}
//SimpleChunkProcessor
chunkProcessor.process(contribution, inputs); //process + writer
chunkProvider.postProcess(contribution, inputs);
...
}
//SimpleChunkProcessor.java
@Override
public final void process(StepContribution contribution, Chunk<I> inputs) throws Exception {
initializeUserData(inputs);
if (isComplete(inputs)) {
return;
}
// Processor 가 지지고볶음
Chunk<O> outputs = transform(contribution, inputs);
contribution.incrementFilterCount(getFilterCount(inputs, outputs));
// Writer 가 지지고볶음
write(contribution, inputs, getAdjustedOutputs(inputs, outputs));
}대충 정리해보면
열심히 읽어 Chunk Size 를 Reader에서 맞춘다.
ChunkSize 다 차면 Processor 로 하나씩 process 한다.
ChunkSize 처리를 다 하고나면 writer 에서 한방에 처리한다.
최신 spring batch reference 에 가니 정확하게 알려주네요.
하지만 꽤 많은 사람들이 잘못 알고 있는 이미지가 있습니다. 아래 이미지인데요, spring-batch 4.0.4 버전 때만해도 reference 가 잘못되어 있었습니다. *다 쓰고 찾아보니 관련 이슈가 있었고, 제법 최근에 수정되었네요.
이로써 spring batch 는 ChunkOrientedTasklet.java 를 유연하게 만들어서 주입받은 RPW 를 통해 chunk 지향처리를 진행함을 알 수 있었습니다.
하지만 정작 궁금했던 Transaction 이 걸리는 타이밍에 관해서는 더 들어가도 위 코드 분석으로는 알 수 없었습니다. 어디서 거는건지 알 수가 없네요.
2. Job 은 어떻게 실행되는가 - 트랜잭션에 관한 이야기
Transaction Manager 를 stepBuilder 에서 주입했는데, chunk 단위 처리를 하는 Tasklet 상 코드를 볼 수 없었습니다.
제가 처음 예상했던 그림은 아래와 같았는데 여지없이 빗나가서 Job 실행단계를 파봐야함을 느꼈습니다.
//ChunkOrientedTasklet.java 내에 이게 있을것이다!!! (X)
read()
트랜잭션매니저.doTransaction start
process()
writer()
트랜잭션매니져.commit()Job을 실행할 때
JobLauncher.launch()를 통해 실행을 합니다.계속 메서드를 따라 들어가다보면, TaskletStep -> doExecute 에서 아래를 볼 수 있습니다.
new TransactionTemplate(transactionManager, transactionAttribute)
.execute(new ChunkTransactionCallback(chunkContext, semaphore));Step 의 ChunkContext를 spring-tx 의 TransactionTemplate 위에서 Callback 형태로 동작 시키는 겁니다.
요걸 Callback 형태로 실행을 해서 따라가기가 좀 힘들긴한데, 결국 step 의 transactionManager 지배하에 Chunk Tasklet 자체가 실행됨을 볼 수 있습니다.
결론은 나왔습니다. read 하기 전부터 transaction 이 걸리는거고, 중간 processor에서 insert 를 하건 뭘하건, 어디든 한 곳에서 뻑나면 모두 rollback이 될 수 있다는걸 소스를 통해 확인했습니다.
//TaskletStep.doInTransaction : 트랜잭션 안에서 이 일을해라...
@Override
public RepeatStatus doInTransaction(TransactionStatus status) {
TransactionSynchronizationManager.registerSynchronization(this);
...
result = tasklet.execute(contribution, chunkContext); //저 tasklet 은 ChunkOrientedTasklet.java 임!
...
}후기
왜 제 예상대로 만들지 않고, 실행단계에서 Chunk 단위로 처리할때 복잡한 콜백 패턴으로 만든건지 생각을 해봣는데, 개발단계에서 이게 더 분업에 효율적이고 모듈화를 잘 된 케이스가 아닐까 싶습니다.
왜냐면
1번 Job 을 어떻게 구성하는가영역은 말그대로 Job 을 "구성" 의 개발만하면 되는 것이고,2번 Job은 어떻게 실행되는가를 구현하는 이들은 "실행"단 개발에 집중할 수 있기 떄문이 아니었을가 싶네요.이건 시간날때 한번 spring batch repo 에 질문해볼 수 있을 것 같습니다.
긴 글 읽어주셔서 감사합니다.
이 글은 개인 블로그에도 업로드 되어있습니다.
오소리 님의 최신 블로그
더보기-
- Spring batch PagingReader 로 Update ＆ Delete 시 데이터 처리 누락되는 경우
- Spring batch 의 ChunkOrientTasklet 에서 Reader 를 PagingItemReader, Writer 에선 삭제 Operation 을 할 때, 누락이 발생할 수 있음을 알리는 글 입니다. Chunk 단위 Transaction 처리 spring batch 는 chunk 단위로 처리를 합니다. 처리할 전체를 읽어서 한번에 넘기는 방식이 아닌! 일부를 읽고 그 일부를 처리한 뒤 commit 하는 사이클을 반복하는데요. 그 chunk 의 사이즈는 step 을 구현할 때 .chunk(int size) 를 통해서 설정합니다. @Bean public Step step1() { return this.stepBuilderFactory.get(＂step1＂...
-
- Spring Batch 는 어떻게 Chunk 지향처리를 하고, Transaction을 언제 걸까
- Spring Batch Chunk Oriented ＆ Transaction 맨날 StepBuilder.java 나 JobBuilder.java 로 기계적으로 job 을 만들다 보면, 너무나 많이 추상화 된 프레임워크에 의해 바보가 되는 느낌을 자주 받습니다. 작업을 하다 이 프레임워크는 어느 타이밍에 Transaction 을 걸고, 어떻게 Chunk 지향처리를 하는지 궁금해졌습니다. Spring Batch 는 chunk 단위로 데이터 처리를 합니다. Job - Step - Tasklet - Reader + Process + Writer 계층을 이루고 있는 점. 그리고 Tasklet 내 chunk 단위 처리 중 에러가 발생하면 롤백 한다는 내용은 많은이가 알고 있습...
DEVOTEE를 활성화 시키면
지금 작성한 댓글에 AI가 댓글을 달아줍니다.