Java8) Java 비동기 프로그래밍과 CompletableFuture

Cf. Concurrent Software란?

• 동시에 여러 작업을 할 수 있는 소프트웨어를 의미합니다.
• Ex) 웹 브라우저로 유튜브를 보면서 키보드로 문서에 타이핑을 할 수 있는 SW
• Ex) 녹화를 하면서 코딩을 하고 워드에 적어둔 문서를 보거나 수정할 수 있는 SW

 

Java에서 제공하는 Concurrent Programming

 멀티 프로세싱과 멀티 스레딩으로 나뉩니다.

• Multi-Processing(ProcessBuilder)
• Multi-Thread

 

Java Multi-Thread Programming

• Thread 상속 또는 Runnable 구현(Runnable이 함수형 인터페이스로 변경되었기 때문에 Lambda로 작성 가능합니다.)

public class App {
    public static void main(String[] args) {
        // 
        Thread1 thread1 = new Thread1();
        thread1.start();

        Thread thread2 = new Thread(() -> {
            System.out.println("[Thread2] " + Thread.currentThread().getName());
        });

        thread2.start();

        System.out.println("[Main] " + Thread.currentThread().getName());
    }

    static class Thread1 extends Thread {
        @Override
        public void run() {
            System.out.println("[Thread1] " + Thread.currentThread().getName());
        }
    }

}

 

interrupt

Cf. void 메서드에서 return을 할 경우 작업을 끝냅니다.

public class App {
    public static void main(String[] args) throws InterruptedException {

        Thread thread = new Thread(() -> {
            while (true) {
                System.out.println("[Thread2] " + Thread.currentThread().getName());

                try {
                    Thread.sleep(1000L); // 다른 스레드에게 리소스 사용권이 우선됨
                } catch (InterruptedException e) {
                    System.out.println("inturrupted !");
                    return; // 해당 스레드 작업 종료
                }
            }
        });

        thread.start();

        System.out.println("[Main] " + Thread.currentThread().getName());
        Thread.sleep(3000L);
        thread.interrupt(); // InturruptedException을 발생시킨다.
    }
}

 

join

스레드가 다른 스레드의 작업이 끝나기를 기다리게 할 때에는 join()을 사용할 수 있습니다.

 

Executors

High-Level Concurrency Programming

• Thread를 만들고 관리하는 작업을 애플리케이션에서 분리하여 Executors에게 위임할 수 있습니다. (Low-Level에서 스레드를 관리하는 코드를 짜지 않아도 됩니다.)
• Thread 생성, Thread 관리, 작업 처리 및 실행(Thread로 실행할 작업을 제공할 수 있는 API 제공)이 가능합니다.
• 만약 Executor에 다수의 작업을 전달했을 때, Executor 내부의 스레드 풀에 그만큼의 개수에 해당하는 스레드들이 없을 경우에는 'Blocking Queue'에 작업들을 넣고 이를 순서대로 처리하게 됩니다.

 

주요 인터페이스

Executor

execute(Runnable r)을 사용해 Runnable 인스턴스만 제공하면 그 외의 일련의 작업은 Executor가 수행해줍니다.

ExecutorService

Executor를 상속받은 인터페이스. Callable도 실행할 수 있으며, Executor를 종료시키거나 여러 Callable을 동시에 실행할 수 있습니다.

• Executors의 static 메서드들로 Thread를 생성할 수 있습니다.
  • Ex) newSingleThreadExecutor() : 하나의 스레드를 사용하는 Executor를 생성합니다.
  • Ex) newFixedThreadPool(int n) : n개의 스레드가 존재하는 스레드 풀을 사용하는 Executor를 생성합니다.
