Item80. 스레드보다는 실행자, 태스크, 스트림을 애용하라

[Mingadinga]

스레드를 직접 다루면 Thread가 작업 단위와 수행 매커니즘 역할을 모두 수행하게 된다. 그보다는 실행자 프레임워크를 사용해 작업 단위와 실행 매커니즘을 분리하고, 작업 수행을 위임하는 것이 좋다.

실행자 프레임워크

java.util.concurrent 패키지는 실행자 프레임워크 인터페이스 기반의 유연한 태스크 실행 기능을 담고 있다.

주요 API

• 생성 : ExecutorService exec = Executors.newSingleThreadExecutor();
• 실행할 태스크 넘기기 : exec.execute(runnable);
• 실행자 우아한 종료(종료 실패 시 VM 자체가 종료되지 않음) : exec.shutdown();
• 특정 태스크 완료 기다림 : get
• 태스크 중 아무거나 하나 완료 기다림 : invokeAny
• 모든 태스크 완료 기다림 : invokeAll
• 실행자 서비스 종료 기다림 : awaitTermination
• 완료된 태스크 결과 차례로 받음 : ExecutorCompletionService 이용
• 태스크를 특정 시간에 혹은 주기적으로 실행 : ScheduledThreadPoolExecutor 이용

CachedThreadPool과 ThreadPoolExecutor

실행자 서비스를 사용하기 까다롭다면, 스레드 풀을 만들어 사용할 수 있다.

CachedThreadPool

• 가벼운 서버에 적합
• 요청받은 태스크가 큐에 쌓이지 않고 즉시 스레드에 위임됨
• 무거운 프로덕션 서버에는 적합하지 않다. 새로운 태스크가 들어오는 족족 스레드가 생성되고 CPU 이용률이 100%로 치닫는다.

ThreadPoolExecutor

• 무거운 프로덕션 서버에 적합
• 스레드 개수를 완전히 통제할 수 있다.
• 스레드 풀에 지정한 개수만큼 스레드를 생성하고 풀에서 스레드를 꺼내 쓴다.

실행자 프레임워크 사용의 장점

• 스레드를 직접 다루면 Thread가 작업 단위와 수행 매커니즘 역할을 모두 수행하게 된다.
• 실행자 프레임워크를 사용하면 작업 단위와 실행 매커니즘을 분리된다.
• 실행자 프레임워크에 작업 수행을 위임하는 것이 더 안전하고 유연하다.

태스크 == 작업 단위

• 태스크는 작업 단위를 나타내는 추상 개념이다.
• 태스크는 Runnable과 Callable 타입이 있다. Callable은 값을 반환하고 임의의 예외를 던질 수 있는 Runnable이다.
• 실행자 서비스는 태스크를 수행하는 일반적인 매커니즘이다.
• 태스크 수행을 실행자 서비스에 맡기면 원하는 태스크 수행 정책을 선택할 수 있고, 생각이 바뀌면 언제든 변경할 수 있다.

포크 조인 풀 fork-join pool과 스트림

자바 7부터 실행자 프레임워크는 포크 조인 태스크를 지원한다.

• 작업 분할 : 병렬화할 수 있는 작업을 재귀적으로 작은 작읍으로 분할한 다음 서브테스크 각각의 결과를 합쳐 전체 결과를 만들도록 설계됨
• 작업 훔치기 : 스레드 간 작업 부하 비슷하게 유지

포크 조인 태스크는 포크 조인 풀이라는 특별한 실행자 서비스가 실행한다.

• ForkJoinTask 인스턴스는 작은 하위 태스크로 구성된다.
• ForkJoinPool을 구성하는 스레드가 이 태스크를 처리한다.
• 일을 먼저 끝낸 스레드는 다른 스레드의 남은 태스크를 가져와 대신 처리한다.
• CPU를 최대한 활용하면서 높은 처리량과 낮은 지연시간을 달성한다.
• 포크 조인 태스크를 직접 작성하고 튜닝하는 것은 매우 어려운 일이지만, 포크 조인 풀을 이용해 만든 병렬 스트림을 이용하면 적은 노력으로 그 이점을 얻을 수 있다.

import java.math.BigInteger;
import java.util.stream.LongStream;

public class ParallelPrimeCounting {
    // Prime-counting stream pipeline - parallel version (Page 225)
    static long pi(long n) {
        return LongStream.rangeClosed(2, n)
                .parallel()
                .mapToObj(BigInteger::valueOf)
                .filter(i -> i.isProbablePrime(50))
                .count();
    }

    public static void main(String[] args) {
        System.out.println(pi(10_000_000));
    }
}