SpringBatch : CompositeItemProcessor / ClassifierCompositeItemProcessor

이 게시글은 인프런 정수원님의 강의를 복습하며 작성한 글입니다.

CompositeItemProcessor

CompositeItemProcessor는 여러 ItemProcessor를 체이닝해서 사용할 수 있도록 지원해주는 클래스다. ItemProcessor 하나가 모두 처리하기에는 무거운 것이 있다고 하면, CompositeItemProcessor를 이용해서 ItemProcessor를 좀 더 분절화해서 코드 복잡도를 낮출 수 있다.

밑에서 간단히 코드를 보면서 확인할 예정이지만, CompositeItemProcessor는 입력받은 Input Chunk의 Item을 하나씩 가지고 와서, 모든 ItemProcessor에 순서대로 처리를 해준다. 그리고 Output Chunk에 1개씩 저장해주는 형식으로 동작한다.

 

CompositeItemProcessor 구조 및 API

CompositeItemProcessor는 클래스 내부에 ItemPrcoessor를 보관하고, Item을 1개 꺼내서 Processor에 전달한다. 그리고 Processor는 처리가 끝난 Item을 다음 Processor의 Input으로 전달을 해주는 식으로 한다. 즉, ItemProcessor가 한 번에 한 개의 Item을 처리하는 동작방식은 그대로 유지가 된다. 

@Bean
public ItemProcessor<Customer, Customer> customItemProcessor() {
    CompositeItemProcessorBuilder<Customer, Customer> builder = new CompositeItemProcessorBuilder<>();
    return builder.
            delegates(subItemProcessor2(), subItemProcessor1())
            .build();
}

만드는 방법은 Builder를 이용하는 방법과 직접 구현체를 생성하는 방법이 있다. 구현체를 생성해서 사용하면, ItemProcessor를 리스트 형태로 만들어서 전달해야한다. 반대로 Builder를 만들 경우에는 ItemProcessor를 각각 생성해서 전달하는 방법도 가능하다.

 

CompositeItemProcessor 내부 동작방식 확인

CompositeItemProcessor

• CompositeItemProcessor는 Process 메서드를 통해 Processor를 동작시킨다. 매개변수로 Item이 주어진다. 즉, Item이 1개씩만 전달된다는 의미다.
• 내부적으로 ItemProcessor를 가지고 있는 delegates에서 for문을 돌리면서 processItem을 통해 각 itemProcessor를 통한 처리를 하고, 그 결과값을 다음 ItemProcessor에 전달하는 역할을 한다.
• For문을 다 돌고 result가 만들어지면, 이 데이터는 SimpleChunkPrcessor로 돌아가게 될 것이고, OutPutChunk에 전달되게 된다. 

 

CompositeItemProcessor ItemProcessor 순서

CompositeItemProcessor는 ItemProcessor 클래스를 내부에 리스트로 저장한다. 리스트는 순서를 가지고 있으므로, ItemProcessor는 해당 리스트에 추가한 순서대로 돌아간다. 즉, 다음과 같다.

Item1 → ItemProcessor1 → ItemProcessor2 → Output1
Item2 → ItemProcessor1 → ItemProcessor2 → Output2
Item3 → ItemProcessor1 → ItemProcessor2 → Output3
Item4 → ItemProcessor1 → ItemProcessor2 → Output4

 

 

ClassifierCompositeItemProcessor

ClassifierCompositeItemProcessor는 특정 Item마다 다른 ItemProcessor로 처리를 할 수 있도록 지원하는 클래스다. ClassifierCompositeItemPrcoessor는 내부적으로 Classifier를 Map 형태로 가지고 있다. 동작방식은 Classifier에 Input을 넣고 해당되는 ItemProcessor를 전달받는 형식이다. 

따라서 ClassifierCompositeItemPrcoessor를 이용하기 위해서는 크게 두 가지를 더 구성해야한다.

1. Classifier 객체 만들고, Classifier 객체에 분류 조건 + ItemProcessor 넣기.
2. ClassifierCompositeItemProcessor에 Classifier 객체 설정하기 

 

ClassifierCompositeItemProcessor의 API

ClassifierCompositeItemProcessor는 Builder를 사용해서 만든다. 이 때, Classifier에만 값을 설정해주면 되고, 그것은 아래 API를 이용해서 할 수 있다. 

• classifier : 분류에 사용될 Classifier 맵을 설정한다.
• build : ClassifierCompositeItemProcessor를 만든다. 

 

ClassifierCompositeItemProcessor의 구조 및 내부 코드 확인