• execute(Runnable r) 혹은 submit(Runnable r)을 통해 실행할 작업을 제공할 수 있습니다.
• 작업을 실행하고 나면 다음 작업이 들어올 때까지 대기하기 때문에 프로세스가 죽지 않습니다. 때문에 필요시 명시적으로 종료시켜 주어야 합니다.
• 종료시 shutdown()을 사용할 수 있습니다. 이 때 shutdown()은 'gracefule shutdown'에 해당합니다.
  Cf. graceful shutdown : 현재 진행중인 작업은 끝까지 마치고 종료한다는 의미
  • 현재 실행중인 작업에 상관없이 당장 종료시키기 위해서는 shutdownNow()를 사용할 수 있습니다.

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class App {
    public static void main(String[] args) {
        ExecutorService service = Executors.newSingleThreadExecutor();
        service.submit(() -> {
            System.out.println("[Thread] " + Thread.currentThread().getName());
        });
        executorService.shutdown();
    }
}

 

ScheduledExecutorService

ExecutorService를 상속받은 인터페이스. 특정 시간 이후 또는 주기적으로 작업을 실행할 수 있습니다.

• schedule(Runnable r, long delay, TimeUnit unit): delay(단위는 unit) 후에 실행될 작업을 전달할 수 있습니다.
• scheduleAtFixedRate(Ruinnable r, long delay, long period, TimeUnit unit): delay 후에 실행되어 period 간격으로 발생합니다.
  • Cf. service.shutdown()가 뒤따라 오게 되면, inturrupt가 발생하면 스레드가 멈추므로 사용하지 않습니다.

import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

public class App {
    public static void main(String[] args) {
        ScheduledExecutorService service = Executors.newSingleThreadScheduledExecutor();
        service.schedule(() -> System.out.println("Hello"), 1, TimeUnit.SECONDS);
        service.shutdown();
    }
}

 

Fork/Join Framework

Java 7에 들어온 Multi-Process Programming에 유용한 프레임워크입니다.

Fork/Join Framework는 Executor의 구현체 Work-Stealing Algorithm을 사용합니다. 이 말인 즉슨, Queue가 아닌 Deque를 사용하여 각 스레드는 수행할 작업이 없는 경우 스스로 Deque에서 작업을 가져가고,
자신이 파생시킨 세부적인 단위의 Task들을 다른 스레드에게 분산시켰다가 이를 다시 모아서 결과값을 도출하는 방식으로 동작합니다.

 

Callable

스레드에서 작업을 실행하고 결과를 가져오고 싶다면 void만 가능한 Runnable 대신 T 타입을 반환하는 Callable을 사용할 수 있습니다.

반환 타입의 유무만 다를 뿐 Runnable과 거의 동일하게 사용됩니다.
-> Future<T> ExecutorService.submit(Callable<T> c)와 같이 사용됩니다.

 

Future

1. isDone(): 작업 상태(가 완료되었는지)를 받을 수 있습니다
2. get(): 작업을 수행하고 그 결과를 반환한다. 작업이 끝날 때까지 기다렸다가 끝나야 이후 코드가 실행됩니다. (Blocking Call)
3. join(): get()과 같이 작업을 수행하고 그 결과를 반환한다. 다만 차이점은, get()이 Checked Exception을 반환한다면 join()은 Unchecked Exception을 반환합니다.
   • Cf. Checked Exception(RuntimeException 상속 X)은 반드시 예외 처리를 해야하고, Unchecked Execption(RuntimeException 상속)은 명시적으로 예외 처리를 하지 않아도 된다는 차이점이 있습니다.
4. cancel(Boolean cancel): 진행중인 작업을 취소할 사용됩니다.
   • 인터럽트 시킬 것인지 여부(현재 작업이 끝나지 않아도 종료할 것인지 여부)를 인자로 전달합니다.
   • Cf. cancel()을 하면 즉시 isDone()은 true가 되고, 실행 중인 작업 중단유무와 상관없이 get()으로 결과를 가져올 수 없습니다.

여러 개의 Callable 작업을 전달할 때

Cf. Runnable이 아닌 Callable만이 가능합니다.

• List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks): 리스트 형태로 Callable 목록을 전달해 Future 리스트를 결과로 받을 수 있습니다.
• T invokeAny(Collection<? extends Callable<T>> tasks): 리스트 형태로 Callable 목록을 전달해 가장 먼저 끝나는 Future의 결과만을 받을 수 있습니다.

 

