스프링 batch + scheduler 정산 시스템 구현

Job은 여러가지 Step, Step은 여러가지 Tesklet 또는 Chunks(아이템처리자)로 나누어져서 실행된다.
아이템 처리자는 Reader(읽기), Processor(변환 작업), Writer(쓰기)를 구현하여 실행 할 수 있다. 아이템 처리자의 방법은 정산과 같은 여러 데이터를 받아오는 값을 나눠서 실행할 수 있다.

job

스프링 batch와 scheduler를 사용하여 기존 만들었던 쇼핑몰 시스템의 주문 목록을 정산하기 위해 활용해보려고 한다. 정산데이터는 매달 15일 새벽4시에 생성된다.

만약 정산하려는 품목이 수만개라면 DB에서 불러올 때 문제가 될 수 있지만 100개씩 받아오거나 할 수 있다. 따라서 필자는 아이템 처리자의 방법을 사용하였다. 

@Bean
public Job makeRebateOrderItemJob(){
    return jobBuilderFactory.get("makeRebateOrderItemJob")
            //.incrementer(new RunIdIncrementer()) //강제로 매번 다른 ID를 실행시에 파라미터로 부여
            .start(makeRebateOrderItemStep1())
            .build();
}

OrderItem 이라는 각 주문에 대한 아이템을 정산 데이터로 만들기 위한 작업이다. 

Job의 이름은 makeRebateOrderItemJob이며 makeRebateOrderItemStep1이라는 Step을 실행시킨다.

 

 

@Bean
@JobScope
public Step makeRebateOrderItemStep1(
) {
    return stepBuilderFactory.get("makeRebateOrderItemStep1")
            .<OrderItem, RebateOrderItem>chunk(100) //입력과 출력, 한번에 받아오는 값
            .reader(orderItemReader())
            .processor(orderItemToRebateOrderItemProcessor())
            .writer(rebateOrderItemWriter())
            .build();
}

Step에서는  설정한 reader, processor, writer를 넣어 하나의 Step을 만들었다.

 

@StepScope
    @Bean
    public RepositoryItemReader<OrderItem> orderItemReader() {
        LocalDateTime now = LocalDateTime.now();
        LocalDateTime toDate = now.plusHours(1);
        LocalDateTime fromDate = now.minusDays(31);
//        System.out.println(toDate);
//        System.out.println(fromDate);

        return new RepositoryItemReaderBuilder<OrderItem>()
                .name("orderItemReader")
                .repository(orderItemRepository)
                .methodName("findAllByPayDateBetween") //여기서 사용하는 메소드는 리스트가 아닌 페이지를 리턴해야함 
                .pageSize(100)
                .arguments(Arrays.asList(fromDate, toDate))
                .sorts(Collections.singletonMap("id", Sort.Direction.ASC))
                .build();
    }

ItemReader에서는 읽을 대상을 설정한다. 저번달부터 현재 시간(+1)까지 실행시키기 위해 toDate와 fromDate라는 변수를 만들었고 이를 findAllByPayDateBetween 함수를 사용하여 OrderItemRepository에서 받아왔다. 이때 반환은 페이지로 받아와야 한다. 이전에 설정한 chunk(100)과 동일하게 pagesize는 100으로 설정해주었다. 

public interface OrderItemRepository extends JpaRepository<OrderItem, Long> {
    Page<OrderItem> findAllByPayDateBetween(LocalDateTime fromDate, LocalDateTime toDate, Pageable pageable);
}

arguments 에서는 매개변수를 설정할 수 있으며 sorts에서 정렬해주었다.

 

@StepScope
    @Bean
    public ItemProcessor<OrderItem, RebateOrderItem> orderItemToRebateOrderItemProcessor() {
        return orderItem -> new RebateOrderItem(orderItem);
    }

ItemProcessor에서는 처리를 진행한다.

기존에 만든 orderItem으로 RebateOrderItem을 생성하는 작업을 했다.

 

