[Spring] Spring에서 비동기 @Async 이해하기

1. 비동기 프로그래밍이 필요한 이유

비동기란 현재 처리중인 쓰레드에서 작업하던 Task를 다른 sub 쓰레드에게 Task를 위임하는 행위이다.

 

비동기를 사용하면 클라이언트는 서버에게 요청을 하고 응답을 기다릴 필요가 없다.

그로 인해 더많은 Task를 처리할 수 있고,  실시간 성 응답이 필요없는 경우 유용하게 사용할 수 있다.

 

비동기가 필요한 경우  예시

• 회원 가입 후 이메일 전송: 사용자는 회원가입을 완료하면, 이메일을 비동기적으로 처리하여 사용자는 회원가입만 완료되면 다른 작업을 처리할 수 있다.
• 알림 기능: 특정 이벤트가 발생한 후 사용자에게 알림을 비동기적으로 보내어, 프로세스가 멈추지 않고 계속 실행하도록 한다.

 

2. Thread Pool 정의

a. Thread Pool이란?

Thread Pool이란 미리 일정한 수의 Thread를 미리 생성해서 준비해두는 메커니즘이다. 

Thread 생성 비용은 비싸기 때문에 미리 필요한만큼 Thread를 Pool에 생성해놓고, 필요할 때 꺼내서 사용할 수 있다.

Thread Pool을 적용하면, Thread 생성 비용을 줄일 수 있다. 또한 과도한 Thread 생성을 방지할 수 있어 시스템 과부하 상태를 방지할 수 있다.

 

b. Thread Pool을 구성하는 요소

• corePoolSize: pool이 유지하는 기본 스레드 수
• maximumPoolSize: pool에서 생성 가능한 최대 스레드 수
• keepAliveTime: corePoolSize를 초과하는 스레드가 사용되지 않고 유지될 수 있는 시간
• workQueue: 실행할 작업을 대기 시키는 큐 

 

c. Thread Pool 작동 방식

1. 스레드 생성:
   • 작업이 들어올 때 필요한 만큼의 스레드를 생성하며, 요청이 계속 들어오면 스레드 수가 점차 corePoolSize까지 증가한다.
2. 요청 처리:
   • 요청이 들어오면, 현재 생성되어 있는 스레드 중 하나가 할당되어 해당 요청을 처리한다.
   • 모든 스레드가 작업 중일 경우, workQueue에 저장된다.
3. 대기열 사용:
   • corePoolSize보다 많은 작업이 들어오면, 추가 작업은 workQueue에 저장된다.
   • 대기열은 작업이 처리되기를 기다리는 큐 역할을 하며, 스레드가 작업을 처리할 준비가 될 때까지 작업을 보관한다.
4. 추가 스레드 생성:
   • 만약 workQueue가 가득 차게 되면, Thread Pool은 추가로 스레드를 생성하여 최대 maximumPoolSize까지 확장할 수 있다.
5. 스레드 제거:
   • 작업이 모두 완료되고 더 이상 처리할 작업이 없으면, corePoolSize를 초과하는 스레드는 keepAliveTime 동안 추가 작업을 기다린다.
   • keepAliveTime 동안 새로운 작업이 들어오지 않으면, 스레드는 종료되고 제거된다.

 

주의할 점은 maximumPoolSize 만큼 스레드가 생성되고, workQueue도 가득 차게 되면 더 이상 작업을 처리할 수 없기 때문에 새로운 작업 요청은 거부될 수 있다. 상황에 따라 무시하거나, 에러를 처리해주어야 한다. 

 

3.  스프링에서 @Async 적용

스프링 애플리케이션에서 @Async 비동기 프로그래밍을 적용하는 방법에 대해 알아보자. 

두 개의 ThreadPool을 생성하고,  서로 다른 Thread Pool을 사용하는 두 개의 비동기 메서드(작업)를 테스트 하는 코드를 살펴보자.  

 

a. Thread Pool 정의와 @EnableAsync

• 두 개의 Thread Pool을 생성하는 예제 코드이다.

@EnableAsync // Spring에서 비동기 처리 @Async를 활성화
@Configuration
public class AppConfig {
    // 빈 설정
    @Bean(name = "defaultTaskExecutor")
    public ThreadPoolTaskExecutor defaultTaskExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(200); // corePoolSize 설정
        executor.setMaxPoolSize(200); // maxPoolSize 설정
        executor.setKeepAliveSeconds(10); // keepAliveTime 설정
        executor.setQueueCapacity(100); // queue 최대 크기 설정
        return executor;
    }

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

 

b. @Async 비동기 메서드(반환값이 없을 때)

비동기 메서드에 반환값이 없으면, void를 리턴타입으로 두고 비동기 메서드를 정의하면 된다. 반환값이 없기 때문에 caller 입장에서는 호출만 한 다음 별 다른 처리를 할 필요가 없게 된다.

 

b-1. 반환값이 없는 비동기 메서드

• 서로 다른 스레드 풀을 사용하는 비동기 메서드를 정의하고, 호출하는 코드이다.

