멀티 스레드와 단일 스레드, 동기와 비동기 제대로 알고 쓰시나요? (feat. 성능 개선)

프로젝트 진행 중 직업 추천 api에서 반환까지 평균 6초 이상의 시간이 걸렸다. 100개가 넘는 직업 csv에 각각의 csv마다 6000개가 넘는 데이터가 있는데 이걸 전부 읽기 때문에 오래 걸리는게 당연했다.

성능 개선을 위해 멀티 스레드를 적용했었는데 프로젝트 마감 기한이 있었던지라 사용법만 익혀 적용했었다.

프로젝트가 끝나고 다시 돌아와 공부해보니 생각보다 어려워 이번 기회에 자세히 공부하고 정리해보았다.

비동기와 멀티스레드는 같은 뜻이 아닙니다.

비동기와 멀티스레드는 비슷하다 착각할 수 있지만 같은 뜻이 아니다.

의미를 간단하게 정리하자면 아래와 같다.

동기= 순서

스레드= 공간 or 일꾼

그럼 이제, 하나씩 살펴보자

동기 & 비동기

동기: 순차적으로 실행되는 것

비동기: 요청을 받은 뒤 먼저 작업이 끝난 순으로 실행되는 것

실생활에서의 동기, 비동기의 차이는 아래 사진과 같은 상황이다.

멀티 스레드 & 단일 스레드

단일스레드: 프로세스가 단일로 동작하는 것

멀티스레드: 프로세스가 다중으로 동작하는 것

그럼 경우의 수가 총 4가지가 생긴다

1. 단일 스레드 - 동기

2. 단일 스레드 - 비동기

3. 멀티 스레드 - 동기

4. 멀티 스레드 - 비동기

위의 4가지를 일상생활에서의 예시로 들면 이해가 쉽다.

1. 단일 스레드 - 동기

- 본점에서 음식을 주문 -> 조리 완료 -> 다음 주문 받음

2. 단일 스레드 - 비동기

- 본점에서 음식을 주문 -> 조리 및 주문 받기 -> 조리가 빠른 순서대로 제공

3. 멀티 스레드 - 동기

- 본점, 1호점, 2호점 .. 중 한곳에서 음식을 주문 -> 조리 완료 -> 다음 주문 받음

4. 멀티스레드 - 비동기

- 본점, 1호점, 2호점 .. 중 한 곳에서 음식을 주문 -> 조리 및 주문 받기 -> 조리가 빠른 순서대로 제공

이제 프로젝트에 본격적으로 적용해보자.

아래 코드는 사용자의 이력서로부터 가장 적합한 직업을 계산하여 정렬 후 반환하는 로직이다.

두가지 방법을 사용해서 성능을 개선시켜보았다.

    public List<JobRecommendResponse> recommendJobWithSkill(Long memberId) throws IOException, CsvException {
        List<String> skillList = skillService.findAllSkill(memberId).stream()
                .map(SkillResponse::getSkillName)
                .collect(Collectors.toList());
        String text = String.join(",", skillList);

        Resource[] resources = resourcePatternResolver.getResources("classpath:_skillspr/*.csv");

        List<JobRecommendResponse> results = new ArrayList<>();
        Set<String> userSkills = new HashSet<>(); // 추가

        // csv 파일 순회
        for (Resource resource : resources) {
            List<String[]> records;
            try (CSVReader csvReader = new CSVReader(new InputStreamReader(resource.getInputStream()))) {
                records = csvReader.readAll();
            }
            double pr = 0;
            boolean skip = true;
            // 개별 csv 데이터 순회
            for (String[] record : records) {
                // 0번째 행 skip
                if (skip) {
                    skip = false;
                    continue;
                }

                // skill 추출
                String skill = record[1];
                double probability = Double.parseDouble(record[5]);

                // skill 이 pdf 내용에 있는 단어일 때
                if (text.contains(skill) && (skill.length() > 1 || "C".equals(skill))) {
                    pr += Math.log(probability);
                    userSkills.add(skill); // 추가
                }
            }
            if (pr == 1) pr = 0;
            results.add(new JobRecommendResponse(resource.getFilename().substring(0, resource.getFilename().length() - 4), pr));
        }
        results.sort(Comparator.comparing(JobRecommendResponse::getProbability).reversed());
        return results;
    }

구현 1. ExecutorService 사용

처음 시도했던 건 java.util.concurrent패키지의 ExecutorService 를 이용하여 멀티쓰레드를 구현하였다.

ExecutorService란?

ExecutorService는 재사용이 가능한 ThreadPool로 Executor 인터페이스를 확장하여 Thread의 라이프사이클을 제어한다.

Thread를 활용하여 다수의 작업(Task)들을 비동기로 수행한다는 것은 간단하지 않다. Thread의 라이프사이클(생성과 제거 등)이나 발생할 수 있는 여러가지 low level의 고려사항들이 존재하는데 이를 개발자가 신경쓰지 않도록 편리하게 추상화한 것이 ExecutorService이다.

