[SPRING] 주문금액 집계 프로젝트 실습 part2

part1에서 진행했던 내용을 기반으로 이어서 진행한다.

 

● User의 totalAmount를 Orders Entity로 변경

• 하나의 User는 N개의 Orders를 포함

● 주문 총 금액은 Orders Entity를 기준으로 합산

● `-date=2021-08` JobParameters 사용

• 주문 금액 집계는 orderStatisticsStep으로 생성

■ `2021년_08월_주문_금액.csv` 파일은 2021년 8월1일~8월31일 주문 통계 내역

     `date` 파라미터가 없는 경우, orderStatisticsStep은 실행하지 않는다.

 

user 한명이 orders의 여러개를 만들 수 있도록 일대다 구조로 만들 수 있게 엔티티를 수정한다.

 

1. Orders 추가

@Entity
@Getter
@NoArgsConstructor
public class Orders {
    @Id
    @GeneratedValue(strategy = GenerationType.IDENTITY)
    private Long id;

    private String itemName;

    private int amount;

    private LocalDate createdDate;

    @Builder
    public Orders(String itemName, int amount, LocalDate createdDate) {
        this.itemName = itemName;
        this.amount = amount;
        this.createdDate = createdDate;
    }
}

2. User 일대다 구조로 수정

@Getter
@Entity
@NoArgsConstructor
public class User {
    @Id
    @GeneratedValue(strategy = GenerationType.IDENTITY)
    private Long id;

    private String username;

    @Enumerated(EnumType.STRING)
    private Level level = Level.NORMAL;

//    private int totalAmount;
    /**
     * 1:N 관계
     * user는 n개의 Orders를 가질 수 있다.
     * user가 저장되면서 Orders를 같이 저장할 수 있도록 영속성 전이(PERSIST)를 적용한다.
    **/
    @OneToMany(cascade = CascadeType.PERSIST)
    @JoinColumn(name = "user_id")
    private List<Orders> orders;

    private LocalDate updatedDate;

    @Builder
    private User(String username, List<Orders> orders) {
        this.username = username;
        this.orders = orders;
    }

    public boolean availableLeveUp() {
        return Level.availableLevelUp(this.getLevel(), this.getTotalAmount());
    }

    private int getTotalAmount() {
        return this.orders.stream()
                .mapToInt(Orders::getAmount)
                .sum();
    }

    public Level levelUp() {
        Level nextLevel = Level.getNextLevel(this.getTotalAmount());
        this.level = nextLevel;
        this.updatedDate = LocalDate.now();

        return nextLevel;
    }
}

3. OrderStatistics (주문금액) 객체를 따로 분리한다.

@Getter
public class OrderStatistics {

    private String amount;

    private LocalDate date;

    @Builder
    public OrderStatistics(String amount, LocalDate date) {
        this.amount = amount;
        this.date = date;
    }
}

 

4. UserConfiguration에 orderStatisticsStep(주문금액집계)를 추가하여 월 단위 주문 통계 내역을 조회하고 csv파일을 생성한다.

@Configuration
@Slf4j
public class UserConfiguration {

    public static final int CHUNK_SIZE = 100;
    private final JobBuilderFactory jobBuilderFactory;
    private final StepBuilderFactory stepBuilderFactory;
    private final UserRepository userRepository;
    private final EntityManagerFactory entityManagerFactory;
    private final DataSource dataSource;

    public UserConfiguration(JobBuilderFactory jobBuilderFactory, StepBuilderFactory stepBuilderFactory, UserRepository userRepository, EntityManagerFactory entityManagerFactory, DataSource dataSource) {
        this.jobBuilderFactory = jobBuilderFactory;
        this.stepBuilderFactory = stepBuilderFactory;
        this.userRepository = userRepository;
        this.entityManagerFactory = entityManagerFactory;
        this.dataSource = dataSource;
    }

    @Bean
    public Job userJob() throws Exception {
        return this.jobBuilderFactory.get("userJob")
                .incrementer(new RunIdIncrementer())
                .start(this.saveUserStep())
                .next(this.userLevelUpStep())
                .next(this.orderStatisticsStep(null))
                .listener(new LevelUpJobExecutionListener(userRepository))
                .build();
    }

    @Bean
    @JobScope
    public Step orderStatisticsStep(@Value("#{jobParameters[date]}") String date) throws Exception {
        return this.stepBuilderFactory.get("orderStatisticsStep")
                .<OrderStatistics, OrderStatistics>chunk(CHUNK_SIZE)
                .reader(orderStatisticsItemReader(date))
                .writer(orderStatisticsItemWriter(date))
                .build();
    }

    private ItemReader<? extends OrderStatistics> orderStatisticsItemReader(String date) throws Exception {
        YearMonth yearMonth = YearMonth.parse(date);

        Map<String, Object> parameters = new HashMap<>();
        parameters.put("startDate", yearMonth.atDay(1)); // 2021년 8월 1일
        parameters.put("endDate", yearMonth.atEndOfMonth()); // 8월의 마지막 날

        Map<String, Order> sortKey = new HashMap<>();
        sortKey.put("created_date", Order.ASCENDING);

        JdbcPagingItemReader<OrderStatistics> itemReader = new JdbcPagingItemReaderBuilder<OrderStatistics>()
                .dataSource(this.dataSource)
                .rowMapper((resultSet, i) -> OrderStatistics.builder()
                        .amount(resultSet.getString(1))
                        .date(LocalDate.parse(resultSet.getString(2), DateTimeFormatter.ISO_DATE))
                        .build())
                .pageSize(CHUNK_SIZE) // 페이징 설정
                .name("orderStatisticsItemReader")
                .selectClause("sum(amount), created_date")
                .fromClause("orders")
                .whereClause("created_date >= :startDate and created_date <= :endDate")
                .groupClause("created_date")
                .parameterValues(parameters)
                .sortKeys(sortKey)
                .build();

        itemReader.afterPropertiesSet();

        return itemReader;
    }

