[Java] 대용량 엑셀파일읽기 성능 개선 - 멀티스레드, 비동기 1

Hyo Kyun Lee·2024년 12월 24일

Java 대용량 데이터 처리 멀티 스레드 비동기

Java

목록 보기

67/87

1. 개요

개인프로젝트를 하면서 엑셀파일읽기 관련한 처리를 한 적이 있는데, 마침 실무에서도 관련한 건이 있었다.

이전에 할 당시에는 단순히 FileInputStream을 사용하여 처리하던 부분을 BufferedStream을 사용하여 메모리 효율 및 성능을 개선하였다.

그런데 더 알아보니까 멀티스레드, 비동기처리를 통해 개선할 수 있다는 것을 알게되었다.

지금까지 멀티스레드나 비동기처리는 애플리케이션 전체적으로, 다수의 유저가 애플리케이션을 실행할때 성능을 개선할 수 있는 방안으로 생각하였으나 특정 하나의 기능, 특히 I/O 관련한 (국소적인) 성능개선을 위해 생각할 수 있는 방안이었다는 것을 알게 되었다.

이 방안을 실무에 적용하고 이후에도 확장성있게 활용하기 위해 이 글을 기록한다.

2. 멀티스레드, 비동기처리를 통한 대용량 데이터 처리 성능 개선 - 이론

일단, 멀티스레드와 비동기 처리에 대한 개념은

멀티스레드 : 작업을 나누어서 한다, 병렬처리
비동기 처리 : 순차적으로 처리하지 않고, 선제 작업의 종료와 관계없이 처리한다.

단일 스레드 - 동기처리, 단일 스레드 - 비동기처리...여러가지 작업 방법이 있는데
가장 핵심은 멀티 스레드 - 비동기처리이다.

중요한 것은 어찌되었든 주어진 일을 모두 끝낸다는 점이고, 중간 중간 일처리를 서로 공유한다던가 침범을 한다는 위험성을 제거해주고 자원 활용률이 많이 늘어난다는 단점을 해결해야 한다.

무엇보다, 모던 자바 인 액션에서는 멀티스레드를 왜 사용하냐고 할 정도로 부정적으로 평가하였는데, 일단 적용해보고 판단해보는게 좋을 것 같다.

일단, 멀티스레드의 자원소비율을 높인다는 단점을 제거하기 위해 스레드풀이라는 개념을 사용하였다.

core thread, excessive thread를 사용하여 최대 자원 활용 수를 관리하고, 과부하할 경우 제거할 수 있다.
일을 할당받기 위한 스레드를 queue에 대기시킨다.

3-1. 멀티스레드, 비동기처리를 통한 대용량 데이터 처리 성능 개선 - 멀티스레드 원리 이해

멀티스레드를 지원해주는 라이브러리가 여러가지 있는데, 이 중 ExecutorService 내용이 있었고 이해가 쉬워서 한번 적용해보려고 한다.

특히 다수의 작업을 지휘할 수 있는 중앙처리장치가 있는, 개발자가 작업의 삭제 및 생성 등의 고려사항을 신경쓰지 않도록 하는 편리한 인터페이스라 하여 유심히 알아보았다.

엑셀파일을 읽어오고 파싱을 하는 부분을 멀티스레드로 처리한다면,