CompletableFuture

• Cf. 기존에 있던 Future로는 다음과 같은 것들이 불가능합니다.
  • 작업을 조합할 수가 없습니다.
  • 외부에서 종료시킬 수도 없다. 취소하거나 get()에 timeout을 설정할 수는 있습니다.
  • 블로킹 코드(get())이 아니고서는 콜백을 실행할 수 없습니다. (get()이 완전히 실행되고서야 하단의 코드가 실행되는 구조)
  • 예외 처리 API를 제공하지 않습니다.
• Future와 마찬가지로 get(), join()을 통해 작업을 수행하고 결과를 반환받아야 합니다.
• Completable이 붙는 이유: 외부에서 Complete 시킬 수 있습니다. Ex) 일정 시간 내에 응답이 안오면 캐시해둔 값으로 대체하는 경우
  • complete(T t), CompletableFuture.completedFuture(T t) 를 통해서 CompletableFuture<T>의 기본값(t)을 정의하고 종료시켜줄 수 있습니다.
    • Cf. 이 때에도 값을 받아오기 위해 get()은 여전히 필요합니다.
• Executor를 만들지 않고, CompletableFuture만 가지고 비동기적으로 작업들을 실행할 수 있습니다.
  • 🔥 CompletableFuture를 사용하면 Executor를 생성하지 않고도 멀티스레드로 비동기 작업을 수행할 수 있습니다. 이는 내부적으로 Fork/Join framework의 CommonPool을 사용하도록 되어있기 때문입니다. - 출력시 ForkJoinPool로 출력
  • 명시적으로 정의한 스레드 풀을 사용하고 싶을 경우 runAsync(), supplyAsync()의 두 번째 인자로 Executor(스레드풀)을 전달할 수 있습니다. (콜백 기능을 위한 메서드들 - thenRunAsync, thenAcceptAsync, thenApplyAsync - 도 마찬가지)
• new 생성자를 통해 만들 수 있습니다.

비동기 작업 수행

• 리턴값이 없는 작업 수행 : runAsync(Runnable r)을 통해서 CompletableFuture<Void>를 생성할 수 있습니다.
• 리턴값이 있는 작업 수행 : supplyAsync(Supplier<T>)을 통해서 CompletableFuture<T>를 생성할 수 있습니다.

콜백 수행

• CompletableFuture<U> thenApply(Function<T, U> f): 결과로 받은 값의 타입을 변경할 수 있습니다.
• void thenAccept(Consumer<T> c): 파라미터를 가지나 리턴값이 없습니다.
  • Cf. void runAsync(Runnable r) + U thenAccept(Function <T, U> f) -> null 전달
• void thenRun(Runnable r): 파라미터를 가지지 않고 리턴값이 없습니다.

작업 조합

• thenCompose(CompletableFuture<T> c1): 두 작업간에 의존성이 있는 경우에 조합하는 경우(CompletableFuture<U> 리턴)
• thenCombine(CompletableFuture<U> c1, BiFunction<T, U, V> bf): 두 작업간에 의존성이 없지만 동시에 비동기적으로 실행하게끔 하는 경우(CompletableFuture<V> 리턴)
• allOf(CompletableFuture<T> c1, CompletableFuture<U> c2, ...): 두개 이상의 작업을 모두 조합하는 경우
• anyOf(CompletableFuture<T> c1, CompletableFuture<U> c2, ...): 두개 이상의 작업중 가장 먼저 끝난 작업의 결과를 도출하는 경우

예외 처리

• exceptionally(Function<Throwable>, T> f): 작업(들)의 내부에서 Exception이 throw될 경우 실행됩니다.
  • Cf. try-catch문의 catch와 유사합니다.
• handle(BiFunction<T, Throwable, U> bf): 정상적으로 종료되었을 때와 에러가 발생했을 때 모두 사용 가능합니다.
  • Cf. 두 개의 파라미터(1.정상적인 경우의 결과, 2.Exception)를 받아서 하나의 결과를 리턴합니다. - BiFuntion을 인자로 받습니다.