    private ItemWriter<? super OrderStatistics> orderStatisticsItemWriter(String date) throws Exception {
        YearMonth yearMonth = YearMonth.parse(date);

        String fileName = yearMonth.getYear() + "년_" + yearMonth.getMonthValue() + "월_일별_주문_금액.csv";

        BeanWrapperFieldExtractor<OrderStatistics> fieldExtractor = new BeanWrapperFieldExtractor<>();
        fieldExtractor.setNames(new String[]{"amount", "date"});

        DelimitedLineAggregator<OrderStatistics> lineAggregator = new DelimitedLineAggregator<>();
        lineAggregator.setDelimiter(",");
        lineAggregator.setFieldExtractor(fieldExtractor);

        FlatFileItemWriter<OrderStatistics> itemWriter = new FlatFileItemWriterBuilder<OrderStatistics>()
                .resource(new FileSystemResource("output/" + fileName))
                .lineAggregator(lineAggregator)
                .name("orderStatisticsItemWriter")
                .encoding("UTF-8")
                .headerCallback(writer -> writer.write("total_amount,date"))
                .build();

        itemWriter.afterPropertiesSet();

        return itemWriter;
    }

    @Bean
    public Step saveUserStep() {
        return this.stepBuilderFactory.get("saveUserStep")
                .tasklet(new SaveUserTasklet(userRepository))
                .build();
    }

    @Bean
    public Step userLevelUpStep() throws Exception {
        return this.stepBuilderFactory.get("userLevelUpStep")
                .<User, User>chunk(CHUNK_SIZE)
                .reader(itemReader())
                .processor(itemProcessor())
                .writer(itemWriter())
                .build();
    }

    private ItemWriter<? super User> itemWriter() {
        return users -> users.forEach(x -> {
                x.levelUp();
                userRepository.save(x);
        });
    }

    private ItemProcessor<? super User, ? extends User> itemProcessor() {
        return user -> {
            if (user.availableLeveUp()) {
                return user;
            }

            return null;
        };
    }

    private ItemReader<? extends User> itemReader() throws Exception {
        JpaPagingItemReader<User> itemReader = new JpaPagingItemReaderBuilder<User>()
                .queryString("select u from User u")
                .entityManagerFactory(entityManagerFactory)
                .pageSize(CHUNK_SIZE)
                .name("userItemReader")
                .build();

        itemReader.afterPropertiesSet();

        return itemReader;
    }
}

 

5. JobExecutionDecider에서 제공하는 주문 금액 집계 step 실행 여부 벨리데이션을 구현한다.

public class JobParameterDecide implements JobExecutionDecider {

    public static final FlowExecutionStatus CONTINUE = new FlowExecutionStatus("CONTINUE");

    private final String key;

    public JobParameterDecide(String key) {
        this.key = key;
    }

    @Override
    public FlowExecutionStatus decide(JobExecution jobExecution, StepExecution stepExecution) {
        String value = jobExecution.getJobParameters().getString(key);

        // key가 없으면
        if(StringUtils.isEmpty(value)) {
            return FlowExecutionStatus.COMPLETED;
        }

        return CONTINUE;
    }
}

* JobExecutionDecider의 인터페이스는 flow결정과_배치실행여부를 결정한다.

6. 5번에서 만든 JobExecutionDecide를 기반으로 step 기능을 추가한다.

 

* UserConfiguration.java 의 userJob() 영역 step 추가

@Bean
public Job userJob() throws Exception {
	return this.jobBuilderFactory.get("userJob")
		.incrementer(new RunIdIncrementer())
        .start(this.saveUserStep())
        .next(this.userLevelUpStep())
        .listener(new LevelUpJobExecutionListener(userRepository))
        .next(new JobParameterDecide("date")) // 가져온값이 아래의 CONTINUE인지 체크
        .on(JobParameterDecide.CONTINUE.getName()) // CONTINUE이면 아래의 to메서드 실행
        .to(this.orderStatisticsStep(null))
        .build()
        .build();
}

 

* 결과 값

 

느낀점 :

1.  user가 저장되면서 Orders를 같이 저장할 수 있도록 영속성 전이(PERSIST)를 적용 함으로 써 JPA의 이해를 한번 더 상기 시켰다.

(잊을만 하면 기존에 기록했던 영속성 전이를 다시 보면서 리마인딩 하자)

 

2. JobExecutionDecider를 통해 배치실행여부를 좀 더 쉽게 확인이 가능한점을 배움.

 

3. FlatFileItemWriter로 데이터 CSV 파일 쓰기를 통해 보다 편하게 생성할 수 있었다.

 

학습사이트 : 패스트 캠퍼스 (스프링 배치편)