public List<JobRecommendResponse> recommendJobWithResume(InputStream pdf) throws IOException, CsvException {
	/*
    * NAS로부터 엑셀파일을 읽어온다.
    */
    Resource[] resources = resourcePatternResolver.getResources("classpath:_skillspr/*.csv");
 
    List<JobRecommendResponse> results = new ArrayList<>();
 	
    /*
    * ExecutorService를 통해 스레드풀 및 스레드 생성
    */
    ExecutorService executorService = Executors.newFixedThreadPool(resources.length);
    
    /*
    * 스레드풀에 스레드를 제출하고, 그 결과값을 future 객체를 반환받는다.
    * 이 결과를 futures에 저장하며, 멀티스레드이므로 다건의 결과 저장을 위해 List 변수 초기화.
    */
    List<Future<JobRecommendResponse>> futures = new ArrayList<>();
 
    // csv 파일 순회
    for (Resource resource : resources) {
        Future<JobRecommendResponse> future = executorService.submit(() -> {
            List<String[]> records;
            try (CSVReader csvReader = new CSVReader(new InputStreamReader(resource.getInputStream()))) {
                records = csvReader.readAll();
            } catch (Exception e) {
                log.info("멀티 스레딩 작업 중 에러");
                throw new CustomException(ErrorCode.MULTI_THREADING_ERROR);
            }
                
            return new JobRecommendResponse(resource.getFilename().substring(0, resource.getFilename().length() - 4), pr);
        });
        
        //futures 리스트에 future 객체 반환값(결과값) 저장
        futures.add(future);
    }

이때 스레드풀에서 한 스레드가 처리를 완료할때까지 대기하며, 완료 시 결과를 future로 반환하고 우리는 futureList에 이를 저장하여 완료시점을 알 수 있다.

JAVA 8부터는 CompletableFuture를 사용하며, 기존 future 사용 시 블로킹처리하므로 비동기 의미가 없어 사용을 거의 안한다고 한다.

3-2. 멀티스레드, 비동기처리를 통한 대용량 데이터 처리 성능 개선 - 비동기 원리 이해

비동기처리 시 @Async를 주입받아 사용하며, 보니까 async를 사용할 경우 기본적으로 멀티스레드 및 이를 관리할 스레드풀을 사용할 것이라는 전제가 있는 것처럼 보인다.

SimpleAsyncTaskExecutor 기본설정은 스레드를 계속 생성하여 스레드풀 관리가 일어나지 않으므로 사용자정의를 통해 스레드풀을 만들어야 한다.

@Configuration
@EnableAsync
public class AsyncConfig {
    private static int CORE_POOL_SIZE = 500; // 동시에 실행할 쓰레드의 갯수를 의미, default 값은 1이다.
    private static int MAX_POOL_SIZE = 3000; // 쓰레드 풀의 최대 크기를 지정, default 값은 Integer.MAX_VALUE
    private static int QUEUE_CAPACITY = 5000; // 큐의 크기를 지정, default 값은 Integer.MAX_VALUE 이다.
    private static String THREAD_NAME_PREFIX = "async-task";
 
    @Bean
    public Executor asyncTaskExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(CORE_POOL_SIZE);
        executor.setMaxPoolSize(MAX_POOL_SIZE);
        executor.setQueueCapacity(QUEUE_CAPACITY);
        executor.setThreadNamePrefix(THREAD_NAME_PREFIX);
        executor.initialize();
        return executor;
    }
}

이 사용자 정의한 비동기 DI를 주입한 메서드가 있다면,

@Slf4j
@Service
public class UserService {
 
    @Async("async-task")
    public void hello(){
        try {
            log.info("비동기 처리를 수행합니다.");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

위 비동기 메서드를 호출한다면 이 동작의 완전한 처리를 기다리지 않고, 다음 로직을 바로 진행한다(논블로킹, 비동기처리).

핵심은 "비동기처리한 메서드를 호출하는 것, 이를 위해 스레드풀 환경상태를 구성해주고 멀티스레드 처리를 사용하는 것.

3-3. 멀티스레드 + 비동기처리 pseudo code

List<CompletableFuture<JobRecommendResponse>> futures = new ArrayList<>();
 
    // csv 파일 순회
    for (Resource resource : resources) {
        CompletableFuture<JobRecommendResponse> future = <엑셀파일처리 비동기 메서드>
        futures.add(future);
    }
 
    // 완료될 때까지 대기
    for (CompletableFuture<JobRecommendResponse> future : futures) {
        try {
            results.add(future.join());
        } catch (Exception e) {
            log.info("멀티 스레딩 작업 중 에러");
            throw new CustomException(ErrorCode.MULTI_THREADING_ERROR);
        }
    }

엑셀파일처리를 하는 로직을 비동기 처리로 만든다.
그 결과를 스레드풀로 관리할 수 있는 futureList를 만들고, future에 add한다.
CompletableFuture 스레드풀에서 futures들을 모아서 각 작업이 완료될때까지 기다린다. future 객체를 반환하는 것과 논블로킹인 CompletableFuture 스레드풀을 만드는 것은 별개이고, 결국엔 futures 객체리스트를 스레드풀에 가져가서 멀티스레드의 작업을 관리해줄 수 있도록 만들어주는게 핵심인 것으로 보인다.

실무에 적용을 해보도록 한다. 적용이 힘들다면 일전에 만든 엑셀 다운로드 관련 개인프로젝트 소스를 이용하여 적용해보자.

그리고 성능분석을 위해 visualvm을 사용해보자.