Java - 병렬처리 (Future 인터페이스 활용)

90만건 가까이 되는 데이터를 Elasticsearch에서 읽어 전처리 후 다시 Elasticsearch에 색인했었습니다.

이 과정에서 약 3~4시간정도 소요가 되는데 이를 어떻게하면 빠르게 처리할 수 있을까 고민하다가 여러개의 파일을 읽고 처리하는 과정에서 병렬처리를 활용하면 어떨까 생각이 들어 공부해보았습니다. 

 

Future 예시 

출처 : https://pjh3749.tistory.com/280

public class FutureMain {
    public static void main(String[] args) {
        ExecutorService executorService = Executors.newCachedThreadPool();
        Future<Double> future = executorService.submit(new Callable<Double>() {
            public Double call() throws InterruptedException {
                return someLongComputation();
            }
        });
        doSomethingElse();
        try {
            Double result = future.get(1, TimeUnit.SECONDS); // <--- 블록 방지
            System.out.println("result : " + result);
        } catch (InterruptedException e) {
            e.printStackTrace();
            // handle e
        } catch (ExecutionException e) {
            e.printStackTrace();
            // handle e
        } catch (TimeoutException e) {
            e.printStackTrace();
            executorService.shutdown();
            // handle e
        }
    }

    private static Double someLongComputation() throws InterruptedException {
        // do something
        System.out.println("someLongComputation");
        Thread.sleep(2000);
        return 1d;
    }

    private static void doSomethingElse() {
        // do something else
        System.out.println("doSomethingElse");

    }
}

 

결과

doSomethingElse
someLongComputation
java.util.concurrent.TimeoutException
at java.base/java.util.concurrent.FutureTask.get(FutureTask.java:204)
at main.FutureMain.main(FutureMain.java:16)

Process finished with exit code 0

 

 

Java 5 버전에 나온 비동기 연산을 할때 사용되는 인터페이스입니다. 시간이 오래걸릴 것 같은 작업을 Future를 활용해 작업을 하고 다른 작업을 할 수 있다는 장점이 있습니다. 



위 코드를 보면 ExecutorService와 Future 인터페이스가 같이쓰이는것을 볼 수 있습니다.

이는 스레드 Pool을 관리해주는 인터페이스이며 Executors를 통해 여러가지의 스레드 풀을 생성할 수 있습니다. 

CachedThreadPool
일정시간동안 스레드를 검색하여 60초 동안 동작이 없을 시 Thread Pool에서 제거한다. Thread의 제한 없이 무한정 생성되며, 해당 Thread가 작업이 60초간 없을 시 제거하는 방식이므로 삭제속도보다 생성되는 양이 더 많다면 효율적이지 못할 수 있기 때문에.
최대 스레드 생성개수, 타임아웃을 설정해주는 것이 바람직해보인다. 

FixedThreadPool
고정된 스레드 개수만큼 생성

SingleThreadExecutor
단일 스레드

 

이와같이 스레드를 만들어 ExecutorService 객체작업을 할당할 수 있습니다. 

할당 방법도 여러가지가 있는데 그 중 2가지를 알아보면 아래와 같습니다.

• execute() : 리턴타입이 없으며 Runnable 객체로 작업을 할당받는다. 
• submit() : Future 객체로 리턴을 하며 Callable 객체로 작업을 할당받는다. 

ExecutorService의 작업 할당과정에서 return 값을 Future 객체로 반환해주기 때문에 같이 쓰이는 것을 확인할 수 있습니다. 

 

Future를 활용하면 비동기적으로 작업을 처리할 수 있다는 장점이 있는데요

• 여러 연산을 결합하기 어렵다
• 비동기 처리 중에 발생하는 예외를 처리하기 어렵다

라는 문제가 있어 개선된 버전인 CompletableFuture 클래스가 Java 8에 업데이트 되었습니다.

 

 

Future vs CompletableFuture

출처 : https://11st-tech.github.io/2024/01/04/completablefuture/

Future	CompletableFuture
Blocking	non-blocking
여러 연산을 함께 연결하기 어려움	여러 연산을 함께 연결
여러 연산 결과를 결합하기 어려움	여러 연산 결과를 결합
연산 성공 여부만 확인할 수 있고 예외처리 어려움	execeptionally(), handle()을 통한 예외 처리

 

이어서 작성하면 좋겠지만

아직 예제 코드들을 충분히 돌려보지 않았기 때문에 다음글에 이어서 작성하도록 하겠습니다..!