ch15. CompletableFuture와 리액티브 프로그래밍 컨셉의 기초

15.1 동시성을 구현하는 자바 지원의 진화

15.1.1 스레드와 높은 수준의 추상화

15.1.2 Executor와 스레드 풀

Java5 ExecutorService: task 제출과 실행을 분리할 수 있는 기능 제공

• 스레드의 문제
  • Java 스레드는 운영체제 스레드에 직접 접근함. 운영체제 스레드를 만들고 종료하려면 비싼 비용을 치러야 하며 숫자도 제한되어 있음.
  • 주어진 프로그램에서 사용할 최적의 자바 스레드 개수는 사용할 수 있는 하드웨어 코어의 개수에 따라 달라짐

 

• 스레드 풀 그리고 스레드 풀이 더 좋은 이유
  • ExecutorService: task를 제출하고 나중에 결과를 수집할 수 있는 인터페이스를 제공함
  • 장점: hw에 맞는 수의 task를 유지함과 동시에 많은 태스크를 스레드 풀에 오버헤드 없이 제출할 수 있음
  •  실행방식)
    • 아래의 메서드는 워커 스레드라 불리는 nThreads를 포함하는 ExecutorService를 만들고 이들을 스레드 풀에 저장함
    • 스레드 풀에서 사용되지 않은 스레드로 제출된 태스크를 먼저 온 순서대로 실행함
    • 태스크 실행이 종료되면 스레드를 스레드 풀로 반환
  • 즉, task(Runnable or Callable)를 제공하면 스레드가 이를 실행함

// 스레드 풀을 만드는 팩토리 메서드
ExecutorService newFixedThreadPool(int nThreads)

 

• 스레드 풀 그리고 스레드 풀이 나쁜 이유
  
  • 스레드를 직접 사용하는 것보다 스레드 풀을 사용하는 편이 대부분 좋음.
  • 주의사항
    • k개의 스레드를 가진 스레드 풀은 오직 k 만큼의 스레드를 동시실행 할 수 있음. 초과로 제출된 task는 큐에 저장하고 이전의 task 중 하나가 종료되기 전까지 스레드에 할당되지 않음. sleep 상태이거나, i/o를 기다리거나 네트워크 연결을 기다리는 task가 있다면 주의해야 함. 스레드를 차지하고 아무 일을 하지 않기 때문에 성능 저하가 생길 수 있음. 블록될 가능성이 있는 task는 되도록이면 스레드 풀에 제출하지 말 것
    • 프로그램 종료 전 모든 스레드 풀을 종료하기

 

 

15.1.3 스레드의 다른 추상화: 중첩되지 않은 메서드 호출

• 리액티브 프로그래밍: 다양한 소스에서 들어오는 데이터 스트림을 비동기적으로 합쳐서 해결하고자 하는 것
• 비동기 메서드: 메서드가 반환된 후에도 만들어진 task의 실행이 계속되는 메서드

 

15.1.4 스레드에 무엇을 바라는가?

• 프로그램을 작은 task 단위로 구조화해 병렬성의 장점을 극대화하도록 만드는 것

---

15.2 동기 API와 비동기 API

아래의 코드를 병렬화시키고자 한다.

int y = f(x);
int z = g(x);
System.out.println(y+z);

 

다음과 같이 코드를 작성할 수 있다.

