@Async, TaskExecutor, CompletableFuture

https://docs.spring.io/spring-framework/reference/integration/scheduling.html

Task Execution and Scheduling :: Spring Framework

 

@Async 어노테이션과 CompletableFuture 에 대해 알아보겠다.

 

@Async 어노테이션은 Spring Framework 에서 사용할 수 있는 비동기를 쉽게 다루게해주는 어노테이션이다.

 

특징

- Spring 프록시 방식의 AOP 를 활용하여 메서드를 비동기로 호출하게 해준다.

따라서, 프록시 AOP 의 제약사항을 주의해야한다.

private 메서드에 적용 불가, 내부 메서드로 호출 시 적용 불가, @PostConstruct 와 함께 사용 불가 등

 

- @Async 어노테이션이 적용된 메서드를 호출하면 호출 스레드는 즉시 반환을 받고

해당 메서드 실행은 TaskExecutor 에 의해 수행된다. (프록시가 TaskExecutor 의 작업 스레드로 호출을 위임)

따라서, 쓰레드가 분리된 상황이라 발생한 예외는 호출 쓰레드로 전달되지 않아서 별도 처리가 필요하다.

그리고 ThreadLocal 를 이용한다면 이 역시 별도 처리가 필요하다.

 

 

 

코드로 바로 알아보자..

@Async 어노테이션을 통한 비동기 기능을 사용하기 위해서는

위와 같이 @EnableAsync 어노테이션을 설정해줘야 한다.

 

 

위는 가장 간단하게 @Async 어노테이션을 사용해보는 코드이다.

@Async 어노테이션이 적용된 asyncMethod 메서드는 별도의 스레드에서 해당 메서드가 실행될 것이다.

 

@Async 를 사용할 때, 별도로 TaskExecutor 를 설정하지 않으면

기본으로 SimpleAsyncTaskExecutor 를 사용한다.

이는 ... 비동기 메서드를 호출 할때마다 새로운 쓰레드를 생성하는 방식이다.

해당 방식은 실제 운영 환경에서는 비효율적이므로 스레드 풀 방식으로 해보도록 하자.

 

AsyncConfigurer 를 implements 하고 getAsyncExecutor 메서드를 오버라이딩하여

@Async 어노테이션이 동작할때 사용될 스레드를 ThreadPoolTaskExecutor 로 사용되도록 한다.

 

setCorePoolSize 메서드

기본적으로 대기해줄 쓰레드 수

 

setMaxPoolSize 메서드

동시에 동작할 최대 할당 가능 쓰레드 수

 

setQueueCapacity 메서드

최대 할당 가능 쓰레드 수가 초과 되면 큐에서 작업이 대기하게 되는데, 이때 대기할 큐의 사이즈

큐에도 가득차게되면 TaskRejectedException 예외가 호출 쓰레드에서 발생한다.

이를 처리하는 try-catch 가 필요 할 수도 있다.

 

setThreadNamePrefix 메서드

쓰레드 이름 prefix

 

<참고>

여러개의 Thread Pool 을 만들어 놓고.. @Async 별로 서로 다른 Thread Pool 을 적용하고 싶을 수도 있다.

빈을 위와 같이 등록하고

 

@Async 어노테이션을 사용할 때, 위와 같이 value 요소에 해당 빈 이름으로 적용하면 된다.

 

 

반환

지금 까지는 @Async 어노테이션을 적용한 메서드의 반환 타입은 void 였다.

@Async 비동기 쓰레드가 해당 메서드를 수행한 결과를

호출 쓰레드에 공유할 필요할 수도 있다.

이를 위해서 Java 나 Spring 이 제공하는 Future 를 상속 받는 애들을 사용할 수 있다.

대표적인 것들을 하나씩 알아보자..

 

결론 부터 알아보자면..

대부분의 경우 CompletableFuture 를 사용하면 된다.

 

Future

Future는 자바에서 제공하는 인터페이스로, 비동기 연산의 결과를 나타낸다.

