Spring Batch commit-interval에 대한 정리

Spring Batch Job 설정 시 commit-interval 이라는 값을 지정할 수 있다. 이 값이 의미하는 바는 commit되는 단위 갯수인데 정확히 어떤 의미인지 정리해본다.

스프링 배치 문서에는 다음과 같이 되어있다.

 

The number of items that will be processed before commit is called for the transaction.  
Either set this or the chunk-completion-policy but not both. Can be specified as an expression
that will be evaluated in the scope of the step (e.g. "#{jobParameters['commit.interval']}").

트랜잭션 커밋이 호출되기 전 처리되어야 하는 아이템들의 갯수.
이것을 설정하거나 chunk-completion-policy를 설정할 수 있으나 둘 다 할 수는 없다.
"#{jobParameters['commit.interval']}" 와 같이 스텝 스코프의 표현식으로도 지정될 수 있다.

 

먼저 commit-interval은 다음과 같이 지정할 수 있다.

<job id="testJob">  
    <step id="testJob-step1">  
        <tasklet>  
            <chunk reader="testReader" processor="testProcessor" writer="testWriter" commit-interval="100"/>  
        </tasklet>  
    </step>  
</job>

 

Java Config 방법으로 설정할 때는 다음과 같이 설정한다.

@Bean  
public Job testJob() {  
    return jobBuilderFactory.get("testJob")  
        .start(testStep())  
        .build();  
}  
  
@Bean  
public Step testStep() {  
    return stepBuilderFactory.get("testStep")  
        .<String, String>chunk(100)  
        .reader(testReader)  
        .processor(testProcessor)  
        .writer(testWriter)  
        .build();  
}

chunk메소드 chunkSize 인자로 입력되는 값이 commit-interval 값이다.

 

이 갯수는 reader가 읽고 processor가 처리해서 writer에 넘겨지는 갯수를 의미한다. 그리고 트랜잭션 설정이 되어있다면 이 갯수 단위로 트랜잭션 커밋이 발생한다.

 

스프링 배치는 reader가 읽고나서 processor가 처리한 갯수가, commit-interval개 만큼 쌓이면 writer에 item들을 보내서 write하게 된다.

스프링 배치 문서에는 다음 코드와 같은 개념이라고 이야기하고 있다.
(예전에는 1건 씩 read, process 했던 것으로 기억하는데 최신 버전에서는 commit-interval 건 씩 하는 것으로 변경된 것 같다.)

List items = new Arraylist();
for(int i = 0; i < commitInterval; i++){
    Object item = itemReader.read();
    if (item != null) {
        items.add(item);
    }
}

List processedItems = new Arraylist();
for(Object item: items){
    Object processedItem = itemProcessor.process(item);
    if (processedItem != null) {
        processedItems.add(processedItem);
    }
}

itemWriter.write(processedItems);

출처 : https://docs.spring.io/spring-batch/reference/step/chunk-oriented-processing.html

 

그림으로는 다음과 같다.

https://docs.spring.io/spring-batch/reference/step/chunk-oriented-processing.html

 

이렇게 동작하기 위해서는 reader에서 읽은 item의 타입과 processor가 받는 input item의 타입이 같아야 하고

writer는 processor의 output 타입의 item을 List로 받아야 한다.

 

ItemReader, ItemProcessor 인터페이스의 메소드가 단 건을 처리하도록 되어있고 ItemWriter 인터페이스의 메소드가 List를 처리하도록 되어있는 이유가 여기에 있다.

 

그렇다면 IbatisPagingItemReader와 같은 여러 건을 한번에 읽는 reader는 읽은 갯수가 어떻게 처리되는 것일까?

예를 들어 commit-interval이 1000으로 설정되어 있고 IbatisPagingItemReader에서 100건씩 페이징 처리하여 읽고 있다면, 실제로는 100000건을 읽은 후에야 writer에 item들이 전달되는 것이 아닐까 생각할 수 있다.

 

하지만 그렇게 되어 있지는 않고 reader 내부에 데이터를 적재해두고 read가 호출될 때마다 1건씩 리턴해주는 방식으로 되어있다.

적재한 데이터가 다 없어지면 새로 쿼리를 실행하고 리턴해준다. 실제 구현 코드를 보면 다음과 같다.

 

IbatisPagingItemReader가 상속하고 있는 페이징 처리 reader 추상클래스인 AbstractPagingItemReader의 doRead 메소드이다.

(상위 클래스의 read메소드에서 호출되는데 read메소드에는 별다른 로직이 없다.)

protected T doRead() throws Exception {  
  
    synchronized (lock) {  
  
        if (results == null || current >= pageSize) { // #1
            if (logger.isDebugEnabled()) {  
                logger.debug("Reading page " + getPage());  
            }  

            doReadPage();  

            page++;  
            if (current >= pageSize) {  
                current = 0;  
            }  
        }  

        int next = current++;  // #2
        if (next < results.size()) {  
            return results.get(next);  
        }  
        else {  
            return null;  
        }  
  
    }  
}

#1 처음 호출되는 것이거나(null) 가지고 온 갯수를 넘어갔다면(current >= pageSize) doReadPage 메소드를 호출해서 구현 클래스에게 데이터를 가져오도록 한다.

#2 데이터가 있거나 새로 읽어왔다면 results에서 한개를 꺼내서 리턴한다.(results는 List<T> 타입의 멤버 변수)

 

구현체인 IbatisPagingItemReader에서는 다음과 같이 doReadPage 메소드를 구현하고 있다.

@Override  
@SuppressWarnings("unchecked")  
protected void doReadPage() {  
    Map<String, Object> parameters = new HashMap<String, Object>();  

    if (parameterValues != null) {  
        parameters.putAll(parameterValues);  
    }  

    parameters.put("_page", getPage());  
    parameters.put("_pagesize", getPageSize());  
    parameters.put("_skiprows", getPage() * getPageSize());  

    if (results == null) {  
        results = new CopyOnWriteArrayList<T>();  
    }  
    else {  
        results.clear();  
    }  

    results.addAll(sqlMapClientTemplate.queryForList(queryId, parameters));  // #1
}

 

_page, _pagesize 등의 파라미터는 우리가 쿼리에 직접 설정해줘야하는 값이다.

#1 : SqlMapClientTemplate을 이용해 쿼리를 실행해서 results에 추가한다.


정리하자면..

reader 내부에서 데이터를 어떻게 가져오는지는 reader에서 구현하기 나름이다. 매번 sql을 실행해서 1건씩 조회하던지 100개씩 묶음으로 가져다놓고 한 건씩 건네줄 것인지.

doRead 메소드에서 1건씩 리턴하기만 한다면 파일을 한줄씩 읽던 100줄씩 읽던 commit-interval의 의미는 같게 유지된다.



그리고 읽어오는 데이터 타입 자체를 List로 구현할 수도 있고 다른 형태로 얼마든지 지정할 수 있으므로 commit-interval 단위는 각자가 원하는대로 하면 된다.