@RequiredArgsConstructor
@Service
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());
    }
}

@RequiredArgsConstructor
@Service
public class AsyncService {
    private final EmailService emailService;

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

 

 

b-2. 테스트

스프링 부트 통합테스트로 스프링 빈을 주입 받아 비동기 메서드가 제대로 동작하는 지 검사해보자.

@SpringBootTest
class AsyncApplicationTests {

	@Autowired private AsyncService asyncService;

	@Test
	void asyncCall_1_test() {
		asyncService.asyncCall_1();
	}
}

 

실행 결과

 

실행 결과를 보면, 서로 다른 Thread Pool을 사용하는 비동기 메서드가 적절하게 호출되는 것을 확인할 수 있다.

즉 작업중이던 스레드에서 sub Thread에게 작업을 위임하는 것을 확인할 수 있다. 

 

 

c. @Async 비동기 메서드(반환값이 있을 때)

비동기 메서드에 반환값이 있으면, Future, ListenableFuture, CompletableFuture 타입을 리턴 타입으로 사용할 수 있다.

 

c-1. Future 반환 티입 사용

• Future 는 블로킹 방식으로 동작하기 때문에 비동기 + 블로킹 방식이므로 성능이 좋지 않으므로 잘사용하지 않는다

@Service
public class EmailService {
    @Async
    public Future<String> sendMailWithFuture(int i) throws InterruptedException {
        System.out.println("task start : " + i);
        Thread.sleep(1000);
        return new AsyncResult<>("task end : " + i);
    }
}

@RequiredArgsConstructor
@Service
public class AsyncService {
    private final EmailService emailService;

    public void asyncFutureCall() throws InterruptedException {
        List<Future<String>> futures = new ArrayList<>();

        // 비동기 메서드 호출
        for (int i = 0; i < 10; i++) {
            futures.add(emailService.sendMailWithFuture(i));
        }

        // 결과 출력
        for (Future<String> future : futures) {
            try {
                System.out.println(future.get()); // Future는 블로킹하여 결과를 출력
            } catch (Exception e) {
                System.err.println("Exception: " + e.getMessage());
            }
        }
    }
}

 

 

c-2. ListenableFuture 반환 타입 사용

• future.get에서 블로킹하면서 작업이 완료될 때까지 결과를 기다리지만, 동시에 콜백함수를 등록해 작업이 완료되면 즉시 등록된 콜백함수를 실행한다.
• 즉, 비동기 작업이 완료되는 순서에 따라 콜백 함수를 실행한다.
• spring에서 제공하므로 스프링 프레임워크에 의존적이다.(spring 6부터 deprecated)

@Service
public class EmailService {
    @Async
    public ListenableFuture<String> sendMailWithListenableFuture(int i) throws InterruptedException {
        System.out.println("task start : " + i);
        Thread.sleep(1000);
        return new AsyncResult<>("task end : " + i);
    }
}

@RequiredArgsConstructor
@Service
public class AsyncService {
    private final EmailService emailService;

    public void asyncListenableFutureCall() throws InterruptedException {
        List<ListenableFuture<String>> futures = new ArrayList<>();

        // 비동기 메서드 호출
        for (int i = 0; i < 10; i++) {
            futures.add(emailService.sendMailWithListenableFuture(i));
        }

        // 결과 출력 및 예외 처리
        for (ListenableFuture<String> future : futures) {
            // 콜백 등록
            future.addCallback(
                    System.out::println, // 정상 완료시 콜백
                    ex -> System.err.println("Exception: " + ex.getMessage()) // 예외 처리 콜백
            );

            try {
                future.get(); // 작업 완료까지 대기
            } catch (Exception e) {
                System.err.println("Exception: " + e.getMessage());
            }
        }
    }
}

 

 

c-3. CompletableFuture 반환 타입 사용

• join에서 블로킹하면서 블로킹하면서 작업이 완료될 때까지 결과를 기다린다.
• 콜백함수를 등록해 작업이 완료되면 즉시 등록된 콜백함수를 실행한다.
• java 8 이상 포함.

@Service
public class EmailService {  
    @Async
    public CompletableFuture<String> sendMailWithCompletableFuture(int i) throws InterruptedException {
        System.out.println("task start : " + i);
        Thread.sleep(1000);
        return CompletableFuture.completedFuture("task end : " + i);
    }
}

@RequiredArgsConstructor
@Service
public class AsyncService {
    private final EmailService emailService;