Future를 사용하면 시간이 오래 걸리는 연산을 별도의 스레드에서 실행시키고, 그 결과를 나중에 받아올 수 있다.

즉, 메인 스레드가 블로킹되지 않고 다른 작업을 계속 진행할 수 있게 해준다.

 

Future 주요 제공 메서드

isDone()

비동기 연산이 완료되었는지 확인한다.

완료되었다면 true를, 아직 완료되지 않았다면 false를 반환한다.

 

get()

비동기 연산의 결과를 반환한다.

만약 연산이 아직 완료되지 않았다면, 완료될 때까지 블로킹된다. 

 

get(long timeout, TimeUnit unit)

비동기 연산의 결과를 반환한다.

연산이 완료될 때까지 블로킹되지만, 지정된 시간이 지나면 TimeoutException을 던진다.

 

isCancelled()

작업이 취소되었는지 확인한다.

취소되었다면 true를, 그렇지 않다면 false를 반환한다. 

 

cancel(boolean mayInterruptIfRunning)

작업을 취소한다.

만약 작업이 이미 시작되었는데, mayInterruptIfRunning 매개변수가 true라면 작업을 중단하려 시도한다.

 

AsyncResult

Spring Framework의 클래스로, 간단한 Future 인터페이스의 구현체이다.

비동기 계산의 결과를 담는데 사용된다.

Spring 6.0 부터 Deprecated 되었고, CompletableFuture 사용을 권장하고 있다.

 

코드로 간단히 알아보겠다.

@Async 어노테이션을 사용하여 TaskExecutor 의 작업 스레드로 비동기로 해당 메서드를 동작시킨다.

5초 후, AsyncResult 객체로 String 값을 호출 스레드로 전달 중이다.

 

호출 스레드는 비동기 메서드(asyncMethodWithFuture) 를 호출하고

Future<String> 타입의 객체를 즉시 반환 받는다.

isDone 메서드로 비동기 메서드가 완료되었는지 확인 한다.

확인이 되면 get 메서드로 해당 값을 사용할 수 있다.

 

<참고>

future.isDone() 을 사용하지 않고, future.get() 을 호출 하면

비동기 메서드 결과("Hello, World!")가 담길 때 까지 블로킹 된다.

 

ListenableFuture

Spring Framework 에서 제공하는 Future 를 상속받아 기능을 확장한 인터페이스이다.

비동기 작업이 완료되었을 때, 콜백을 통해 작업 스레드가 추가 작업을 수행하도록 할 수 있다.

Spring 6.0 부터 Deprecated 되었고, CompletableFuture 사용을 권장하고 있다.

 

 

CompletableFuture

Java 8에서 소개된 Future 인터페이스의 확장 버전이다.

기본적인 Future 는 결과가 준비되면 가져올 수 있다. (그 동안 블로킹)

반면에 CompletableFuture는 Future 외에 CompletionStage 도 구현하였다.

특정 시간 내에 완료되지 않으면 기본 값을 사용할 수 있거나,

결과가 준비되었을 때 콜백을 실행하도록 설정할 수 있다.

또한, 두 개 이상의 Future의 결과를 체인으로 연결하거나 조합하는 등의 기능을 제공한다.

 

CompletableFuture 주요 제공 메서드

CompletableFuture<Void> runAsync(Runnable)

비동기 작업을 시작하기 위해 사용되는 정적 메서드이다.

Runnable 인스턴스를 인자로 받아서 해당 작업을 비동기로 동작 시킨다.

Runnable 표준 함수형 인터페이스는

매개변수, 반환 값이 없는 작업을 정의하는데 사용되므로..

runAsync 메서드로 생성된 CompletableFuture 는 결과가 없는 작업에 사용된다.

CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
    // Long running task...
});

 

 

CompletableFuture<T> supplyAsync(Supplier<T>)

Supplier 인스턴스를 인자로 받아, 해당 작업을 비동기적으로 실행하고 결과를 반환한다.

Supplier 표준 함수형 인터페이스는

