[Java] 비동기 프로그래밍

[1] 개요

- 자바에서 멀티스레딩과 비동기 처리를 통해 애플리케이션의 성능을 향상시키고, 사용자 경험을 개선할 수 있음

- 멀티스레딩은 동시에 여러 작업을 처리할 수 있게 해주며, 비동기 처리는 작업의 완료를 기다리지 않고 다음 작업을 진행할 수 있게 해줌 (이러한 처리 방식은 특히 대규모 데이터 처리나 네트워크 통신과 같은 I/O 작업에서 효율이 높음)

- 자바에서는 ExecutorService, CompletableFuture, ForkJoinPool 등 다양한 방법으로 멀티스레딩과 비동기 처리를 구현 가능

 

1.  동기, 비동기

- 동기와 비동기를 구분하는 기준은 '작업 순서의 보장 여부'

 

1) 동기 (Sync)

- 작업의 순서가 보장됨

- 프로세스는 요청한 작업의 결과를 받고 나서 다음 작업 처리

 

2) 비동기 (Async)

- 작업의 순서가 보장되지 않음 (두 주체가 서로의 작업 시간 및 종료 시간에 영향을 받지 않고, 별도의 작업 시간/종료 시간을 가짐)

- 프로세스는 요청한 작업의 결과를 받지 않더라도 다음 작업 처리

- 현재 작업중인 스레드가 아닌 새로운 스레드를 만들어 해당 스레드에서 작업을 수행하는 것 (병렬적으로 테스크를 수행)

 

2. 스레드

- 스레드를 새로 만들고 폐기하는 것은 그만큼 CPU를 더 많이 사용하므로 많은 비용이 소모됨

- 그러므로 스레드 풀을 사용해서 미리 스레드를 만들어놓고 사용하는 게 효율적 (스레드 사용 후에는 스레드 풀로 반환)

 

* 스레드풀 - 스레드 개수를 미리 정해 놓고, 작업 큐에 들어오는 요청을 미리 생성해 놓은 스레드들에게 할당하는 방식

생성된 스레드 풀보다 실제로 들어온 요청의 수가 훨씬 적다면 나머지 할당되지 않은 스레드는 메모리만 차지하게 되므로 메모리 낭비가 발생할 수도 있음 

 

[2] 자바의 비동기 처리 기술 

- 자바에서는 ExecutorService, CompletableFuture, ForkJoinPool 등 다양한 방법으로 멀티스레딩과 비동기 처리를 구현 가능

 

1. ExecutorService

- ExecutorService는 자바에서 멀티스레딩을 구현하기 위한 표준 방법 중 하나로 이를 통해 개발자는 스레드 풀을 관리하고, 작업을 비동기적으로 실행할 수 있음

- 각기 다른 Thread를 생성해서 작업을 처리하고, 처리가 완료되면 해당 Thread를 제거하는 작업을 손수 진행 해야하는 것을 ExecutorService 클래스를 이용하면 쉽게 처리 가능

- ExecutorService 인터페이스는 작업의 진행을 통제하고 서비스 종료를 관리하기 위한 많은 메소드를 가지고 있으며, 이 인터페이스를 사용하여, 작업을 실행할 수 있고 리턴된 Future 인스턴스를 사용하여 실행을 제어할 수 있음

ExecutorService executorService = Executors.newFixedThreadPool(10);
Future<String> future = executorService.submit(() -> "Hello World");
// some operations
String result = future.get();

2. CompletableFuture

- CompletableFuture는 자바 8부터 도입된 비동기 프로그래밍을 위한 API (이를 사용해 복잡한 비동기 로직도 명확하고 간결하게 표현할 수 있음)

- Future와 CompletionStage 인터페이스를 구현

- 예를 들어, 여러 외부 API 호출 결과를 병렬로 처리하고, 모든 결과가 도착하면 최종 결과를 조합하는 로직을 CompletableFuture를 사용하여 구현할 수 있음 (이러한 방식은 코드의 가독성을 높이고, 유지보수를 용이하게 함)

-  Future와 CompletionStage 인터페이스를 구현하이를 통해 비동기 처리를 보다 간편하고 효과적으로 수행할 수 있는 방법을 제공합니다.

 

1) Blocking vs Non-Blocking

• Future는 get() 메서드를 통해 비동기 연산의 결과를 얻음. 하지만 이 메서드는 연산이 완료될 때까지 현재 스레드를 차단(blocking)하기 때문에 비동기 처리 결과를 기다리는 동안 현재 스레드는 대기 상태에 머무르게 됨
• CompletableFuture는 연산이 완료되면 자동으로 콜백 함수를 통해 결과를 반환, 이를 통해 현재 스레드에서 비동기 처리가 완료되기를 기다리는 동안에도 다른 작업을 계속 진행할 수 있음(이는 Non-Blocking 방식의 비동기 처리를 가능하게 함)
• 결론적으로, 복잡한 비동기 로직을 구현하거나, 비동기 처리 도중에 다른 작업을 진행해야 하는 경우에는 CompletableFuture를 사용하는 것이 더 유용

 