@StepScope
    @Bean
    public ItemWriter<RebateOrderItem> rebateOrderItemWriter() {
        return items -> items.forEach(item -> {
            RebateOrderItem oldRebateOrderItem = rebateOrderItemRepository.findByOrderItemId(item.getOrderItem().getId()).orElse(null);

            if (oldRebateOrderItem != null) {
                rebateOrderItemRepository.delete(oldRebateOrderItem);
            }

            rebateOrderItemRepository.save(item);
        });
    }

ItemWriter에서는  쓸 대상에 대해 설정한다. 각 아이템을 돌면서 null 값인 경우를 제외하고 DB에 저장하였다.

 

이제 Scheduler에 관련해서 살펴보자.

@Component
@RequiredArgsConstructor
public class RebateScheduler {
    private final JobLauncher jobLauncher;
    private final RebateOrderItemJobConfig rebateOrderItemJobConfig;

    //매달 15일 새벽4시 정산
    @Scheduled(cron="0 0 4 15 * *")
    public void runJob() {
        // job parameter 설정
        Map<String, JobParameter> confMap = new HashMap<>();
        confMap.put("time", new JobParameter(System.currentTimeMillis()));
        JobParameters jobParameters = new JobParameters(confMap);
        //JobParamter의 역할은 반복해서 실행되는 Job의 유일한 ID이다.
        //Job Parameter에 동일한 값이 세팅되면 두번째부터 실행이 안되기 때문이다.

        try {
            jobLauncher.run(rebateOrderItemJobConfig.makeRebateOrderItemJob(), jobParameters);
        } catch (JobExecutionAlreadyRunningException | JobInstanceAlreadyCompleteException
                 | JobParametersInvalidException | org.springframework.batch.core.repository.JobRestartException e) {
            System.out.println(e.getMessage());
        }
    }
}

Scheduler은 구글링을 해보니 간단했다.

cron 표현식은 http://www.cronmaker.com/;jsessionid=node036kd6cwcu4cq1g27h7np87d95919501.node0?0 참고해서 만들어보았다.

JobParamter는 유일한 Job의 ID를 생성해주기 위해 설정하였다. 동일한 값이 셋팅되면 두번째부터 실행되는 Job의 스텝이 실행되지 않을 것 이다.  정산이 주된 목적이면 파라미터로 연도를 넘기는 방법이 더 좋을 수도 있다. 정산을 기간동안 Job은 동일한 결과를 가져야 하기 때문이다. 

 

 

Rebate에 관련한 JobConfig 전체 소스코드는 다음과 같다.

package com.example.ll.finalproject.job.rebateOrderItem;

import com.example.ll.finalproject.app.order.entity.OrderItem;
import com.example.ll.finalproject.app.order.repository.OrderItemRepository;
import com.example.ll.finalproject.app.rebate.entity.RebateOrderItem;
import com.example.ll.finalproject.app.rebate.repository.RebateOrderItemRepository;
import com.example.ll.finalproject.app.util.Ut;
import lombok.RequiredArgsConstructor;
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.JobScope;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepScope;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.data.RepositoryItemReader;
import org.springframework.batch.item.data.builder.RepositoryItemReaderBuilder;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.domain.Sort;

import java.time.LocalDateTime;
import java.util.Arrays;
import java.util.Collections;

//잡은 여러가지 스텝, 스텝은 여러가지 테스클릿이나 Chunks(아이템처리자)로 나누어져서 실행
//아이템 처리자는 Reader(읽기), Processor(변환 작업), Writer(쓰기) => 데이터를 나눠서 실행


//기존적으로 @Bean을 붙이면 싱글톤 = > 스프링부트앱이 꺼지기 전까지 살아 있음 , 빈들이 다 객체화 되어 저장됨, 공유 자원임
//더 수명이 짧은 건 세션 @SessionScope => 처음 접근 시 세션이 활성화 됨, 세션이 끝날 때 까지 살아있음, 브라우저당 1개
//@RequestScope => 요청 당 객체가 1개 요청이 끝날 때 까지 살아 있음,  수명이 더 짧음
//@PrototypeScope= > 그냥 매번 새로 만듬
@Configuration
@RequiredArgsConstructor
public class RebateOrderItemJobConfig {
    private final JobBuilderFactory jobBuilderFactory;
    private final StepBuilderFactory stepBuilderFactory;
    private final OrderItemRepository orderItemRepository; //읽을 대상
    private final RebateOrderItemRepository rebateOrderItemRepository; //쓸 대상