매개변수는 없고 반환값이 있는 작업을 정의하는데 사용되는 함수형 인터페이스이므로,

supplyAsync로 생성된 CompletableFuture는 작업의 결과(T)를 담게 된다.

CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
    // Long running task...
    return "Hello, World!";
});

 

<참고>

runAsync, supplyAsync 두 메서드 모두 ForkJoinPool.commonPool() 를 기본 실행자로 사용한다.

하지만, 특정 Executor 를 명시적으로 지정하는 오버로드된 버전의 메서드도 있다.

Executor executor = Executors.newFixedThreadPool(10);
CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
    // Long running task...
    return "Hello, World!";
}, executor);

 

 

CompletableFuture<R> thenApply(Function<T, R>)

CompletableFuture 의 결과(T)에

주어진 표준 함수형 인터페이스 인 Function<T, R> 의 인스턴스를 적용하므로

그 결과(R)를 포함하는 새로운 CompletableFuture를 반환한다.

CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
    return "Hello";
}).thenApply(s -> s + " World");

 

CompletableFuture<Void> thenAccept(Consumer<T>)

CompletableFuture의 결과(T)에

주어진 표준 함수형 인터페이스인 Consumer<T> 의 인스턴스를 적용하므로

결과(T)를 소비만 한다.

CompletableFuture.supplyAsync(() -> {
    return "Hello";
}).thenAccept(s -> System.out.println(s + " World"));

 

 

CompletableFuture<Void> thenRun(Runnable)

CompletableFuture 의 결과에

주어진 표준 함수형 인터페이스인 Runnable 의 인스턴스를 적용하므로

CompletableFuture 의 결과와 상관없이 주어진 작업을 실행한다.

CompletableFuture.supplyAsync(() -> {
    return "Hello";
}).thenRun(() -> System.out.println("Async finished."));

 

 

CompletableFuture<U> thenCompose(Function<T, CompletableFuture<U>>)

CompletableFuture 의 결과(T)에

주어진 표준 함수형 인터페이스인 Function<T, CompletableFuture<U>> 의 인스턴스를 적용하므로

새로운 CompletableFuture<U> 를 반환한다.

간단하게 말해서 비동기 작업의 결과를 이용해 또 다른 비동기 작업을 실행해야 하는 경우에 주로 사용된다.

(Java Stream 의 flatMap 과 in/out 구조가 비슷하다.)

CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
    return "Hello";
}).thenCompose(s -> CompletableFuture.supplyAsync(() -> {
    return s + " World";
}));

 

CompletableFuture<V> thenCombine(CompletableFuture<U>, BiFunction<T, U, V>)

두 개의 CompletableFuture 결과(T, U)를

주어진 표준 함수형 인터페이스인 BiFunction<T, U, V> 의 인스턴스를 적용하므로,

함수로 조합하고 반환 타입이 CompletableFuture<V> 이므로

최종 결과(V)를 포함하는 새로운 CompletableFuture를 반환한다.

CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {
    return "Hello";
});

CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> {
    return " World";
});

CompletableFuture<String> combinedFuture = future1.thenCombine(future2, (s1, s2) -> s1 + s2);

 

CompletableFuture<T> exceptionally(Function<Throwable, T>)

CompletableFuture 의 결과(T) 를 만들던 도중..

예외가 발생했을 때

수행할 표준 함수형 인터페이스인 Function<Throwable, T> 의 인스턴스를 적용하므로,

CompletableFuture 가 성공적으로 완료되었을 때와 같은 타입의 대체 값(T)을 제공해 줄 수 있다.

따라서 예외가 발생하더라도 기본 값을 바탕으로 체인을 계속해서 진행할 수 있다.

CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
    throw new RuntimeException("Exception occurred.");
}).exceptionally(ex -> "Default Value");

 

CompletableFuture<U> completedFuture(U Value)

이미 완료된 상태의 CompletableFuture 를 생성하는데 사용된다. 

지정된 값(U)을 결과로 가지는 CompletableFuture 를 즉시 반환한다. 