ItemProcessor답게 Process 메서드를 통해서 실제 처리가 진행된다. 그렇지만 다른 ItemProcessor와 조금 다른 부분은 process를 하는 시점에 어떤 ItemProcessor가 사용될지 결정되지 않았기 때문에 사용될 ItemProcessor를 찾아온 다음에 그 Processor에서 process가 진행된다는 점이다. 

ClassifierCompositeItemProcessor

ClassifierCompositeItemProcessor로 들어와보면, Process 메서드를 먼저 볼 수 있다. Process 메서드에서는 아직 어떤 ItemProcessor인지 모르기 때문에 ItemProcessor를 찾아와야한다. 그래서 먼저 classifier.classify(item) 메서드를 통해 ItemProcessor를 찾아온다.

Classifier

Classifier.classify로 가보면 Interface라는 것을 알 수 있다. 따라서 개발자가 해야하는 것은 Classifier를 만들고, classify 메서드를 구현하는 것이다. 그리고 key는 input 조건, value는 ItemProcessor를 이용해서 적절하게 전달하면 된다. 

ClassifierCompositeItemProcessor

classifier.classify() 메서드의 결과로 받은 itemProcessor와 item을 processItem 메서드로 넘기고, processItem은 전달받은 Processor를 통해 실제 Process를 진행한다. 여기서 Process의 실행 결과를 받아서 Return 해주게 된다. 

 

ClassifierCompositeItemProcessor 생성 위한 코드 작성

ClassifierCompositeItemProcessor 생성

public ItemProcessor<? super Customer, ? extends Customer> custom1ItemProcessor() {

    // ClassifierCompositeItemProcessor Builder 클래스 생성
    ClassifierCompositeItemProcessorBuilder<Customer, Customer> builder = new ClassifierCompositeItemProcessorBuilder<>();

    // Builder 클래스에 전달한 Classifier 클래스 생성
    ProcessorClassifier<? super Customer, ItemProcessor<?, ? extends Customer>> classifier2 = new ProcessorClassifier<>();

    //Classifier 내부에 가지고 있는 ParamMap(구분자 생성 및 초기화)
    HashMap<Integer, ItemProcessor<Customer, Customer>> paramMap = new HashMap<>();
    paramMap.put(0, new CustomItemProcessor1());
    paramMap.put(1, new CustomItemProcessor2());
    paramMap.put(2, new CustomItemProcessor3());

    //Classifier 셋팅
    classifier2.setParamMap(paramMap);

    // Builder를 통해서 Processor 생성
    return builder.classifier(classifier2).build();
}

• ClassifierCompositeItemProcessor 객체를 생성한다.
• ClassifierCompositeItemProcessor는 내부에 구분자인 Classifier를 가져야하므로, 구분자 역할을 할 Classifier 생성 및 셋팅
• 이 때, Classifier는 Map의 형태를 가지고 Key에 대응되는 ItemProcessor를 각각 넣어줌. 이 Classfier를 통해서 구분자를 구현했음.

 

Classifier 클래스 구현 

@Setter
public class ProcessorClassifier<C,T> implements Classifier<C,T> {

    // 구분자 객체
    private Map<Integer, ItemProcessor<Customer, Customer>> paramMap;

    // Processor Return 로직. 
    // ClassifierItemProcessor는 process 메서드를 할 때, classify 메서드에서 ItemProcessor를 전달받아야함.
    @Override
    public T classify(C classifiable) {
        Customer key = (Customer) classifiable;
        int i = key.getId().intValue() % 3;
        
        // ItemProcessor로 타입 캐스팅 후, 반환.
        return (T)paramMap.get(i);
    }
}

ClassifierCompositeItemProcessor는 Process 메서드를 실행할 때, Classifier.classify 메서드를 통해 ItemProcessor를 얻어서 동작한다. 따라서 classify에 구분자에 ItemProcessor를 잘 꺼내서 전달하는 것을 구현하면 된다. 

 

 

테스트 코드 

 

GitHub - chickenchickenlove/springbatchstudy

 

 

테스트 코드 실행

전달되는 Item 객체의 Id값을 3으로 나눠서 그 나머지에 대해서 서로 다른 ItemProcessor가 출력되도록 ClassifierCompositeItemProcessor 로직을 구현했다. 실행 결과는 아래와 같다. 

1 → 2 → 3 → 1 ... 의 순서대로 각각의 ItemProcessor가 호출되는 것이 확인된다. 즉, Id값에 따라 동적으로 itemProcessor가 배정되어서 Processing을 하는 것으로 이해할 수 있다.