Spring Batch 구현 - 5

 
♧ 전체 코드 : https://github.com/woodisco/pass-batch

GitHub - woodisco/pass-batch

 

수업 종료후, 이용권 차감 배치

① UsePassesJobConfig 작성

@Configuration
public class UsePassesJobConfig {

    private final int CHUNK_SIZE = 5;

    private final EntityManagerFactory entityManagerFactory;
    private final PassRepository passRepository;
    private final BookingRepository bookingRepository;

    public UsePassesJobConfig(EntityManagerFactory entityManagerFactory, PassRepository passRepository, BookingRepository bookingRepository) {
        this.entityManagerFactory = entityManagerFactory;
        this.passRepository = passRepository;
        this.bookingRepository = bookingRepository;
    }

    @Bean
    public Job usePassesJob(JobRepository jobRepository, Step usePassesStep) {
        return new JobBuilder("usePassesJob", jobRepository)
                .start(usePassesStep)
                .build();
    }

    @Bean
    public Step usePassesStep (JobRepository jobRepository, PlatformTransactionManager transactionManager) {
        return new StepBuilder("usePassesStep", jobRepository)
                .<BookingEntity, Future<BookingEntity>>chunk(CHUNK_SIZE, transactionManager)
                .reader(usePassesItemReader())
                .processor(usePassesAsyncItemProcessor())
                .writer(usePassesAsyncItemWriter())
                .build();
    }

    @Bean
    public JpaCursorItemReader usePassesItemReader() {
        return new JpaCursorItemReaderBuilder<BookingEntity>()
                .name("usePassesItemReader")
                .entityManagerFactory(entityManagerFactory)
                // 상태(status)가 완료이며, 종료 일시(endedAt)이 과거인 예약이 이용권 차감 대상이 됩니다.
                .queryString("select b from BookingEntity b join fetch b.passEntity where b.status = :status and b.usedPass = false and b.endedAt < :endedAt")
                .parameterValues(Map.of("status", BookingStatus.COMPLETED, "endedAt", LocalDateTime.now()))
                .build();
    }

    @Bean
    public AsyncItemProcessor<BookingEntity, BookingEntity> usePassesAsyncItemProcessor() {
        AsyncItemProcessor<BookingEntity, BookingEntity> asyncItemProcessor = new AsyncItemProcessor<>();
        asyncItemProcessor.setDelegate(usePassesItemProcessor()); // usePassesItemProcessor로 위임하고 결과를 Future에 저장합니다.
        asyncItemProcessor.setTaskExecutor(new SimpleAsyncTaskExecutor());
        return asyncItemProcessor;

    }

    @Bean
    public ItemProcessor<BookingEntity, BookingEntity> usePassesItemProcessor() {
        return bookingEntity -> {
            // 이용권 잔여 횟수는 차감합니다.
            PassEntity passEntity = bookingEntity.getPassEntity();
            passEntity.setRemainingCount(passEntity.getRemainingCount() - 1);
            bookingEntity.setPassEntity(passEntity);

            // 이용권 사용 여부를 업데이트합니다.
            bookingEntity.setUsedPass(true);
            return bookingEntity;

        };
    }

    @Bean
    public AsyncItemWriter<BookingEntity> usePassesAsyncItemWriter() {
        AsyncItemWriter<BookingEntity> asyncItemWriter = new AsyncItemWriter<>();
        asyncItemWriter.setDelegate(usePassesItemWriter()); // usePassesItemWriter 최종 결과값을 넘겨주고 작업을 위임합니다.
        return asyncItemWriter;
    }

    @Bean
    public ItemWriter<BookingEntity> usePassesItemWriter() {
        return bookingEntities -> {
            for (BookingEntity bookingEntity : bookingEntities) {
                // 잔여 횟수를 업데이트 합니다.
                int updatedCount = passRepository.updateRemainingCount(bookingEntity.getPassSeq(), bookingEntity.getPassEntity().getRemainingCount());
                // 잔여 횟수가 업데이트 완료되면, 이용권 사용 여부를 업데이트합니다.
                if (updatedCount > 0) {
                    bookingRepository.updateUsedPass(bookingEntity.getPassSeq(), bookingEntity.isUsedPass());
                }
            }
        };
    }
}

▷ JpaCursorItemReader :
이 리더는 Java Persistence API (JPA)를 사용하여 데이터베이스에서 데이터를 읽어오는 데 특화되어 있습니다.
이 리더는 커서(cursor) 기반의 방식으로 데이터를 읽어오는데, 일정한 크기의 데이터 덩어리를 메모리에 한꺼번에 로드하는 것이 아니라, 데이터베이스의 커서(cursor)를 사용하여 한 번에 한 행씩 데이터를 가져옵니다. 이는 메모리 사용량을 최소화하면서 대규모의 데이터를 처리할 때 유용합니다.

▷ AsyncItemProcessor :
ItemProcessor에게 별도의 스레드가 할당되어서 처리되는 방식이다. ItemProcessor의 처리가 오래 걸리는 경우에 AsyncItemProcessor 혹은 AsyncItemWriter를 사용하여 성능을 향상할 수 있다. AsyncItemProcessor에서 List<Future>가 AsyncItemWriter로 전달되어 ItemWriter에서 처리가 되는데 이때 AsyncItemProcessor와 AsyncItemWriter를 같이 사용해야만 Future를 사용할 수 있다.

※ Future :
비동기 작업의 결과를 나타내는 인터페이스입니다. 비동기 작업은 작업이 완료되기를 기다리지 않고 다른 작업을 계속할 수 있게 해 줍니다. Future를 사용하면 비동기 작업이 완료될 때까지 기다릴 필요 없이 결과를 가져올 수 있습니다. 기본적으로 Future는 작업의 현재 상태를 쿼리하고 작업이 완료되기를 기다리는 메커니즘을 제공합니다. 나중에 Future를 사용하여 작업이 완료됐을 때 결과를 가져오거나 작업이 완료되기를 기다리는 등의 작업을 할 수 있습니다.
※ Delegate :
다른 객체의 동작을 대신해서 호출하는 객체를 말합니다. 이는 객체지향 프로그래밍에서 중요한 개념 중 하나입니다.

② PassRepository에 추가
③ BookingRepository에 추가
 

출처 : 패스트캠퍼스 10개 프로젝트로 완성하는 백엔드 웹개발(Java/Spring) 초격차 패키지 Online