    public void asyncCompletableFutureCall() throws InterruptedException {
        List<CompletableFuture<String>> futures = new ArrayList<>();

        // 비동기 메서드 호출
        for (int i = 0; i < 10; i++) {
            futures.add(emailService.sendMailWithCompletableFuture(i));
        }

        // CompletableFuture로 결과 출력
        for (CompletableFuture<String> future : futures) {
            // 콜백 등록
            future.thenAccept(System.out::println) // 작업 성공 시 콜백
                    .exceptionally(error -> { // 예외 처리 콜백
                        System.err.println("Exception: " + error.getMessage());
                        return null;
                    });
        }

        // 모든 작업 완료까지 대기
        CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
    }
}

 

 

4. @Async 작동 방식과 주의사항

a. @Async 동작 방식

1. @EnableAsync를 통해 스프링이 비동기 메서드를 처리하도록 설정

2. @Async 메서드가 붙은 메서드를 탐색하여 프록시 객체 생성

3.  프록시 객체는 AOP를 통해 실제 객체의 메서드 호출을 가로채어,  비동기적으로 실행한다.

 

동작 방식을 보면, 호출자(caller)는 실제 메서드를 호출하는 것이 아닌 스프링에서 생성한 프록시 객체를 통해 호출한다. 프록시는 호출된 메서드를 비동기적으로 실행한다. 

 

b. 주의 사항

@Async 동작 방식을 보면 스프링에서 제공하는 AOP 기능을 활용해 비동기 메서드를 지원한다. 그러므로 당연하게도 스프링 빈으로 등록된 객체의 @Async가 붙은 메서드만 비동기적으로 동작한다.  

 

예제를 살펴보자.

b-1. Spring Bean에서만 가능

예제코드 - 스프링 빈이 아닌 객체의 비동기 메서드를 호출한 경우 

@RequiredArgsConstructor
@Service
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());
    }
}

@RequiredArgsConstructor
@Service
public class AsyncService {
    private final EmailService emailService;

    public void asyncCall_2() {
        System.out.println("[asyncCall_2] :: " + Thread.currentThread().getName());
        EmailService emailService1 = new EmailService();
        emailService1.sendMail();
        emailService1.sendMailWithCustomThreadPool();
    }
}

@SpringBootTest
class AsyncApplicationTests {

	@Autowired private AsyncService asyncService;
    
	@Test
	void asyncCall_2_test() {
		asyncService.asyncCall_2();
	}
}

 

실행결과

 

실행결과를 보면 모두 같은 스레드에서 동작하는 것을 확인할 수 있다. 즉 @Async가 제대로 동작하지 않았다는 것이다. 

 

b-2. 내부 메서드 / private 메서드 불가

예제 코드 - 내부 메서드 및 private 메서드를 호출한 경우

@RequiredArgsConstructor
@Service
public class AsyncService {
    private final EmailService emailService;

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

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

@SpringBootTest
class AsyncApplicationTests {
	@Autowired private AsyncService asyncService;

	@Test
	void asyncCall_3_test() {
		asyncService.asyncCall_3();
	}
}

 

실행결과

 

실행결과를 보면 모두 같은 스레드에서 동작하는 것을 확인할 수 있다. 즉 @Async가 제대로 동작하지 않았다는 것이다. 

내부 메서드 / private 메서드 같은 경우도 프록시 영향을 받지 않으므로 비동기적으로 호출되지 않는다.

 

b-3. Thread Pool 초과 요청시 예외 핸들링

위에서 언급했듯이 작업을 처리할 Thread가 없으면 에러가 발생한다고 설명 했었다.

더 이상 작업을 처리할 Thread가 존재하지 않으면 예외가 발생하기 때문에 예외를 처리해줘야 한다. 

 

예제 코드 -  Thread Pool 초과 요청

Thread Pool size를 줄여서 비동기 요청을 시도해보자.

@EnableAsync
@Configuration
public class AppConfig {
    @Bean(name = "smallTaskExecutor")
    public ThreadPoolTaskExecutor smallTaskExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(1);
        executor.setMaxPoolSize(2);
        executor.setQueueCapacity(5);
        return executor;
    }
}

@RequiredArgsConstructor
@Service
public class EmailService {
    @Async("smallTaskExecutor")
    public void sendMailWithSmallTaskExecutor() {
        System.out.println("[sendMailWithSmallTaskExecutor] :: " + Thread.currentThread().getName());
    }
}

@RequiredArgsConstructor
@Service
public class AsyncService {
    private final EmailService emailService;

    public void asyncCall_5() {
        System.out.println("[asyncCall_1] :: " + Thread.currentThread().getName());
        for (int i = 0; i < 100; i++) {
            emailService.sendMailWithSmallTaskExecutor();
        }
    }
}

@SpringBootTest
class AsyncApplicationTests {
	@Autowired private AsyncService asyncService;

	@Test
	void asyncCall_5_test() {
		asyncService.asyncCall_5();
	}
}

 

 

실행결과

작업을 처리하다가 Thread Pool에 더 이상 처리할 Thread가 없으면, TaskRejectedException이 발생한다.

 

Thread Pool을 사용한다면, 상황에 맞게 Thread Pool의 사이즈를 적절하게 조절해주어야 한다. 하지만 예기치 않는 상황에 대비해 try-catch나 에러 핸들링을 통해 예외를 꼭 처리해주도록 하자!!  

public void asyncCall_5() {
    System.out.println("[asyncCall_1] :: " + Thread.currentThread().getName());
    try {
        for (int i = 0; i < 100; i++) {
            emailService.sendMailWithSmallTaskExecutor();
        }    
    } catch (TaskRejectedException e) {
        // 예외 처리
    }
}