class ThreadExample{
	public static void main(String[] args) throws InterruptedException{
    	int x = 1337;
        Result result = new Result();
        
        Thread t1 = new Thread(()->{result.left = f(x);};
        Thread t2 = new Thread(()->{result.right = g(x);};
        t1.start();
        t2.start();
        t1.join();
        t2.join();
        System.out.println(result.left + result.right);
    }
}

 

 ExecutorService로 스레드 풀을 설정했다고 가정한다면 다음처럼 코드를 구현할 수 있음

public class ExecutorServiceExample{
	public static void main(String[] args) throws ExectuionException, InterruptedException{
    	int x = 1337;
        
        ExecutorService executorService = Executors.newFixedThreadPool(2);
        Future<Integer> y = executorService.submit(()->f(x));
        Future<Integer> z = executorService.submit(()->g(x));
        System.out.println(y.get() + z.get());
        
        executorService.shutdown();
    }
}

 

두 병렬화 코드 모두 명시적으로 스레드를 시작하는 코드가 포함되어 있다. 이를 해결하려면 비동기 API라는 기능으로 API를 바꿔서 해결할 수 있다.

 

15.2.1 Future 형식 API

Future<Integer> f(int x); 
Future<Integer> g(int x);

/////////////////////////////////////////////////////////////////////////
Future<Integer> y = f(x);
Future<Integer> z = g(x);
System.out.println(y.get() + z.get());

• 일회성의 값을 처리하는데 적절함
• 계산이 오래 걸리는 메서드(수 밀리초 이상)에 활용하면 효율성이 향상 됨

 

15.2.2 리액티브 형식 API

void f(int x, IntConsumer dealWithResult);

• 대안은 f, g의 시그니처를 바꿔서 콜백 형식의 프로그래밍을 이용하는 것이다. f에 인수로 콜백(람다)을 전달해서 f의 바디에서는 결과가 준비되면 이를 람다로 호출하는 task를 만든다. f는 바디를 실행하면서 태스크를 만든 다음 즉시 반환하므로 코드 형식이 다음과 같이 바뀐다.

public class CallbackStyleExample{
	public static void main(String[] args){
    	int x = 1337;
        Result result = new Result();
        
        f(x, (int y)-> {result.left = y; System.out.println((result.left + result.right));});
        g(x, (int z)-> {result.right = z; System.out.println((result.left + result.right));});
    }
}

• 위의 코드는 정확성을 보장하지 않는다.
  • if-then-else 적절한 락을 이용해 두 콜백이 모두 호출되었는지 확인한 다음 println 호출
  • 리액티브 형식 API는 일련의 이벤트에 반응하도록 설계되었으므로 Future를 이용하는 것이 더 적절함

• 네트워크나 사람의 입력을 기다리는 메서드에 API를 잘 활용하면 효율성이 향상됨
  • 리소스 낭비 없이 효율적으로 하단 시스템 활용 가능

 

15.2.3 sleep(그리고 기타 블로킹 동작)은 해로운 것으로 간주

• sleep 메서드나 기타 블로킹 동작을 통해 실행이 중지된 스레드는 여전히 시스템 자원을 점유함. 운영체제가 task를 관리하므로 일단 스레드로 할당된 task는 중지할 수 없음
  
  • 블록 동작: 태스크가 어떤 동작을 완료하기를 기다리는 동작 / 외부 상호작용을 기다리는 동작
• 기다리는 상황을 없애려면 무엇을 할 수 있을까?
  • 기다리는 일을 만들지 말거나, 코드에서 예외를 일으켜서 처리

work1();
Thread.sleep(10000);
work2();

위의 코드는 워커 스레드를 점유한 상태로 아무것도 하지 않고 10초간 멈춘다. 그리고 깨어나서 work2를 실행한 다음 작업을 종료하고 워커 스레드를 해제한다.

 

public class ScheduledExecutorServiceExample{
	public static void main(String[] args){
    	ScheduledExecutorService ses = Executors.newScheduledThreadPool(1);
        
        work1(); // 먼저 실행
        ses.schedule(ScheduledExecutorServiceExample::work2, 10, TimeUnit.SECONDS);
        // 10초 뒤, work2를 태스크로 제출
        
        ses.shutdown();
    }
}

위의 코드는 work1을 실행하고 종료하고, work2가 10초 뒤에 실행될 수 있도록 큐에 추가한다. 위위의 코드는 자는 동안 스레드 자원을 점유하지만 이 코드는 다른 작업이 실행될 수 있도록 허용한다.

 

15.2.4 현실성 확인

모든 동작을 비동기 호출로 구현한다면 병렬 하드웨어를 최대한 활용할 수 있지만, 현실적으로는 힘들다. 

자바의 개선된 동시성 API를 사용해보길 권장한다. 또, 네트워크 서버의 블록/비볼록 API를 일관적으로 제공하는 Netty와 같은 새로운 라이브러리를 사용해보는것도 도움이 된다.

 

 

15.2.5 비동기 API에서 예외는 어떻게 처리하는가?

• Future나 리액티브 형식의 비동기 API에서 호출된 메서드의 실제 바디는 별도의 스레드에서 호출됨. 이때 발생하는 어떤 에러는 이미 호출한 메서드의 실행범위와는 관계가 없는 상황이 된다. 예상치 못한 상황에서는 예외를 발생시켜야하는데 어떻게 구현할 수 있을가?
  • Future를 구현한 CompletableFuture: 런타임 get() 메서드에 예외를 처리할 수 있는 기능을 제공하며 예외에서 회복할 수 있도록 exceptionally()와 같은 메서드를 제공
  • 리액티브 형식의 비동기 API: return 대신 기존 콜백이 호출되므로 예외가 발생했을 때 실행될 추가 콜백을 만들어 인터페이스를 바꿔야함

void f(int x, Consumer<Integer> dealWithResult, Consumer<Throwable> dealWithException);

• 콜백이 여러 개면 한 객체로 메서드들을 감싸는게 좋음
  • ex) Java 9의 플로 API

void onComplete()
void onError(Throwable throwable)
void onNext(T item)
//////////////////////////////////////////////////////////////////

void f(int x, Subscriber<Integer> s);

/////////////////////////////////////////////////////////////////

s.onError(t);

• 이벤트/메시지: 여러 콜백을 포함하는 API들이 일련의 데이터를 만들어낸 다음 처리가 끝나면 마지막으로 끝났다는 알림을 만드는 호출

 

---

15.3 박스와 채널 모델

동시성 모델을 설계하고 개념화하기 위한 기법

---

15.4 CompletableFuture와 콤비네이터를 이용한 동시성

• CompletableFuture는 Future를 구현하고 있음
  • 실행할 코드 없이 Future를 만들 수 있음
  • complete() 메서드는 나중에 어떤 값을 이용해 다른 스레드가 이를 완료할 수 있고 get()으로 값을 얻을 수 있도록 허용함
  • thenCombine() 메서드를 사용해서 연산 결과를 효과적으로 더할 수 있음