아래코드는 ExecutorService를 사용해 멀티쓰레드를 적용한 후의 코드이다.

public List<JobRecommendResponse> recommendJobWithResume(InputStream pdf) throws IOException, CsvException {
    // pdf -> text 추출
    String text = pdfToText(pdf);
    text = text.toUpperCase();
    final String finalText = text;
    Resource[] resources = resourcePatternResolver.getResources("classpath:_skillspr/*.csv");

    List<JobRecommendResponse> results = new ArrayList<>();
    Set<String> userSkills = new HashSet<>(); // 추가

    ExecutorService executorService = Executors.newFixedThreadPool(resources.length);
    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);
            }
            double pr = 0;
            boolean skip = true;
            // 개별 csv 데이터 순회
            for (String[] record : records) {
                // 0번째 행 skip
                if (skip) {
                    skip = false;
                    continue;
                }

                // skill 추출
                String skill = record[1];
                double probability = Double.parseDouble(record[5]);

                // skill 이 pdf 내용에 있는 단어일 때
                if (finalText.contains(skill) && (skill.length() > 1 || "C".equals(skill))) {
                    pr += Math.log(probability);
                    userSkills.add(skill); // 추가
                }
            }
            if (pr == 1) pr = 0;
            return new JobRecommendResponse(resource.getFilename().substring(0, resource.getFilename().length() - 4), pr);
        });
        futures.add(future);
    }

- newFixedThreadPool 메소드를 사용하여 고정 크기의 스레드 풀을 생성

- executorService.submit() : 스레드풀에 제출. 응답값으로 future을 받음.

- 작업이 완료될 때까지 대기하고 결과를 future로부터 가져옴.

for (Future<JobRecommendResponse> future : futures) {
    try {
        results.add(future.get());
    } catch (Exception e) {
        log.info("멀티 스레딩 작업 중 에러");
        throw new CustomException(ErrorCode.MULTI_THREADING_ERROR);
    }
}

구현 2. Async 사용

Async에 대해 자세히 알아보자.

@Async의 기본 설정은 SimpleAsyncTaskExecutor를 사용하도록 되어 있는데, 이것은 스레드 풀이 아니고 단순히 스레드를 만들어내는 역할을 한다.

기본 설정은 아래와 같다.

corePoolSize : 1
maxPoolSize : Integer.MAX_VALUE
keepAliveSeconds : 60(second)
QueueCapacity : Integer.MAX_VALUE
AllowCoreThreadTimeOut : false
WaitForTasksToCompleteOnShutdown : false
RejectedExecutionHandler : AbortPolicy

스레드를 재사용하지 않고 호출마다 새로운 스레드를 시작하며, AbortPolicy 이므로 처리할 수 없는 수준이 되면 Exception을 발생시키며 종료된다.

따라서 따로 커스텀 해주어야 한다.

@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;
    }
}

CorePoolSize : 기본적으로 생성해두고 실행 대기하는 Thread의 수
MaxPoolSize : 동시 동작하는 최대 Thread의 수
QueueCapacity MaxPoolSize 초과 요청에서 Thread 생성 요청시, 해당 요청을 Queue에 저장하는데 이때 최대 수용 가능한 Queue의 크기 Queue에 저장되어있다가 Thread에 자리가 생기면 하나씩 빠져나가 스레드 할당
ThreadNamePrefix : 생성되는 Thread 접두사 지정
initialize() : 생성하고 사용할 수 있도록하는 설정

반환값에 따른 동작

@Async가 붙으면 비동기로 동작하므로 메서드 반환 타입은 다음과 같다.

Future
Completable
FutureListenable
void

@Async 애너테이션으로 선언된 메서드는 리턴 타입에 따라 내부적으로 상이하게 동작한다.

void

@Slf4j
@Service
@RequiredArgsConstructor
public class CallService {

    private final UserService userService;

    public void call(){
        userService.hello();
        log.info("call");
    }
}

@Slf4j
@Service
public class UserService {

