🛣️ Async – Spring에서 비동기 처리 구현하기

이 글에서는 Spring 서버에서 일부 로직을 비동기로 처리하는 방법에 대해 다룹니다.
구현 방법은 🧭 환경 설정 부터 나와있습니다.

비동기 처리를 해야 한다고요?

비동기.....? 아 나 진짜 들어봤는데

일단 나는 동기 처리와 비동기 처리를 생각하면서 구현해본 경험이 없었다. 물론 그게 뭔지 배우긴 했지만 학부 때에는 일반적으로 동기 처리만 사용하기에, 비동기 처리가 필요한 특수한 상황을 직접적으로 겪어본 적이 없었기 때문이다. 그런데 짜잔~ 그 특수한 상황이 생겼다. 그래서 일단 비동기 처리에 대한 개념을 알아보게 되었다. API 측면에서 내가 이해한 동기, 비동기 처리는 다음과 같다.

동기 처리(Synchronous)는 순차적으로 일을 처리하는 것이다. A 요청 후에 B 요청이 들어왔을 때, A 요청 처리가 완료되면 응답을 보내고 B 요청을 처리하는 식이다. 일반적으로 우리가 익숙한 방식이다.

비동기 처리(Asynchronous)는 응답과 무관하게 다음 일을 처리하는 것이다. 위와 같은 예시로 들어보자면, A 요청을 처리하는 동시에 B 요청을 처리한다. 각 요청의 응답은 각 요청의 처리가 완료된 시점에 발송된다. 이때 하나의 작업이 끝날 때까지 기다리지 않고 다음 작업을 수행한다는 점이 동기 처리와 명확한 차이점을 보인다.

그래서 이걸 왜 써야 한다고?

진행 중인 프로젝트에서 파일을 업로드하고, 해당 파일에 대한 인퍼런스 진행 후 결과값을 저장하는 API가 있다. 편의상 upload api라 부르겠다. upload api의 전체 로직을 실행하고 응답을 줬을 때 응답 시간이 너무 오래 걸렸다. 특히 인퍼런스 시간이 길었다. 2~3개의 파일을 업로드했을 때에는 괜찮았지만, 500~1000개의 파일을 올렸을 경우 시스템이 부하를 견디지 못하고 터지는 현상이 발생했다. 또한 대량의 파일을 업로드한 경우 응답 시간이 과도하게 오래 걸렸다.
이를 해결하기 위해 이미지를 저장하는 부분까지는 동기로, 이후의 인퍼런스 과정은 비동기로 처리하도록 설계했다. 아래의 로직을 처리하기 위해 하나의 API에서 동기 로직과 비동기 로직을 구분해 구현했다.

Async? 그거 어떻게 동작하는데?

Spring에서는 @Async 어노테이션으로 비동기 메소드를 설정한다. 해당 어노테이션은 Spring AOP에 의해 프록시 방식으로 작동된다.

Spring Context에 등록된 Async Bean이 호출되면 Spring이 개입해 해당 Async Bean을 프록시 객체로 Wrapping한다. 컨테이너에 의해 Bean으로 등록되는 시점에 프록시 객체화하는 것이다.
호출한 객체는 실질적으로 AOP를 통해 만들어진 프록시 객체화된 Async Bean을 참조하게 된다.

어떤 점을 조심해야 하는가

@Async를 사용할 때에는 아래의 유의사항을 지켜 사용해야 한다.

1. 비동기 메서드의 접근 지정자 private 사용 불가
2. self-invocation(자가 호출) 불가, 즉 inner method 사용 불가

비동기 메서드의 접근 지정자를 private으로 지정하면 AOP가 가로채 프록시 객체를 만들 때 이에 접근할 때 이에 접근할 수 없으므로 private method를 사용할 수 없다.
self-invocation의 경우, 프록시 객체를 거치지 않고 직접 메서드를 호출하기에 Async가 동작하지 않는다. 그렇기에 꼭 유의사항을 지켜서 사용해야 한다.

🧭 환경 설정

environment

• Spring Boot v3.3.1
• Gradle build

의존성 설정

기본적으로 spring-boot-starter-web에 내장되어 있지만, 그래도 의존성 설정이 잘 되어있는지 한 번 더 확인한다.

build.gradle

...
dependencies {
    implementation 'org.springframework.boot:spring-boot-starter-web'
}
...

AsyncConfig.java

비동기 설정을 위해 AsyncConfig 클래스를 만들어 관리해준다. 이때 @EnableAsync 어노테이션을 달아 비동기 메서드 처리를 할 수 있도록 아래와 같이 설정한다.