// "Hello, world!"라는 결과를 가진 CompletableFuture<String> 생성
CompletableFuture<String> future = CompletableFuture.completedFuture("Hello, world!");

// 비동기 작업이 없으므로 대기 없이 결과를 얻을 수 있다.
System.out.println(future.join()); // 출력: Hello, world!

 

알아본 메서드 종류 분류해보자..

비동기 작업 수행 관련

runAsync, supplyAsync

 

비동기 작업 콜백

thenApply, thenAccept, thenRun

 

작업 조합

thenCompose, thenCombine

 

예외 처리

exceptionally

 

생성

completedFuture

 

쓰레드 연습

코드와 동작 쓰레드를 이해하는 연습이므로 적절한 사용은 아니다.. (@Async 를 쓸 이유 없음)

해당 메서드를 호출하면 총 3개의 쓰레드가 동작한다..

- 해당 메서드를 호출한 호출 쓰레드

- @Async 어노테이션을 통한 쓰레드

- CompletableFuture.supplyAsync 메서드로 전달된 executor 쓰레드

 

동작 과정

1. 호출 쓰레드가 asyncMethodWithCompletableFuture 메서드를 호출

2. 호출 쓰레드는 메서드 실행을 @Async 쓰레드에 위임하고 

결과 처리를 위해 쓰레드간 공유되는 CompletableFuture 을 가지고 즉시 반환된다. (비동기)

3. 호출 스레드의 역할은 끝이다.

4. @Async 쓰레드는 CompletableFuture.supplyAsync 메서드를 호출

5. @Async 쓰레드는 supplyAsync 내부 로직 실행을 supplyAsync 의 executor 쓰레드에 위임한다.

6. @Async 쓰레드의 역할은 끝이며, @Async 쓰레드 풀에 반환된다.

7. supplyAsync 의 executor 쓰레드는 5초간 쓰레드를 재우고, 문자열을 반환하는 작업을 수행한다.

8. 작업이 완료되면 반환된 값은 supplyAsync 쓰레드가 CompletableFuture 에 넣는다.

 

메서드 호출 결과에 thenAccept 메서드를 적용하여 콜백을 등록했다.

 

동작 과정

1. 호출 쓰레드는 해당 메서드를 호출하고 CompletableFuture 를 즉시 반환 받는다.

2. 호출 쓰레드는 String 값("Request finished") 을 클라이언트에게 응답으로 내려준다.

3. 5초 후, supplyAsync 의 executor 쓰레드가 콜백을 수행한다. (CompletableFuture 결과 값 출력)

4. 이로써 supplyAsync 의 executor 쓰레드의 역할이 끝났다.

 

<참고>

위와 같이 Controller 에서 CompletableFuture 을 바로 반환해줘도 동작한다.

CompletableFuture 로 포장된 supplyAsync 의 executor 쓰레드의 작업의 결과 값이 클라이언트에게 전달된다.

 

<참고>

이 때, 실제 클라이언트에 응답을 전송하는 쓰레드는

Tomcat 쓰레드 일 수도.. Spring MVC 내부 쓰레드 일 수도 있다고 한다..(정확하지 않음, 알아봐야함)

 

 

활용

Web Application 에서 요청 쓰레드는 Tomcat 에서 관리하는 Thread Pool 에서 할당되어 처리된다.

클라이언트에서 요청이 왔는데 외부 API 를 호출해줘야 하거나..

해당 서버에서 굉장히 오래걸리는 작업을 수행해줘야한다거나..

이런 요청 상황에서..

 

1. 응답을 줄때 해당 비동기 작업 결과를 리턴안해도 된다거나..

 

2. 요청 쓰레드와 작업 쓰레드를 따로 분리해서 사용하면 좋을 것 같거나..

-> 즉, 요청 쓰레드 입장에서 비동기, 논블로킹 을 지원할 필요가 있을 때 사용하면 좋을 듯..

-> (요청 쓰레드가 비동기, 논블로킹이 실제로 되는지는 확인 해봐야한다.)