    @Async("HELLO_WORLD_EXECUTOR")
    public void hello(){
        try {
            Thread.sleep(1000);
            log.info("{}","hello");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

CallService에서 call 메서드가 UserService의 비동기 메서드 hello를 호출한다. 비동기 메서드이므로 call을 호출한 스레드는 call의 수행을 기다리지 않고 (논블로킹) 바로 log.info를 호출한다.

Future

메서드의 결과값은 전달받아야 한다면 Future을 사용해야 한다. 앞서 설명한 나머지 타입 모두 Future 계열에 해당하고 Future이 기본이 된다. Spring에서 제공하는 AsyncResult는 Future의 구현체이며 이를 사용해 Future 타입으로 리턴할 수 있다.

@Slf4j
@Service
@RequiredArgsConstructor
public class CallService {

    private final UserService userService;

    public void futureCall() {
        Future<String> future = userService.returnFuture();
        try {
            log.info("{}",future.get());
        } catch (Exception e) {
            e.printStackTrace();
        }
        log.info("origin void call");
    }
}

@Slf4j
@Service
public class UserService {

    @Async
    public Future<String> returnFuture(){
        try {
            Thread.sleep(1000);
            log.info("{}","async return future");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return new AsyncResult<>("async return future");
    }
}

futureCall 메서드에서 비동기 메서드 returnFuture을 호출하여 결과값을 get으로 꺼낸다. 비동기 메서드는 비동기로 동작하지만 future의 get 메서드는 메서드의 결과를 조회할 때까지 계속 기다리게 된다. 즉, 블로킹 현상이 발생합니다. Future의 경우 비동기 블로킹 방식이 되기 때문에 성능이 좋지 않아 잘 사용하지 않습니다.

CompletableFuture

@Slf4j
@Service
@RequiredArgsConstructor
public class CallService {

    private final UserService userService;

    public void completableFutureCall() {
        CompletableFuture<String> future = userService.returnCompletableFuture();
        try {
            future.thenAccept(s -> log.info("{}",s));
        } catch (Exception e) {
            e.printStackTrace();
        }
        log.info("origin completableFutureC call");
    }
}

@Slf4j
@Service
public class UserService {

    @Async
    public CompletableFuture<String> returnCompletableFuture(){
        try {
            Thread.sleep(1000);
            log.info("{}","async return CompletableFuture");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return CompletableFuture.completedFuture("async return CompletableFuture");
    }
}

기존에는 Future나 ListenableFuture를 이용하여 해결했지만 JAVA 8버전 부터는 CompletableFuture를 제공한다.

thenAccept 메서드로 결과값이 나오면 수행할 일을 정의해주었다. 즉, thenAccept 메서드를 통해 나중에 결과값이 나오면 처리하도록 하고 바로 다른 작업을 처리하러 갈 수 있는 논블로킹 으로 동작한다.

실제 프로젝트에선 다음과 같이 작성하였다.

public List<JobRecommendResponse> recommendJobWithResumeV2(InputStream pdf) throws IOException {
    long startTime = System.currentTimeMillis(); // 시작 시간 기록

    // pdf -> text 추출
    String text = pdfToText(pdf);
    text = text.toUpperCase();
    final String finalText = text;
    Resource[] resources = resourcePatternResolver.getResources("classpath:_skillspr/*.csv");

    List<JobRecommendResponse> results = new ArrayList<>();
    Set<String> userSkills = new HashSet<>(); // 추가

    List<CompletableFuture<JobRecommendResponse>> futures = new ArrayList<>();

    // csv 파일 순회
    for (Resource resource : resources) {
        CompletableFuture<JobRecommendResponse> future = processCsvFileAsync(resource, finalText, userSkills);
        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);
        }
    }

    long endTime = System.currentTimeMillis(); // 종료 시간 기록
    long duration = endTime - startTime; // 실행 시간 계산
    System.out.println("실행 시간: " + duration + "ms");

    return results;
}

@Async
public CompletableFuture<JobRecommendResponse> processCsvFileAsync(Resource resource, String finalText, Set<String> userSkills) {
    CompletableFuture<JobRecommendResponse> completableFuture = CompletableFuture.supplyAsync(() -> {
        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);
        }
        double pr = 0;
        boolean skip = true;
        // 개별 csv 데이터 순회
        for (String[] record : records) {
            // 0번째 행 skip
            if (skip) {
                skip = false;
                continue;
            }

            // skill 추출
            String skill = record[1];
            double probability = Double.parseDouble(record[5]);

            // skill 이 pdf 내용에 있는 단어일 때
            if (finalText.contains(skill) && (skill.length() > 1 || "C".equals(skill))) {
                pr += Math.log(probability);
                userSkills.add(skill); // 추가
            }
        }
        if (pr == 1) pr = 0;
        return new JobRecommendResponse(resource.getFilename().substring(0, resource.getFilename().length() - 4), pr);
    });

    return completableFuture;
}

비동기 멀티쓰레드로 변경 후 api 응답 속도 87%를 개선시킬 수 있었다.

'Spring Boot, JAVA 🌱' 카테고리의 다른 글

[Redis] Spring boot 에서 Redis로 성능 개선하기 (0)	2024.01.08
[Spring] AOP 개념 및 적용하기 (1)	2024.01.06
[HttpSession] session의 동작 원리 살펴보기 (0)	2023.12.30
AWS S3 이미지 관련 로직과 DB로직 트랜잭션 분리 (DB 커넥션 문제) (0)	2023.11.21
테스트 코드에 대한 정리 및 고민 기록 (1)	2023.10.29