    @Bean
    public Job makeRebateOrderItemJob(){
        return jobBuilderFactory.get("makeRebateOrderItemJob")
                //.incrementer(new RunIdIncrementer()) //강제로 매번 다른 ID를 실행시에 파라미터로 부여
                .start(makeRebateOrderItemStep1())
                .build();
    }
    @Bean
    @JobScope
    public Step makeRebateOrderItemStep1(
    ) {
        return stepBuilderFactory.get("makeRebateOrderItemStep1")
                .<OrderItem, RebateOrderItem>chunk(100) //입력과 출력, 한번에 받아오는 값
                .reader(orderItemReader())
                .processor(orderItemToRebateOrderItemProcessor())
                .writer(rebateOrderItemWriter())
                .build();
    }


    @StepScope
    @Bean
    public RepositoryItemReader<OrderItem> orderItemReader() {
        LocalDateTime now = LocalDateTime.now();
        LocalDateTime toDate = now.plusHours(1);
        LocalDateTime fromDate = now.minusDays(31);
//        System.out.println(toDate);
//        System.out.println(fromDate);

        return new RepositoryItemReaderBuilder<OrderItem>()
                .name("orderItemReader")
                .repository(orderItemRepository)
                .methodName("findAllByPayDateBetween") //여기서 사용하는 메소드는 리스트가 아닌 페이지를 리턴해야함 
                .pageSize(100)
                .arguments(Arrays.asList(fromDate, toDate))
                .sorts(Collections.singletonMap("id", Sort.Direction.ASC))
                .build();
    }

    @StepScope
    @Bean
    public ItemProcessor<OrderItem, RebateOrderItem> orderItemToRebateOrderItemProcessor() {
        return orderItem -> new RebateOrderItem(orderItem);
    }

    @StepScope
    @Bean
    public ItemWriter<RebateOrderItem> rebateOrderItemWriter() {
        return items -> items.forEach(item -> {
            RebateOrderItem oldRebateOrderItem = rebateOrderItemRepository.findByOrderItemId(item.getOrderItem().getId()).orElse(null);

            if (oldRebateOrderItem != null) {
                rebateOrderItemRepository.delete(oldRebateOrderItem);
            }

            rebateOrderItemRepository.save(item);
        });
    }

}

 

 

**참고** 

JobBuilderFactory,  StepBuilderFactory

• JobBuilderFactory릴 Job을 만드는게 아니고 JobBuilder를 만드는 것이다.
• JobBuilderFactory 클래스의 get() 메서드로 JobBuilder를 생성할 수 있다.
  • JobBuilder get(String name);
• 이렇게 생성된 JobBuilder로 Job을 생성한다.

Scope라는 범위를 뜻한다.

흔히 볼 수 있는 @Bean을 붙이면 싱글톤을 뜻한다. 싱글톤은 스프링부트앱이 꺼지기 전까지 살아 있음 , 빈들이 다 객체화 되어 저장되며 공유 자원이다.
더 수명이 짧은 건 세션 @SessionScope이다. 처음 접근 시 세션이 활성화 되며 세션이 끝날 때 까지 살아있다. 즉 브라우저당 1개라고 생각하면 된다.
@RequestScope는 요청 당 객체가 1개 요청이 끝날 때 까지 살아 있다. 즉  수명이 더 짧다.
@PrototypeScope는 그냥 매번 새로 만든다.

baeldung에 더 자세한 정보가 나와있으니 참고하자. 스코프에 대한 개념은 더욱 공부를 해봐야 할 것 같다.!  

https://www.baeldung.com/?s=scope 

scope | Baeldung