2) 비동기 작업 실행

- supplyAsync, runAsync

 

2-1) supplyAsync()

• supplyAsync() 메서드는 Supplier 함수형 인터페이스를 인자로 받아, 해당 함수형 인터페이스의 get() 메서드가 실행하는 작업을 비동기적으로 수행
• 이 메서드는 작업의 결과를 CompletableFuture 객체로 감싸서 반환하는데, 이 CompletableFuture 객체는 비동기 작업의 결과를 나타내며 작업의 상태 정보를 담고 있음

CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            // 비동기적으로 실행할 작업
            String result = "Hello, World!";
            return result;
        });

 

2-2) runAsync() 

• runAsync() 메서드는 Runnable 인터페이스를 인자로 받아, 해당 인터페이스의 run() 메서드가 실행하는 작업을 비동기적으로 수행
• 이 메서드는 작업의 결과를 반환하지 않으므로, 반환 타입은 CompletableFuture<Void>가 됨

CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
            // 비동기적으로 실행할 작업
            System.out.println("Hello, World!");
        });

 

 

** 또한, ExecutorService와 CompletableFuture를 함께 사용하면 더욱 강력한 비동기 프로그래밍이 가능

 

3) 작업 콜백

- thenApply, thenAccept, thenRun

 

3-1) thenApply

• 반환 값을 받아서 다른 값을 반환함 : 출력 가능

CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            // 비동기적으로 실행할 작업
            String result = "Hello, World!";
            return result;
        }).thenApply(s -> s.toUpperCase());

        try {
            System.out.println(future.get());
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }

 

3-2) thenAccept

• 반환 값을 받아 처리하고, 값을 반환하지 않음 

// thenApply는 결과를 반환해주고, thenAccept는 결과를 반환하지 않고 소비(처리)만 하기 때문에 아래와 같이 변경해야 함
CompletableFuture<Void> future = CompletableFuture.supplyAsync(() -> {
            // 비동기적으로 실행할 작업
            String result = "Hello, World!";
            return result.toUpperCase(); // 여기서 대문자로 변환
        }).thenAccept(System.out::println); // 변환된 결과를 소비

        try {
            // future.get()은 CompletableFuture가 완료될 때까지 기다림
            future.get();
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }

 

3-3) thenRun

• 반환 값을 받지 않고, 다른 작업을 실행함

CompletableFuture<Void> future = CompletableFuture.supplyAsync(() -> {
            // 비동기적으로 실행할 작업
            String result = "Hello, World!";
            String upperCaseResult = result.toUpperCase();
            System.out.println(upperCaseResult); // 여기서 결과를 출력
            return null; // 리턴값이 필요 없으므로 null 반환
        }).thenRun(() -> {
            // 별도의 작업 실행
            System.out.println("Task completed.");
        });

        try {
            future.get();
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }

 

3. ForkJoinPool

- ForkJoinPool은 자바 7에서 도입된 고성능 멀티스레딩 프레임워크

- 이는 큰 작업을 작은 작업으로 분할하고(Fork), 분할된 작업을 병렬로 처리한 후 결과를 합치는(Join) 분할 정복 알고리즘을 구현하기 위해 설계됨

- 예를 들어, 대규모 배열의 정렬, 복잡한 계산 작업 등을 ForkJoinPool을 사용하여 효율적으로 처리할 수 있음

- 특히, ForkJoinPool은 작업 도둑질(Work Stealing) 알고리즘을 사용하여 스레드들이 일을 고르게 분담하도록 하여, 멀티스레딩 환경에서의 성능을 최적화함

- 다만, ForkJoinPool의 사용은 적절한 상황에서만 효과적이며, 작업의 성격과 데이터의 크기를 고려하여 사용해야 함 : 따라서, ForkJoinPool을 사용하기 전에는 해당 작업이 병렬 처리에 적합한지 충분히 고려해야 함

 

 

출처: https://junghyungil.tistory.com/103

 

- 위의 그림과 같이 Fork를 통해 업무를 분담하고, Join을 통해 업무를 취합함

 

출처: https://junghyungil.tistory.com/103

 

- 하나의 작업 큐를 가지고 있으며, 서로 작업을 하려고 큐에서 작업을 가져감 (스레드들이 관리하고 있는 큐는 dequeue)

- 각 스레드들은 부모 큐에서 가져간 작업들을 내부 큐(inbound queue)에 담아 관리하며, 스레드들은 서로의 내부 큐에 접근하여 작업들을 가져가서 처리 

- 이러한 방법들은 놀고 있는 스레드가 없도록 하기 위해 도입됨