Spring 비동기 프로그래밍

김기현·2022년 7월 24일

비동기 프로그래밍

비동기 프로그래밍?

비동기 프로그래밍이란 Async한 통신으로 실시간성 응답을 필요로 하지 않는 상황에서 사용합니다. 현재 작업의 응답이 끝남과 동시에 다음 작업이 요청되는 동기 프로그래밍과는 달리, 비동기 프로그래밍에서는 현재 작업의 응답이 끝나지 않은 상태에서 다음 작업이 요청됩니다. 비동기적 방식을 처리하는 방법을 사용하는 이유는 함수의 과정이 끝나기 전에 다음 프로세스로 진행될 수 있기 때문에 속도가 빠르다는 장점이 있습니다.

Spring에서 비동기 프로그래밍

Spring에서 비동기 프로그래밍을 진행하려면 우선 ThreadPool을 짚고 넘어가야 합니다. 왜냐하면 비동기는 Main Thread가 아닌 Sub Thread에서 작업이 진행되기 때문입니다. 즉 Java에서는 ThreadPool을 생성해 Async 작업을 처리합니다.

ThreadPool Option

ThreadPool을 생성할 때는 크게 다음과 같은 옵션이 주어집니다.
1. corePoolSize 2. maxPoolSize 3. workQueue 4. keepAliveTime

corePoolSize
시간 초과 없이 활성 상태를 유지하기 위한 최소 작업자 수 입니다.
maxPoolSize
풀에서 생성될 수 있는 최대 쓰레드 수 입니다. 대기열의 항목 수가 queueCapacity를 초과하는 경우에만 새 쓰레드를 생성하며 queueCapacity에 의존합니다.
workQueue
작업이 실행되기 전에 작업을 유지하는데 사용할 대기열입니다.
keepAliveTime
쓰레스 수가 core보다 많을 때 유휴 쓰레드가 종료되기 전에 새 작업을 기다리는 최대 시간입니다.

오라클 공식문서에 더 자세히 나와있습니다

그래서 아래와 같이 ThreadPoolExecutor 생성자를 표현할 수 있습니다.

ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQUeue<Runnable> workQueue)

아래의 코드는 corePoolSize 값은 5, maximumPoolSize는 10, keepAliveTime는 3(3초로 controll), 그리고 이 queue에는 50개의 task까지 담을 수 있다고 해석하면 됩니다. (keepAliveTime과 TimeUnit은 세트라고 보면 됩니다.)

ThreadPoolExecutor executorPool = new ThreadPoolExecutor(5, 10, 3, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(50))

ThreadPool 생성 시 주의사항

Thread 수
Thread에 대해 다룬 블로그에서 ThreadPool을 생성할 때 주의사항으로 아래와 같이 언급하였습니다.

쓰레드는 너무 많으면 해당 쓰레드가 CPU의 자원을 두고 경합하게 되어 처리속도가 늦어지고 너무 적으면 CPU 자원을 최적으로 활용하지 못하기 때문에 처리 속도가 늦어집니다. 적절한 수로 유지되는 것이 가장 효율적입니다.

따라서 CorePoolSize값을 너무 크게 설정할 경우 Side Effect로 CPU의 자원을 두고 경합하게 되어 처리속도가 늦어지고, 반대로 너무 작다면 CPU 자원을 최적으로 활용하지 못해 늦어질 수 있음을 예상할 수 있습니다.

IllegalArgumentException
아래 4가지 경우 중 한가지라도 해당될 때 IllegalArgumentException 예외가 발생합니다.

corePoolSize < 0
코어 쓰레드 수가 0개 미만인 경우
keepAliveTime < 0
Alive한 Time이 0초이기 때문에 시작되자마자 종료됨
maximumPoolSize <= 0
최대 쓰레드 풀 사이드가 0개인 경우
maximumPoolSize < corePoolSize
최대 풀사이즈가 코어의 수보다 작을 경우

ThreadPool 생성과 요청거절

Case1
Thread 수 < CorePoolSize
새로운 쓰레드를 생성합니다.
Case2
Thread > CorePoolSize
Queue에 요청을 추가합니다.
Case3
Queue Full && Thread 수 < MaxPoolSize
새로운 쓰레드를 생성합니다.
Case4
Queue Full && Thread 수 > MaxPoolSize
요청을 거절합니다.

Spring Async 환경 구성

깃허브에 자세히 나와있습니다.

Config

AsyncConfig

@EnableAsync annotaion으로 비동기처리를 해줄 수 있습니다.

package dev.backend.prac_aysnc.config;

import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.EnableAsync;

@Configuration
@EnableAsync
public class AsyncConfig {
}

AppConfig

두개의 쓰레드 풀을 정의합니다.

package dev.backend.prac_aysnc.config;

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

@Configuration
public class AppConfig {

    @Bean(name = "defaultTaskExecutor", destroyMethod = "shutdown")
    public ThreadPoolTaskExecutor defaultTaskExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(200);
        executor.setMaxPoolSize(200);
        return executor;
    }

    @Bean(name = "messagingTaskExecutor", destroyMethod = "shutdown")
    public ThreadPoolTaskExecutor messagingTaskExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(200);
        executor.setMaxPoolSize(200);
        return executor;
    }
}

그리고 위의 코드처럼 선언을 해줍니다.

Controller

package dev.backend.prac_aysnc.controller;

import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

import dev.backend.prac_aysnc.service.AsyncService;
import lombok.RequiredArgsConstructor;