이 경우에는 서버가 터지지 않도록 느려도 안정적으로 운영하기 위해 CORE_POOL_SIZE와 MAX_POOL_SIZE를 1로 설정했다. 각 상수의 의미는 다음과 같다.

CORE_POOL_SIZE : 스레드 풀에서 기본적으로 유지되는 스레드 수 정의
→ 예상되는 최대 동시 작업 수에 가까운 값으로 설정하는 것을 권장

MAX_POOL_SIZE : 스레드 풀이 확장할 수 있는 최대 스레드 수 정의
→ 아래의 queue가 가득 찼을 때 새로운 요청이 들어온 경우, 확장할 스레드 풀 수

QUEUE_CAPACITY : 스레드 풀에서 사용할 최대 큐의 크기

package example.global.config;

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

/**
 * 비동기 처리를 위한 설정
 */
@EnableAsync
@Configuration
public class AsyncConfig {

    private static final int CORE_POOL_SIZE = 1;
    private static final int MAX_POOL_SIZE = 1;
    private static final int QUEUE_CAPACITY = 1000;

    /**
     * 비동기 처리를 위한 Executor 설정, Core Pool Size, Max Pool Size, Queue Capacity 설정
     *
     * @return ThreadPoolTaskExecutor
     */
    @Bean
    public ThreadPoolTaskExecutor threadPoolTaskExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(CORE_POOL_SIZE);
        executor.setMaxPoolSize(MAX_POOL_SIZE);
        executor.setQueueCapacity(QUEUE_CAPACITY);
        executor.setThreadNamePrefix("EXAMPLE-ASYNC-");
        executor.initialize();
        return executor;
    }
}

🚀 How to use

이 경우에는 향후 유지보수성과 코드의 유연성, 가독성을 위해 Async 메서드를 따로 관리하는 AsyncService 클래스를 별도로 구현해 관리했다. 비동기 메서드에 @Async 어노테이션을 달아주면 간단하게 구현할 수 있다.
비동기로 처리되는 메서드의 내부적으로 사용하는 하위 메서드가 있는 경우에는 다른 클래스에 @Async 어노테이션을 달아 구현해야한다. 동일한 클래스에 구현 시 self-invocation의 우려가 있다.

AsyncService.java

아래의 클래스는 인퍼런스를 진행하고, 그 결과를 저장하는 비동기 메서드를 구현한 것이다.

/**
 * Async service - to process async task
 */
@Service
@RequiredArgsConstructor
@Slf4j
public class AsyncService {

    private final StudyDao studyDao;
    private final PatientDao patientDao;
    private final InvalidInstanceDao invalidInstanceDao;
    private final InstanceDao instanceDao;
    private final SecondaryCaptureDao secondaryCaptureDao;
    private final ReportDao reportDao;

    private final InferenceManager inferenceManager;

    /**
     * Async process(1. save instance information in database,
     *               2. do inference and save result in database
     *               3. get secondary capture list
     *               4. get report
     *               5. save report in database)
     *
     * @param originPath origin dicom file path
     * @param accountEntity account entity
     */
    @Async
    protected void asyncProcess(String originPath, DicomTagValue dicomTagValue, AccountEntity accountEntity, Boolean isInvalid, String errorMessage) {

        String pngPath = FileManager.getInstanceThumbnailPath(originPath);

        // save instance information in database
        InstanceEntity instanceEntity = saveInstanceInfo(originPath, dicomTagValue, accountEntity, pngPath);
        log.info("instanceEntity id = {}", instanceEntity.getInstanceId());

        if (isInvalid) {
            invalidInstanceDao.saveInvalidInstance(instanceEntity, errorMessage);
            inferenceManager.getThumbnail(instanceEntity.getOriginDcmPath(),
                instanceEntity.getOriginPngPath());
            return ;
        }

        try {
            // do inference and save result in database
            InferenceResponseDto inferenceResponseDto = inferenceManager.getInferenceResult(instanceEntity, pngPath);

            // save inference result in database
            secondaryCaptureDao.saveWithSecondaryCaptureInfo(inferenceResponseDto, instanceEntity);

            // get secondary capture list and get report
            StudyEntity studyEntity = instanceEntity.getStudy();

            // get report
            ReportResponseDto reportResponseDto = inferenceManager.getReportFromStudyEntity(studyEntity);
            if (reportResponseDto == null) {
                log.error("Failed to get report");
                return ;
            }

            // save report in database
            reportDao.saveReportWithReportResponseDto(reportResponseDto, studyEntity);

        } catch (InferenceFailureException e) {
            invalidInstanceDao.saveInvalidInstance(instanceEntity, e.getMessage());
        } catch (Exception e) {
            log.error("Failed to get inference result", e);
        }
    }
    // private method들은 생략
}

