Java Concurrent Programming - Executors, Callable, Future, CompletableFuture

백기선님의 인프런 더 자바8 강의 내용 정리.

---

Concurrent Programming

이전까지의 자바 Concurrent Programming을 알아야 한다.

자바의 기본 Concurrent Programming

Concurrent 소프트웨어

• 동시에 여러 작업을 할 수 있는 SW

자바에서는 멀티프로세싱, 멀티쓰레드 둘 다 가능하다. 이 중 멀티쓰레드를 다룸.

package com.inspirit941.java8to11;

public class testConcurrentProgramming {
    static class MyThead extends Thread {
        @Override
        public void run() {
            System.out.println("Thread: " + Thread.currentThread().getName());
        }
    }
    public static void main(String[] args) {
        // 메인 메소드가 작동하는 메인 쓰레드
        System.out.println("main Thread");
        System.out.println(Thread.currentThread().getName());
        // 메인쓰레드에서 다른 쓰레드를 생성하는 방법은 두 가지.

        // 1. 불편한 방법 - 쓰레드 상속받은 객체 생성
        MyThead myThead = new MyThead();
        myThead.start();
        // return 순서는 보장하지 않는다. 콘솔에 출력되는 순서는 매번 다름

        // 2. Runnable 구현체로 생성
        Thread thread = new Thread(() -> {
            while (true) {
                System.out.println("Thread: " + Thread.currentThread().getName());
                try {
                    // sleep : 현재 쓰레드를 대기상태로 두는 것.
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    // 자는 동안 누군가가 interrupt할 경우
//                    e.printStackTrace();
                    // 누군가가 쓰레드를 깨울 경우 종료하기.
                    return;
                }
            }
        });
        thread.start();

        // 1. sleep : 현재 쓰레드를 대기 상태로 두는 것.
        // 2. interrupt : 쓰레드가 외부의 개입을 받는 것. 이걸로 특정 쓰레드를 깨우거나 종료하는 등의 작업이 가능.
        // 종료시키는 메소드가 따로 있는 게 아니라, 이런 식으로 interrupt를 받았을 때 적용할 수 있는 방법 중 하나.

        try {
            Thread.sleep(3000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        // thread에 interrupt 명령어를 보내서 return하도록 설정.
        thread.interrupt();


        // 3. join : 다른 쓰레드를 기다리는 것.
        Thread thread2 = new Thread(() -> {
            System.out.println("Thread: " + Thread.currentThread().getName());
            try {
                // 2초 대기했다가 종료되는 쓰레드.
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                return;
            }
        });
        thread2.start();

        try {
            // 쓰레드가 작동을 끝낼 때까지 대기
            thread2.join();
        } catch (InterruptedException e) {
            // 해당 쓰레드를 대기하는 중에 interrupt될 경우
            // (이 경우는 메인쓰레드에 interrupt가 들어올 경우)
            e.printStackTrace();
        }
        System.out.println(thread2+ " is finished");

        // 다 좋은데, 쓰레드 두 개만 생성했는데도 코드량이 많다.
        // 개수가 많아질수록 코딩으로 쓰레드를 관리할 수가 없음
        // 그래서 등장한 게 Executor.
    }
}

Executors

Runnable이나 Thread처럼 Low-level API로 코드를 다루는 대신

보다 하이레벨의 Concurrency 프로그래밍을 의미함.

담당 작업

• 쓰레드 생성. 개발자는 Runnable 안에 로직만 제공한다.
• 쓰레드 생명주기 관리
• 작업 처리 및 실행을 담당하는 API 제공.

주요 인터페이스

• Executor : execute(Runnable). 직접 이 타입을 쓰진 않음.
• ExecutorService : Executor 상속받은 인터페이스. Callable도 실행 가능하며, Executor를 종료하거나 여러 Callable을 동시에 실행하는 기능 제공.
• ScheduledExecutorService : ExecutorService 상속받은 것. 특정 시간대에 작업을 실행하거나 / 주기적으로 작업을 실행하고 싶을 때 등등에 사용

package com.inspirit941.java8to11;

import java.util.concurrent.*;

public class testExecutors {
    public static Runnable getRunnable(String message) {
        return () -> System.out.println(message + Thread.currentThread().getName());
    }
    public static void main(String[] args) {
        // 쓰레드 하나만 쓰는 Executor : Executors.newSingleThreadExecutor();
        ExecutorService executorService = Executors.newFixedThreadPool(2);
        // 실행하기. execute 대신 submit을 써도 된다.
        // ex) executorService.execute(() -> System.out.println("Thread " + Thread.currentThread().getName()));
        executorService.submit(getRunnable("hello"));
        executorService.submit(getRunnable("hello2"));
        executorService.submit(getRunnable("hello3"));
        executorService.submit(getRunnable("hello4"));
        executorService.submit(getRunnable("hello5"));
        // 이렇게 되면, 2개의 쓰레드로 다섯 개의 실행요청을 처리한다.

        // ExecutorService에서 Thread Pool을 관리. 여기에 쓰레드 개수가 존재
        // Blocking Queue: 쓰레드가 해야 할 작업을 쌓아놓는 곳.
        // 이전 작업을 마치면 Queue에서 순차적으로 새 작업을 쓰레드에 할당하는 식으로 동작한다.

        executorService.shutdown();
        // ExecutorService는 한 번 실행해서 작업을 수행하면, 다음 작업이 들어올 때까지 대기상태가 된다.
        // 따라서 shutdown()을 명시적으로 해줘야 함.
        // shutdown (): Graceful shutdown. 실행하는 작업들 다 마친 후에 종료시키는 작업.
        // shutdownNow() : 그런 거 없이 실행 즉시 바로 종료.

        ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor();
        // 5초 뒤에 메소드 실행하도록 세팅
        scheduledExecutorService.schedule(getRunnable("hello"), 5, TimeUnit.SECONDS);
        scheduledExecutorService.shutdown();
        // 1초 대기 후 첫 실행, 2초마다 반복실행
        scheduledExecutorService.scheduleAtFixedRate(getRunnable("fixedRate"), 1, 2, TimeUnit.SECONDS);

        // fork / join 기능도 있지만, 이건 멀티프로세스 프로그래밍에서 유용한 개념.
    }
}

쓰레드에서 무언가 결과를 리턴받고 싶다면, Runnable은 사용할 수 없다. void return이기 때문.

자바 1.5 버전에서부터 Callable이 등장했다. 리턴 기능을 제공.

리턴값을 받아올 수 있는 무언가가 필요함. 그게 Future.

Callable과 Future

리턴값을 가질 수 있는 Runnable = Callable.

package com.inspirit941.java8to11;

import java.util.Arrays;
import java.util.List;
import java.util.concurrent.*;

public class testCallable {
    public static void main(String[] args) {
        ExecutorService executorService = Executors.newSingleThreadExecutor();

        Callable<String> hello = () -> {
            Thread.sleep(2000);
            return "hello";
        };
        // callable의 리턴 타입이 Function.
        Future<String> submit = executorService.submit(hello);
        System.out.println("Started");
        // get을 만나는 순간 메인쓰레드는 멈추고, submit이 연산을 마칠 때까지 기다린다.
        // isDone() : submit 쓰레드가 끝났는지 아닌지 확인하는 메소드
        System.out.println(submit.isDone()); // return false
        // cancel(bool) : 해당 쓰레드 종료 명령. 이렇게 종료된 service는 get()으로 결과를 가져올 수 없다.
        // true 넣으면 진행중인 작업을 interrupt하고 종료, false는 연산 끝날때까지 기다렸다가 종료

        try {
            String returns = submit.get();
            System.out.println(returns);
            // return hello
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }

        Callable<String> world = () -> {
            Thread.sleep(3000);
            return "world";
        };
        Callable<String> java = () -> {
            Thread.sleep(1000);
            return "java";
        };

        try {
            // 여러 개의 callable을 한번에 보내기. 리턴값으로 Future 리스트를 반환함.
            // invokeAll : 투입한 모든 Callable이 전부 끝날 때까지 대기함.

            List<Future<String>> futures = executorService.invokeAll(Arrays.asList(hello, java, world));
            futures.forEach(stringFuture -> {
                try {
                    System.out.println(stringFuture.get());
                    // return hello, java, world
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } catch (ExecutionException e) {
                    e.printStackTrace();
                }
            });
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        executorService.shutdown();

        // invokeAny : 모든 쓰레드가 끝날 때까지 기다리는 대신 뭐라도 하나 결과가 나오면 바로 리턴.
        // 리턴타입도 Future 안에 정의된 가상클래스.
        ExecutorService newExecutorService = Executors.newFixedThreadPool(4);
        try {
            // 쓰레드 4개, 메소드 3개 -> 각각의 스레드에 하나씩 메소드 할당됨.
            // 가장 먼저 리턴되는 결과값을 Future<클래스>의 "클래스" 타입으로 반환한다.
            String result = newExecutorService.invokeAny(Arrays.asList(hello, java, world));
            System.out.println(result);
            // return java
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }
        newExecutorService.shutdown();
    }
}

CompletableFuture

비동기 프로그램에 가까운 코딩이 가능해진다.

Future의 단점

• 외부에서 완료시킬 수 없음. get()에 타임아웃을 설정하는 식으로 해결해야 함
• 블로킹 코드 (ex)get)을 실행하지 않으면 작업이 끝났을 때 콜백 실행이 불가능
• 여러 Future를 조합할 수 없다. (event 정보를 가져온 뒤, 이벤트에 참석하는 회원 목록 불러오기 같은 기능 구현이 불가능)
• 예외처리 API 없음

ExecutorService executorService = Executors.newFixedThreadPool(4);
Future<String> future = executorService.submit(()->"hello");
// future 결과값을 가지고 어떤 작업을 하려면
// 무조건 get 코드 뒤에서 작업해야 함. = 비동기 코딩패턴을 만들기 어려웠음
future.get();

외부에서 Complete을 시킬 수 있어서 CompleteFuture.

package com.inspirit941.java8to11;

import java.util.concurrent.*;

public class testCompletableFuture {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        // 명시적으로 Executor 생성할 필요 없음.
        CompletableFuture<String> future = new CompletableFuture<>();
        // future의 기본값을 hello로 명시.
        future.complete("hello");
        System.out.println(future.get());
        // 위와 같은 기능을 수행하는 static 메소드.
        // CompletableFuture<String> future2 = CompletableFuture.completedFuture("hello");


        // 1. 리턴이 없는 경우 runAsync 메소드 사용.
        CompletableFuture<Void> future3 = CompletableFuture.runAsync(() ->{
            System.out.println("hello " + Thread.currentThread().getName());
        });
        future3.get();

        // 2. 리턴이 있는 경우 supplyAsync 사용.
        // 콜백을 주는 방법: .thenApply()
        // 따로 리턴을 하지 않을 경우 .thenAccept(Comsumer) 사용. 이 경우 CompleteFuture<Void>가 리턴타입이 됨.
        // 인풋을 따로 받지 않고 리턴값도 없을 경우 .thenRun(Runnable).
        CompletableFuture<String> stringCompletableFuture = CompletableFuture.supplyAsync(() -> {
            System.out.println("supplyAsync hello " + Thread.currentThread().getName());
            // currentThread 이름으로 ForkJoinPool이 리턴됨.
            return "SupplyAsync";
        }).thenApply((string) -> {
            return string.toUpperCase();
        });
        System.out.println(stringCompletableFuture.get());

        // Q. 따로 쓰레드 풀 정의도 하지 않았는데 어떻게 쓰레드를 생성해서 사용하는가
        // A. ForkJoinPool 때문. 자바 7에서 도입되었고, Executor 구현체 중 하나.
        // Deque 형태. 쓰레드가 자기 할 일이 없으면 직접 가져와서 처리 + 작업 단위를 서브테스크 단위로 쪼개서 다른 쓰레드에 분산처리 + join하는 것이라고 함

        // 물론, ThreadPool을 직접 정의해 사용하는 것도 가능하다.
        // runAsync나 supplyAsync 메소드의 두번째 인자값으로 executorService를 입력.
        // thenRun이나 thenApply같은 콜백메소드에 사용할 경우
        // 메소드 뒤에 Async 붙이고, 두번째 인자값으로 쓰레드 풀을 입력할 수 있음.
        ExecutorService executorService = Executors.newFixedThreadPool(4);
        CompletableFuture<Void> future2 = CompletableFuture.supplyAsync(() -> {
            return "fixedThreadPool" + Thread.currentThread().getName();
        }, executorService).thenRunAsync(() -> {
            System.out.println("thenRun");
        }, executorService);
        System.out.println(future2.get());

        executorService.shutdown();
    }
}

여러 작업을 조합해 사용하기, 예외처리하기

package com.inspirit941.java8to11;

import java.util.Arrays;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.stream.Collectors;

public class testCompletableFuture2 {
    private static CompletableFuture<String> getWorld(String message) {
        return CompletableFuture.supplyAsync(() -> {
            System.out.println("world " + Thread.currentThread().getName());
            return message + " world";
        });
    }

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        CompletableFuture<String> hello = CompletableFuture.supplyAsync(() -> {
            System.out.println("hello " + Thread.currentThread().getName());
            return "hello";
        });
        // 1. hello 끝나고 world해야 함. (의존성 존재하는 경우) : thenCompose
        // 이전까지는 hello.get(); world.get(); 형태로 수행
        // thenCompose 메소드를 사용할 수 있다.
        CompletableFuture<String> future = hello.thenCompose(s -> getWorld(s));
        System.out.println(future.get());

