자바의 비동기 처리
동기
A라는 작업이 끝나는 동시에 B라는 작업을 시작한다.
비동기
A가 작업을 끝내든 말든 상관없이 B가 자신의 작업을 시작한다.
작업들이 서로의 작업 시작 및 종료 시간에 영향을 받지 않고 별도의 작업 시작/종료 시간을 가진다.
모든 비동기 방식은 멀티 스레드에서 작동한다.
Thread를 사용하지 않은 코드
멀티스레딩을 사용하지 않은 코드이다.
public class AsyncExample {
public static void main(String[] args) {
// 작업 1 - 1.5초 소요
System.out.println("작업1 시작");
try{
Thread.sleep(1500);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("작업1 종료");
// 작업 2 - 0.5초 소요
System.out.println("작업2 시작");
try {
Thread.sleep(500);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("작업2 종료");
}
}
실행 결과
작업1 시작
작업1 종료
작업2 시작
작업2 종료
작업1과 작업2는 서로 연관이 없고 작업2가 더 빨리 끝나지만 비돟기이므로 작업1이 끝난 후 작업2가 시작된다. 이런 비효율적인 경우에 스레드를 나누어 오래 걸리는 작업을 다른 주체에게 맡겨 다른 작업과 동시에 실행되게 만들면 시간을 아낄 수 있다.
작업1을 스레드로 변경한 코드
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class ThreadExample {
public static void main(String[] args) {
// 스레드 풀 생성
// 스레드 풀은 내부적으로 필요한 만큼 스레드를 만들어서 작업을 처리
ExecutorService executorService = Executors.newCachedThreadPool();
// 작업1 (스레드) -> 작업이 별도의 스레드에서 실행
executorService.submit(() -> {
log("작업 1 시작");
try {
Thread.sleep(1500);
} catch (InterruptedException e) {
e.printStackTrace();
}
log("작업 1 종료");
});
// 작업2 -> 그냥 메인 스레드에서 실행됨
log("작업 2 시작");
try {
Thread.sleep(500);
} catch (InterruptedException e) {
e.printStackTrace();
}
log("작업 2 종료");
executorService.shutdown();
}
}
실행 결과
main> 작업 2 시작
pool-1-thread-1> 작업 1 시작
main> 작업 2 종료
pool-1-thread-1> 작업 1 종료
이처럼 동시에 두 개의 작업이 서로 다른 스레드에서 실행되기 때문에 이 코드는 멀티스레딩을 사용한 코드이다.
자바의 비동기 처리
크게 Future 객체를 사용하는 방식과 Callback을 구현하는 방식이 있다.
Future 객체는 다른 주체에게 작업을 맡긴 상태에서 본 주체 쪽에서 작업이 끝났는지 직접 확인하는 방법이고,
Callback은 다른 주체에게 맡긴 작업이 끝나면 다른 주체 쪽에서 본 주체가 전해준 콜백 함수를 실행하는 방법이다.
Callback
콜백을 구현하는 방법에는 Completion Handler를 사용하는 방법과 함수형 인터페이스를 이용해서 구현할 수 있다.
그 외에도 많은 방법이 존재한다.
CompletionHandler
import java.nio.channels.CompletionHandler;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class CallbackExample1 {
private static ExecutorService executorService;
// CompletionHandler를 구현한다.
private static final CompletionHandler<String, Void> completionHandler = new CompletionHandler<>() {
// 작업 1이 성공적으로 종료된 경우 불리는 콜백 (작업 2)
@Override
public void completed(String result, Void attachment) {
log("작업 2 시작 (작업 1의 결과: " + result + ")");
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
log("작업 2 종료");
}
// 작업 1이 실패했을 경우 불리는 콜백
@Override
public void failed(Throwable exc, Void attachment) {
log("작업 1 실패: " + exc.toString());
}
};
public static void main(String[] args) {
executorService = Executors.newCachedThreadPool();
// 작업 1
executorService.submit(() -> {
log("작업 1 시작");
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
log("작업 1 종료");
String result = "Alice";
if (result.equals("Alice")) { // 작업 성공
completionHandler.completed(result, null);
} else { // 작업 실패
completionHandler.failed(new IllegalStateException(), null);
}
});
// 별개로 돌아가는 작업 3
log("작업 3 시작");
try {
Thread.sleep(1500);
} catch (InterruptedException e) {
e.printStackTrace();
}
log("작업 3 종료");
}
}
실행 결과
main> 작업 3 시작
pool-1-thread-1> 작업 1 시작
pool-1-thread-1> 작업 1 종료
pool-1-thread-1> 작업 2 시작 (작업 1의 결과: Alice)
main> 작업 3 종료
pool-1-thread-1> 작업 2 종료
CompletionHandler는 비동기 I/O 작업의 결과를 처리하기 위한 목적으로 만들어졌고 콜백 객체를 만드는 데 사용된다.
completed() 메소드를 오버라이드해서 콜백을 구현하고 failed() 메소드를 오버라이드해 작업이 실패했을 때의 처리를 구현하면 된다.
위의 코드를 보면 스레드 pool-thread-1 쪽에서 콜백 함술를 호출해 콜백도 계속해서 main 스레드와 별개로 비동기적으로 실행되는 걸 볼 수 있다.
함수형 인터페이스
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.function.Consumer;
public class CallbackExample2 {
private static ExecutorService executorService;
public static void main(String[] args) {
executorService = Executors.newCachedThreadPool();
// execute 함수의 인자로 callback의 구현체를 넣는다.
execute(parameter -> {
log("작업 2 시작 (작업 1의 결과: " + parameter + ")");
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
log("작업 2 종료");
});
// 별개로 돌아가는 작업 3
log("작업 3 시작");
try {
Thread.sleep(1500);
} catch (InterruptedException e) {
e.printStackTrace();
}
log("작업 3 종료");
}
public static void execute(Consumer<String> callback) {
executorService.submit(() -> {
log("작업 1 시작");
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
String result = "Alice";
log("작업 1 종료");
// 작업을 마친 후 인자로 받아온 callback의 구현체를 비동기로 실행한다.
callback.accept(result);
});
}
}
실행 결과
main> 작업 3 시작
pool-1-thread-1> 작업 1 시작
pool-1-thread-1> 작업 1 종료
pool-1-thread-1> 작업 2 시작 (작업 1의 결과: Alice)
main> 작업 3 종료
pool-1-thread-1> 작업 2 종료
작업1을 마친 뒤 callback으로 받아온 함수형 인터페이스를 실행하는 메소드를 호출한다.
유의할 점은 execute()의 인자로 execute()의 작업이 모두 끝난 뒤 실행될 콜백을 작성해 넣어야 한다는 점이다.
콜백의 타입은 인자를 받아서 사용하기 위해 Consumer를 사용한다.
Consumer<T>
T타입의 인자를 받고, 아무것도 리턴하지 않는 함수형 인터페이스이다.
콜백을 구현하는 데 있어 유의할 점
다른 스레드와 공유하는 변수 등에 접근할 경우 race condition이 발생할 수 있으므로
반드시 synchronized 블록 등의 기법을 통해 자원을 동기화해서 사용해야 한다.
Future
Future 객체를 사용한 비동기 처리 방식은 다른 주체에게 작업을 맡긴 상태에서 본 주체 쪽에서 작업이 끝났는지 물어보면서 직접 확인하는 방식이다.
확인하는 방법은 isDone()과 isCancel() 메소드가 있다. 블로킹 없이 작업을 완료했는지의 여부만 확인하는 방법이다.
또 다른 확인 방법은 get()으로 작업이 완료될 때까지 블로킹된 상태로 대기하는 방법이다. 오래 걸리는 작업을 다른 주체에게 맡겨 두고 get()을 호출하기 전까지 이 쪽에서 할 일을 하다가 작업을 마치면 get()을 호출해 작업의 결과를 받아오는 식으로 사용한다. get() 메소드를 통해 Future 객체에 담긴 작업 결과를 얻을 수 있다.
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
public class FutureExample {
public static void main(String[] args) {
ExecutorService executorService = Executors.newCachedThreadPool();
// 작업1 Callable이 리턴한 값을 future에 담는다.
Future<String> future = executorService.submit(() -> {
log("작업 1 시작");
Thread.sleep(1000);
log("작업 1 종료");
return "Alice";
});
log("작업 2 시작 (작업 1 종료 대기)");
String result = "";
try {
// 논블로킹으로 작업 1이 종료되었는지 확인한다.
log("작업 1 종료 여부: " + future.isDone());
// 블로킹 상태에서 작업 1이 끝날 때까지 대기한다.
result = future.get();
// 논블로킹으로 작업 1이 종료되었는지 확인한다.
log("작업 1 종료 여부: " + future.isDone());
} catch (InterruptedException | ExecutionException e) {
e.printStackTrace();
}
log("작업 1의 결과: " + result);
log("작업 2 종료");
}
}
실행 결과
main> 작업 2 시작 (작업 1 종료 대기)
pool-1-thread-1> 작업 1 시작
main> 작업 1 종료 여부: false
pool-1-thread-1> 작업 1 종료
main> 작업 1 종료 여부: true
main> 작업 1의 결과: Alice
main> 작업 2 종료
CompletableFuture
Future는 결국 다른 주체의 작업 결과를 얻어오려면 잠시라도 블로킹 상태에 들어갈 수 밖에 없기 때문에 사용하는 데 한계가 있다. 그래서 CompletableFuture가 등장했다.
import java.util.concurrent.*;
public class FutureExample {
public static void main(String[] args) {
new Thread(() -> {
try {
CompletableFuture
.supplyAsync(FutureExample::work1)
.thenAccept(FutureExample::work2)
.get();
} catch (InterruptedException | ExecutionException e) {
e.printStackTrace();
}
}).start();
work3();
}
private static String work1() {
log("작업 1 시작");
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
log("작업 1 종료");
return "Alice";
}
private static void work2(String result) {
log("작업 1의 결과: " + result);
log("작업 2 시작");
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
log("작업 2 종료");
}
private static void work3() {
log("작업 3 시작");
try {
Thread.sleep(1500);
} catch (InterruptedException e) {
e.printStackTrace();
}
log("작업 3 종료");
}
}
실행 결과
main> 작업 3 시작
ForkJoinPool.commonPool-worker-1> 작업 1 시작
ForkJoinPool.commonPool-worker-1> 작업 1 종료
ForkJoinPool.commonPool-worker-1> 작업 1의 결과: Alice
ForkJoinPool.commonPool-worker-1> 작업 2 시작
main> 작업 3 종료
ForkJoinPool.commonPool-worker-1> 작업 2 종료
스레드를 생성한 뒤 그 안에서 작업1을 supplyAsync()를 통해 호출하고 작업2를 작업1이 끝난 직후에 블로킹 없이 시작할 수 있도록 thenAccept()를 통해 호출한다.
CompletableFuture를 사용하면 이전 작업의 결과를 get()을 통해 블로킹으로 가져올 필요없이 then..() 함수를 통해 논블로킹을 유지하며 바로 사용할 수 있다.
Future vs CompletableFuture
| Future | CompletableFuture | |
| 등장 시기 | Java 5 | Java 8 |
| 비동기 처리 | 가능 (ExecutorService) | 가능 + 더 유연하게 구성 가능 |
| 블로킹 | get() 호출 시 블로킹 | get()은 블로킹이지만, 콜백 기반 처리로 논블로킹 가능 |
| 리턴값 처리 | get()으로 직접 꺼내야 함 | thenApply, thenAccept 등으로 체이닝&콜백 처리 가능 |
| 조합 | 불가능 | 여러 작업 조합(thenCombine, allOf, anyOf 등) 가능 |
| 예외 처리 | try-catch만 사용 가능 | exceptionally, handle 등으로 비동기 예외 처리 가능 |