Spring Batch ItemReader

ItemReader는 배치에서 처리할 아이템을 만들거나 읽는 역할을 합니다.

 

- 배치 대상 데이터를 읽기 위한 설정

  : 파일, DB , 네트워크 등에서 읽기 위함이다.

 

- Step에 ItemReader 기본 구현체

  : file, jdbc, jpa, hibernate, kafka, etc..

 

ItemReader 인터페이스를 구현한 JpaCursorItemReader 와 JdbcCursorItemReader가 존재합니다.

이 구현체들은 ItemStream또한 구현하고 있습니다.

ItemReader의 인터페이스입니다.

ItemStream 인터페이스입니다. 아이템을 스트림 처리할 수 있게 도와줍니다.

매개변수는 Job과 Step의 상태를 저장하는 ExecutionContext인걸 확인할 수 있습니다.

이건 ItemStream구현체들이 데이터를 읽고 쓰면서 상태를 관리할 수 있다는 것입니다.

 

 

- ItemReader 구현체가 없으면 직접 개발

- ItemStream은 ExecutionContext로 read 와 write 정보를 저장합니다.

 

 

우선 예제를 진행할 때 마다 동일한 Configuration파일을 만들어왔기에 인텔리제이의 템플릿 기능을 활용하겠습니다.

가장 비슷한 템플릿 코드를 작성합니다.

@Configuration
@Slf4j
public class TemplateConfiguration {

    private final JobBuilderFactory jobBuilderFactory;
    private final StepBuilderFactory stepBuilderFactory;


    public TemplateConfiguration(JobBuilderFactory jobBuilderFactory, StepBuilderFactory stepBuilderFactory) {
        this.jobBuilderFactory = jobBuilderFactory;
        this.stepBuilderFactory = stepBuilderFactory;
    }

    @Bean
    public Job job(){

        return this.jobBuilderFactory.get("job")
                .incrementer(new RunIdIncrementer())
                .start(this.step())
                .build();

    }

    @Bean
    public Step step() {

        return  this.stepBuilderFactory.get("")
                .chunk()
                .reader()
                .processor()
                .writer()
                .build();
    }
}

 

이제 인텔리제이 Tools의 save File as Template이 존재합니다.

그럼 ${}로 변수를 설정할 수가 있습니다.

 

package ${PACKAGE_NAME};


import lombok.extern.slf4j.Slf4j;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.core.launch.support.RunIdIncrementer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
@Slf4j
public class ${NAME} {

    private final JobBuilderFactory jobBuilderFactory;
    private final StepBuilderFactory stepBuilderFactory;


    public ${NAME}(JobBuilderFactory jobBuilderFactory, StepBuilderFactory stepBuilderFactory) {
        this.jobBuilderFactory = jobBuilderFactory;
        this.stepBuilderFactory = stepBuilderFactory;
    }

    @Bean
    public Job ${job_name}(){

        return this.jobBuilderFactory.get("${job_name}")
                .incrementer(new RunIdIncrementer())
                .start(this.${step_name}())
                .build();

    }

    @Bean
    public Step ${step_name}() {

        return  this.stepBuilderFactory.get("${step_name}")
                .chunk()
                .reader()
                .processor()
                .writer()
                .build();
    }
}

 

이제 파일을 생성할 때 적어준 이름으로 파일을 생성할 수 있습니다. 매개변수를 넣어줄 수도 있고요

 

반복해서 작성을 쉽게 만들 수 있습니다.

 

 

Custom 한 ItemReader를 만들어 보겠습니다.

 

public class CustomItemReader<T> implements ItemReader<T> {

    private final List<T> items;

    public CustomItemReader(List<T> items) {
        this.items = new ArrayList<>(items);
    }



    @Override
    public T read() throws Exception, UnexpectedInputException, ParseException, NonTransientResourceException {

        if (!items.isEmpty()){
            return items.remove(0);
        }

        return null;
    }
}

 

 

이 클래스는 List를 생성자로 주입을 받고 element를 하나 씩 꺼내서 제거하는 동시에 리턴하며,  리스트에 요소가 없는 경우는 null을 리턴합니다. null을 리턴하게 되면 chunk반복의 끝을 의미하기 때문입니다.

@Configuration
@Slf4j
public class ItemReaderConfiguration {

    private final JobBuilderFactory jobBuilderFactory;
    private final StepBuilderFactory stepBuilderFactory;


    public ItemReaderConfiguration(JobBuilderFactory jobBuilderFactory, StepBuilderFactory stepBuilderFactory) {
        this.jobBuilderFactory = jobBuilderFactory;
        this.stepBuilderFactory = stepBuilderFactory;
    }

    @Bean
    public Job itemReaderJob() {

        return this.jobBuilderFactory.get("itemReaderJob")
                .incrementer(new RunIdIncrementer())
                .start(this.customItemReaderStep())
                .build();

    }

    @Bean
    public Step customItemReaderStep() {

        return this.stepBuilderFactory.get("customItemReaderStep")
                .<Person,Person>chunk(10)//input과 output 타입을 Person으로 정의했다.
                .reader(new CustomItemReader<>(getItems()))  //getItems로 CustomItemReader로 처리할 데이터를 만들어 줄 것이다.
                .writer(itemWriter())
                .build();
    }

    private ItemWriter<Person> itemWriter() {
        return items -> log.info(items.stream()
                .map(Person::getName)
                .collect(Collectors.joining(","))); //reader에서 읽은 person item List를 ,기준으로 이름만 나열하겠습니다.
    }

    private List<Person> getItems() {
        List<Person> items = new ArrayList<>();

        for (int n = 0; n < 10; n++) {
            items.add(new Person(n+1,"test_name" + n,"test age", "test address"));
        }
        return items;
    }
}

 

Spring에서 제공하는 ListItemReader와 동일한 클래스를 만든 것입니다.