        // 2. 둘 사이에 연관관계가 없는 경우 - 비동기적으로 실행 (따로따로) : thenCombine
        CompletableFuture<String> world = CompletableFuture.supplyAsync(() -> {
            System.out.println("world " + Thread.currentThread().getName());
            return "world";
        });
        // hello와 world를 각각 실행한 후
        // 그 결과를 받아 처리하는 인터페이스 사용 = BiFunction
        CompletableFuture<String> future2 = hello.thenCombine(world, (h, w) -> {
            return h + " " + w;
        });
        System.out.println(future2.get());

        // 3. 2개 이상의 subtask를 모두 실행한 뒤에 다음 작업 진행하기
        // allOf()
        // 그런데, 각각의 subtask가 리턴타입이 다를 경우 / Interrupt로 종료될 경우 null을 리턴.
        CompletableFuture.allOf(hello, world)
                .thenAccept(System.out::println); // hello와 world 리턴타입이 다를 경우 null 리턴

        // 해결법
        List<CompletableFuture<String>> futures = Arrays.asList(hello, world);
        CompletableFuture[] futureArray = futures.toArray(new CompletableFuture[futures.size()]);

        CompletableFuture<List<String>> results = CompletableFuture.allOf(futureArray)
                .thenApply(v -> {
                    // thenApply로 넘어온 시점에서, 이미 쓰레드 각각의 연산은 다 끝난 상황.
                    return futures.stream()
                    // tip. get과 join은 기능상 동일, get보다 코드가 깔끔함.
                    .map(CompletableFuture::join)
                    .collect(Collectors.toList());
                });
        // results는 모든 쓰레드 연산이 끝났을 때 결과값을 받아오게 됨.
        // non-blocking방식으로 처리
        results.get().forEach(System.out::println);

        // 아무거나 먼저 결과나오는 것 빨리 받기. 출력하는 메소드 적용이므로 thenAccept
        CompletableFuture<Void> voidCompletableFuture =
                CompletableFuture.anyOf(hello, world)
                        .thenAccept((s) -> System.out.println(s));
        voidCompletableFuture.get();

        // 예외 처리
        // 비동기처리 task에서 에러 발생 시
        // exceptionally 메소드가 Exception 인자를 받은 뒤 필요한 로직을 처리하도록 생성.
        // handle()의 경우 정상적으로 종료될 시 / 그렇지 않을 경우를 나눠 연
        boolean throwError = true;
        CompletableFuture<String> throwExceptionFuture = CompletableFuture.supplyAsync(() -> {
            if (throwError) throw new IllegalArgumentException();
            System.out.println("hello" + Thread.currentThread().getName());
            return "hello";
        }).handle((result, error) -> {
            if (error != null) return "error";
            return result;
        });
//       .exceptionally(ex -> {
//            System.out.println(ex);
//            return "error";
//        });
        System.out.println(throwExceptionFuture.get());
    }
}