@RestController
@RequiredArgsConstructor
public class AsyncController {

    private final AsyncService asyncService;

    @GetMapping("/1")
    public String asyncCall_1() {
        asyncService.asyncCall_1();
        return "success";
    }

    @GetMapping("/2")
    public String asyncCall_2() {
        asyncService.asyncCall_2();
        return "fail";
    }

    @GetMapping("/3")
    public String asyncCall_3() {
        asyncService.asyncCall_3();
        return "fail";
    }
}

Service 1 - AsyncService

package dev.backend.prac_aysnc.service;

import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;

import lombok.RequiredArgsConstructor;

@Service
@RequiredArgsConstructor
public class AsyncService {

    private final EmailService emailService; // emailService은 빈 주입을 받음
    // AsyncService는 EmailService를 참조

    // 빈 주입 O
    public void asyncCall_1() {
        System.out.println("[asyncCall_1] :: " + Thread.currentThread().getName());
        emailService.sendMail();
        emailService.sendMailWithCustomThreadPool();
    }

    // 빈 주입 X
    public void asyncCall_2() {
        System.out.println("[asyncCall_2] :: " + Thread.currentThread().getName());
        EmailService emailService = new EmailService(); // 새로운 클래스를 선언해서 인스턴스를 만듦
        emailService.sendMail();
        emailService.sendMailWithCustomThreadPool();
    }

    public void asyncCall_3() {
        System.out.println("[asyncCall_3] :: " + Thread.currentThread().getName());
        sendMail();
    }

    @Async
    public void sendMail() {
        System.out.println("[sendMail] :: " + Thread.currentThread().getName());
    }
}

Service 2 - EmailService

package dev.backend.prac_aysnc.service;

import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;

import lombok.RequiredArgsConstructor;

@Service
@RequiredArgsConstructor
public class EmailService {

    @Async("defaultTaskExecutor")
    public void sendMail() {
        System.out.println("[sendMail] :: "
                           + Thread.currentThread().getName());
    }

    @Async("messagingTaskExecutor")
    public void sendMailWithCustomThreadPool() {
        System.out.println("[sendMailWithCustomThreadPool] :: "
                           + Thread.currentThread().getName());
    }
}

쓰레드 풀 지정
@Async("defaultTaskExecutor"), @Async("messagingTaskExecutor")처럼 AppConfig에서 정의한 쓰레드풀을 사용하도록 명시해줄 수도 있습니다.

Spring에서 Async를 사용할 때 주의점

AsyncService Case 1
URI 엔드포인트가 1로 요청을 보냈을 때 "success"를 Return하도록 curl 명령어를 통해 요청을 보냅니다.그리고 로그를 보면 아래와 같이 나옵니다.로그를 보면 다 다른 쓰레드에서 요청이 처리가 된 것을 확인할 수 있습니다. 이는 즉 비동기로 작업이 완료되었다는 의미입니다.

아래는 이 과정을 표현한 그림입니다.

AsyncService Case 2
AsyncService Case 1과는 달리 빈을 주입받지 않은 경우입니다.
URI 엔드포인트가 2로 요청을 보냈을 때 "fail"를 Return하도록 curl 명령어를 통해 요청을 보냅니다.그리고 로그를 보면 아래와 같이 나옵니다.로그를 보면 모두 같은 쓰레드에서 요청이 처리가 된 것을 확인할 수 있습니다. 이는 즉 동기로 작업이 완료되었다는 의미입니다.

이 이유는 요청을 하면 Spring Container에서 Bean을 받아오는데, 받아왔을 때 Async 처리가 필요하다면 Proxy라는 개념이 들어가서 Bean을 한 번 래핑합니다. 하지만 AsyncService Case 2는 AsyncService코드에서 주석으로 언급했듯이 Bean이 아닌, 자바에서 만든 새로운 클래스이기 때문에 비동기 처리를 할 수 없습니다.

AsyncService Case 3
URI 엔드포인트가 3으로 요청을 보냈을 때 "fail"를 Return하도록 curl 명령어를 통해 요청을 보냅니다.그리고 로그를 보면 아래와 같이 나옵니다.마찬가지로 로그를 보면 모두 같은 쓰레드에서 요청이 처리가 되었으며 동기로 작업이 완료되었음을 확인할 수 있습니다.

분명 AsyncService에서 @Async annotation을 붙여서 비동기처리가 될 것 처럼 보입니다. 비동기 처리가 되지 않는 이유는 다음과 같습니다.

AsyncService Case 1에선 AsyncService이 이미 Bean이고, EmailService은 Bean에서 다른 Bean을 주입받았습니다. 요청이 들어온다면 다른 빈을 참조할 때 asyncCall_1요청이 들어오면 emailService라는 Bean을 나에게 주는데, Async로 동작을 해야 하니 순수한 Bean을 주는 것이 아닌 Proxy 기능이 래핑된 Bean을 줍니다.

AsyncService Case 3의 경우는 sendMail()이 내부에 선언되어 있습니다. 그래서AsyncService안에서 이미 Bean이 왔는데, Bean 안에서는 Proxy처럼 래핑을 할 수 없게 됩니다. Spring Container를 거쳐야 하는데, 내부에 선언되어있으니 갈 필요가 없어졌고(아래의 사진처럼 하나의 Proxy 객체 안에 있기 때문) Proxy 래핑을 받을 수 없게 됩니다.

바로 아래의 그림처럼 되어버립니다.