이후 비동기 메서드를 service 로직에서 호출하면 자동으로 비동기 처리가 되는 것을 확인할 수 있다.
내 경우에는 인퍼런스를 진행하기 직전까지의 로직을 구현한 InstanceService.java 클래스에서 호출해줬다.

InstanceService.java

/**
 * Instance service
 */
@Service
@RequiredArgsConstructor
@Slf4j
public class InstanceService {

    private final AsyncService asyncService;
    private final DicomValidator dicomValidator;
    private final AccountDao accountDao;

    /**
     * Upload dicom image
     *
     * @param file      image file to upload
     * @param principal user data(email)
     */
    public void uploadInstance(MultipartFile file, Principal principal) {
        // 동기 로직 처리 생략
        
        // 비동기 로직 호출
        asyncService.asyncProcess(originPath, dicomTagValue, accountEntity, isInvalid, errorMessage);
    }
}

오~ 생각보다 더 위험한데~

여기서 더? 뭘? 어떻게?

앞서 언급했지만, 비동기 처리 시 스레드가 굉장히 중요하다. 그 중에서도 가장 주의해야 할 스레드 관련 이슈는 다음과 같다.

스레드 풀 관리

비동기 처리는 주로 별도의 스레드를 생성하여 작업을 처리한다. Spring에서는 @Async 사용 시 기본적으로 스레드 풀(Thread pool)을 사용하는데, 이를 제대로 관리하지 않으면 아래의 문제가 발생할 우려가 있다.

• 스레드 고갈(Thread exhaustion) : 스레드 풀이 꽉 차면 더 이상 새로운 작업을 처리할 수 없고, 대기 상태에 빠지거나 요청이 거부될 수 있다. 이를 방지하기 위해 스레드 풀의 크기와 작업 큐의 크기를 적절하게 설정해야 한다.
• 과도한 스레드 생성 : 스레드 풀이 너무 클 경우, 시스템 리소스(CPU, RAM 등)가 과도하게 소비되어 서버의 성능 저하를 유발할 우려가 있다.

이러한 문제들을 방지하기 위해 스레드 풀의 크기를 적절하게 설정해야 한다. 이는 CPU 코어 수와 시스템의 성능, 부하 분산 등을 고려해 결정해야 한다. 또한 작업 큐의 크기와 스레드 풀의 리젝션 정책을 적절히 설정하는 것을 통해 과도한 스레드 생성을 막을 수 있다.

스레드 안전성

비동기 처리에서는 여러 스레드가 동시에 작업을 수행하므로, 공유 자원에 접근할 때 스레드 안전성을 확보해야 한다. DB의 concurrency control과도 비슷한 문제이다. 동기화되지 않은 공유 자원에 여러 스레드가 동시에 접근 시 데이터 손상이나 예상하지 못한 동작이 발생할 우려가 있다.

이를 방지하기 위해 공유 자원에 접근할 때에는 synchronized 블록, ReentrantLock 등으로 동기화 처리를 한다. 가능하다면 불변 객체(Immutable object)를 사용하여 스레드 안전성을 보장할 수 있다. 또한 ConcurrentHashMap, CopyOnWriteArrayList 등과 같이 스레드에 안전한 자료 구조를 사용하는 것이 권장된다.

어떻게 보완해야 하는가

위에서 언급한 문제들 외에도 비동기 처리 시에는 리소스 관리, 의존성 관리 등의 문제가 발생할 우려가 있다. 이처럼 많은 위험을 안고 있는 방식을 사용할 때에는 이에 대한 테스트가 선행되어야 한다.
특히 스레드의 경우, 어떤 문제가 발생할 지 어느 정도 예측할 수 있기 때문에 이에 대한 보완책을 수립하고 적용한 후 테스트를 거쳐 내 환경에 맞는 설정을 찾는 것이 중요할 것으로 보인다.
리소스의 경우, 테스트 시 적절한 설정을 찾지 못했다면 혹은 원하는 성능이 나오지 않는데 비동기 요청은 써야 한다면 부하 분산을 위해 서버 증설 등의 대책도 고려해볼 수 있다.