자바 스레드 풀 소개(Threadpool)

카테고리 없음

자바 스레드 풀 소개(Threadpool)

기록만이살길 2021. 3. 2. 09:39

1. 소개

이 기사는 표준 Java 라이브러리의 다양한 구현으로 시작한 다음 Google의 Guava 라이브러리를 살펴 보는 Java의 스레드 풀을 살펴 봅니다.

2. 스레드 풀

Java에서 스레드는 운영 체제의 리소스 인 시스템 수준 스레드에 매핑됩니다. 제어 할 수없는 방식으로 스레드를 생성하면 이러한 리소스가 빠르게 부족해질 수 있습니다.스레드 간의 컨텍스트 전환은 병렬 처리를 에뮬레이트하기 위해 운영 체제에서도 수행됩니다. 단순한보기는 더 많은 스레드를 생성할수록 각 스레드가 실제 작업을 수행하는 데 소요되는 시간이 줄어든다는 것입니다.스레드 풀 패턴은 다중 스레드 응용 프로그램에서 리소스를 절약하고 사전 정의 된 특정 제한에 병렬 처리를 포함하는 데 도움이됩니다.스레드 풀을 사용하는 경우

병렬 작업의 형태로 동시 코드를 작성하고 실행을 위해 스레드 풀의 인스턴스에 제출합니다

. 이 인스턴스는 이러한 작업을 실행하기 위해 재사용되는 여러 스레드를 제어합니다.

쓰레드풀

이 패턴을 사용하면

애플리케이션이 생성하는 스레드 수와

수명주기 를 제어 할 수 있을뿐만 아니라 작업 실행을 예약하고 들어오는 작업을 대기열에 보관할 수 있습니다.

3. 자바의 스레드 풀

3.1. Executors , Executor 및 ExecutorService

실행자의 헬퍼 클래스는 당신을 위해 사전 구성된 스레드 풀 인스턴스의 생성을위한 여러 가지 방법이 포함되어 있습니다. 이러한 클래스는 시작하기에 좋은 곳입니다. 사용자 정의 미세 조정을 적용 할 필요가없는 경우 사용하십시오.

집행자 및 ExecutorService의 인터페이스는 자바에서 다른 스레드 풀 구현과 사업에 사용된다. 일반적으로 스레드 풀의 실제 구현에서 코드를 분리하고 애플리케이션 전체에서 이러한 인터페이스를 사용해야합니다. 실행자 인터페이스는 하나가 실행 제출 방법 의 Runnable 실행을 위해 인스턴스를.

다음 은 Executors API를 사용하여 작업을 순차적으로 실행하기 위해 단일 스레드 풀 및 제한되지 않은 대기열로 지원되는 Executor 인스턴스 를 획득하는 방법에 대한 간단한 예 입니다 . 여기에서는 단순히 화면에 " Hello World "를 인쇄하는 단일 작업을 실행합니다 . 태스크는 Runnable 로 추론되는 람다 (Java 8 기능)로 제출됩니다 .

Executor executor = Executors.newSingleThreadExecutor();
executor.execute(() -> System.out.println("Hello World"));

ExecutorService를의 인터페이스는 방법 많은 수의 포함 서비스의 종료를 작업의 진행 상황을 제어 및 관리를 . 이 인터페이스를 사용하여 실행을 위해 작업을 제출하고 반환 된 Future 인스턴스를 사용하여 실행을 제어 할 수도 있습니다 .

다음 예제 에서는 ExecutorService를 만들고 작업을 제출 한 다음 반환 된 Future 의 get 메서드를 사용하여 제출 된 작업이 완료되고 값이 반환 될 때까지 기다립니다.

ExecutorService executorService = Executors.newFixedThreadPool(10);
Future<String> future = executorService.submit(() -> "Hello World");
// some operations
String result = future.get();

물론 실제 시나리오에서는 일반적으로 future.get ()을 즉시 호출하고 싶지 않지만 실제로 계산 값이 필요할 때까지 호출을 연기합니다.

제출 방법 중 취할 과부하 의 Runnable 또는 호출 가능을 (자바 8로 시작하는) 기능 인터페이스이며 람다로 전달 될 수있는 둘.

Runnable 의 단일 메서드는 예외를 throw하지 않으며 값을 반환하지 않습니다. 호출 가능 예외를 던질 값을 반환 우리를 허용하는 인터페이스는 더 편리 할 수 있습니다.

마지막으로 컴파일러가 Callable 유형을 추론하도록 하려면 람다에서 값을 반환하면됩니다.

ExecutorService 인터페이스 및 퓨처 사용에 대한 더 많은 예제는 “ A Guide to the Java ExecutorService ”를 참조하십시오.

3.2. ThreadPoolExecutor

있는 ThreadPoolExecutor은

미세 조정에 대한 매개 변수를 많이하고 후크 확장 스레드 풀 구현입니다.

여기서 논의 할 주요 구성 매개 변수는 corePoolSize , maximumPoolSize 및 keepAliveTime 입니다.

풀은 항상 내부에 보관되는 고정 된 수의 코어 스레드와 더 이상 필요하지 않을 때 생성 된 후 종료 될 수있는 과도한 스레드로 구성됩니다. corePoolSize를의 매개 변수는 인스턴스와 풀에 보관됩니다 코어 스레드의 수입니다. 새 작업이 들어올 때 모든 코어 스레드가 사용 중이고 내부 대기열이 가득 차면 풀이 maximumPoolSize 까지 증가 할 수 있습니다.

이 KeepAliveTime의 매개 변수 (의 초과 인스턴스화 과도한 스레드되는 시간의 간격 corePoolSize를가 ) 유휴 상태에 존재하는 수 있습니다. 기본적으로 ThreadPoolExecutor 는 제거 할 비 코어 스레드 만 고려합니다. 코어 스레드에 동일한 제거 정책을 적용하기 위해 allowCoreThreadTimeOut (true) 메서드를 사용할 수 있습니다 .

이러한 매개 변수는 광범위한 사용 사례를 다루지 만 가장 일반적인 구성은 Executors 정적 메서드에 미리 정의되어 있습니다 .

예를 들어 , newFixedThreadPool 메서드는 corePoolSize 및 maximumPoolSize 매개 변수 값 이 같고 keepAliveTime 이 0 인 ThreadPoolExecutor 를 만듭니다 . 이는이 스레드 풀의 스레드 수가 항상 동일 함을 의미합니다.

ThreadPoolExecutor executor = 
  (ThreadPoolExecutor) Executors.newFixedThreadPool(2);
executor.submit(() -> {
    Thread.sleep(1000);
    return null;
});
executor.submit(() -> {
    Thread.sleep(1000);
    return null;
});
executor.submit(() -> {
    Thread.sleep(1000);
    return null;
});

assertEquals(2, executor.getPoolSize());
assertEquals(1, executor.getQueue().size());

위의 예 에서 고정 스레드 수가 2 인 ThreadPoolExecutor 를 인스턴스화합니다 . 즉, 동시에 실행되는 작업 수가 항상 2 개 이하이면 즉시 실행됩니다. 그렇지 않으면 이러한 작업 중 일부는 차례를 기다리기 위해 대기열에 넣을 수 있습니다 .

우리는 1000 밀리 초 동안 수면을 취함으로써 무거운 작업을 모방하는 3 개의 Callable 작업을 만들었습니다 . 처음 두 작업은 한 번에 실행되고 세 번째 작업은 대기열에서 대기해야합니다. 작업을 제출 한 직후 getPoolSize () 및 getQueue (). size () 메서드 를 호출하여 확인할 수 있습니다 .

Executors.newCachedThreadPool () 메서드를 사용하여 미리 구성된 또 다른 ThreadPoolExecutor 를 만들 수 있습니다 . 이 메서드는 여러 스레드를 전혀받지 않습니다. corePoolSize를가 실제로 0으로 설정하고, maximumPoolSize를가 로 설정되어 Integer.MAX_VALUE를 이 인스턴스에 대해. 이것에 대한 keepAliveTime 은 60 초입니다.

이러한 매개 변수 값은 캐시 된 스레드 풀이 제출 된 작업의 수를 수용 할 수 있도록 제한없이 커질 수 있음을 의미 합니다 . 그러나 스레드가 더 이상 필요하지 않으면 60 초 동안 활동이 없으면 폐기됩니다. 일반적인 사용 사례는 애플리케이션에 수명이 짧은 작업이 많을 때입니다.

ThreadPoolExecutor executor = 
  (ThreadPoolExecutor) Executors.newCachedThreadPool();
executor.submit(() -> {
    Thread.sleep(1000);
    return null;
});
executor.submit(() -> {
    Thread.sleep(1000);
    return null;
});
executor.submit(() -> {
    Thread.sleep(1000);
    return null;
});

assertEquals(3, executor.getPoolSize());
assertEquals(0, executor.getQueue().size());

위 예의 대기열 크기는 내부적으로 SynchronousQueue 인스턴스가 사용 되기 때문에 항상 0 입니다. A의 SynchronousQueue는 , 한 쌍의 삽입 및 제거 큐가 실제로 아무것도 포함하지 않도록 작업은 항상 동시에 발생합니다.

Executors.newSingleThreadExecutor () API는 다른 일반적인 형태로 생성 ThreadPoolExecutor에 하나의 스레드를 포함한다. 단일 스레드 실행기는 이벤트 루프를 만드는 데 이상적입니다. corePoolSize를 하고 maximumPoolSize를 파라미터 1과 동일하고, 이 KeepAliveTime은 제로이다.

위 예제의 작업은 순차적으로 실행되므로 작업 완료 후 플래그 값은 2가됩니다.

AtomicInteger counter = new AtomicInteger();

ExecutorService executor = Executors.newSingleThreadExecutor();
executor.submit(() -> {
    counter.set(1);
});
executor.submit(() -> {
    counter.compareAndSet(1, 2);
});

또한이 ThreadPoolExecutor 는 변경 불가능한 래퍼로 장식되어 있으므로 생성 후 재구성 할 수 없습니다. 또한 이것이 ThreadPoolExecutor로 캐스트 할 수없는 이유 입니다.

3.3. ScheduledThreadPoolExecutor

로 스케줄은 확장 가능한 ThreadPoolExecutor의 클래스를 또한 구현 끝난 ScheduledExecutorService 몇 가지 추가 방법과 인터페이스를 :

일정 방법은 지정된 지연 후 작업을 한 번 실행할 수 있습니다.
scheduleAtFixedRate 메서드를 사용하면 지정된 초기 지연 후 작업을 실행 한 다음 일정 기간 동안 반복적으로 실행할 수 있습니다. 주기 인수는 시간 태스크의 시작 시간 사이에서 측정 실행 속도가 고정되어 있으므로;
scheduleWithFixedDelay 메서드는 주어진 작업을 반복적으로 실행한다는 점에서 scheduleAtFixedRate 와 유사 하지만 지정된 지연은 이전 작업의 끝과 다음 작업의 시작 사이에 측정됩니다 . 실행 속도는 주어진 작업을 실행하는 데 걸리는 시간에 따라 달라질 수 있습니다.

Executors.newScheduledThreadPool () 메소드는 일반적으로 만드는 데 사용되는 스케줄 할 소정와 corePoolSize를 , 무제한 maximumPoolSize를 제로 KeepaliveTime은이 . 500 밀리 초 내에 작업을 실행하도록 예약하는 방법은 다음과 같습니다.

ScheduledExecutorService executor = Executors.newScheduledThreadPool(5);
executor.schedule(() -> {
    System.out.println("Hello World");
}, 500, TimeUnit.MILLISECONDS);

다음 코드는 500 밀리 초 지연 후 작업을 실행 한 다음 100 밀리 초마다 반복하는 방법을 보여줍니다. 는 3 회 발사 때까지 작업을 예약 한 후, 우리는 기다려야 해, CountDownLatch의 잠금을 , 다음 사용을 취소 Future.cancel () 메소드를.

CountDownLatch lock = new CountDownLatch(3);

ScheduledExecutorService executor = Executors.newScheduledThreadPool(5);
ScheduledFuture<?> future = executor.scheduleAtFixedRate(() -> {
    System.out.println("Hello World");
    lock.countDown();
}, 500, 100, TimeUnit.MILLISECONDS);

lock.await(1000, TimeUnit.MILLISECONDS);
future.cancel(true);

3.4. ForkJoinPool

ForkJoinPool 은 Java 7에 도입 된 fork / join 프레임 워크 의 핵심 부분입니다 . 이는 재귀 알고리즘에서 여러 작업 을 생성 하는 일반적인 문제를 해결합니다 . 간단한 ThreadPoolExecutor를 사용하면 모든 작업 또는 하위 작업을 실행하려면 자체 스레드가 필요하기 때문에 스레드가 빠르게 소모됩니다.

A의 포크 / 조인 프레임 워크를, 어떤 작업 (산란 할 수 포크 ) 하위의 수와는 사용하여 완료 될 때까지 기다린 가입 방법. fork / join 프레임 워크 의 이점은 각 작업 또는 하위 작업에 대해 새 스레드를 생성하지 않고 대신 Work Stealing 알고리즘을 구현한다는 것입니다. 이 프레임 워크는 철저하게 문서에서 설명하는 " 포크에 가이드 / 자바 프레임 워크에 참여 "

ForkJoinPool 을 사용하여 노드 트리를 탐색하고 모든 리프 값의 합계를 계산 하는 간단한 예를 살펴 보겠습니다 . 다음은 노드, int 값 및 자식 노드 집합 으로 구성된 트리의 간단한 구현입니다 .

static class TreeNode {

    int value;

    Set<TreeNode> children;

    TreeNode(int value, TreeNode... children) {
        this.value = value;
        this.children = Sets.newHashSet(children);
    }
}

이제 트리의 모든 값을 병렬로 합하려면 RecursiveTask 인터페이스 를 구현해야합니다 . 각 작업은 자체 노드를 수신하고 해당 값을

하위 값의 합계에 더합니다 . 하위 값 의 합계를 계산하기 위해 태스크 구현은 다음을 수행합니다.

아이들 세트를 스트리밍하고 ,
이 스트림에 매핑 하여 각 요소에 대한 새 CountingTask 를 만듭니다.
분기하여 각 하위 작업을 실행합니다.
분기 된 각 작업에 대해 join 메서드를 호출하여 결과를 수집합니다 .
Collectors.summingInt 수집기를 사용하여 결과를 합산합니다 .

public static class CountingTask extends RecursiveTask<Integer> {

    private final TreeNode node;

    public CountingTask(TreeNode node) {
        this.node = node;
    }

    @Override
    protected Integer compute() {
        return node.value + node.children.stream()
          .map(childNode -> new CountingTask(childNode).fork())
          .collect(Collectors.summingInt(ForkJoinTask::join));
    }
}

실제 트리에서 계산을 실행하는 코드는 매우 간단합니다.

TreeNode tree = new TreeNode(5,
  new TreeNode(3), new TreeNode(2,
    new TreeNode(2), new TreeNode(8)));

ForkJoinPool forkJoinPool = ForkJoinPool.commonPool();
int sum = forkJoinPool.invoke(new CountingTask(tree));

4. 구아바에서 스레드 풀의 구현

Guava 는 인기있는 Google 유틸리티 라이브러리입니다. ExecutorService의 몇 가지 편리한 구현을 포함하여 많은 유용한 동시성 클래스가 있습니다. 구현 클래스는 직접 인스턴스화 또는 서브 클래 싱에 액세스 할 수 없으므로 인스턴스를 만드는 유일한 진입 점은 MoreExecutors 도우미 클래스입니다.

4.1. Guava를 Maven 종속성으로 추가

프로젝트에 Guava 라이브러리를 포함하려면 Maven pom 파일에 다음 종속성을 추가합니다. Maven Central 저장소 에서 최신 버전의 Guava 라이브러리를 찾을 수 있습니다 .

<dependency>
    <groupId>com.google.guava</groupId>
    <artifactId>guava</artifactId>
    <version>19.0</version>
</dependency>

4.2. 직접 실행자 및 직접 실행자 서비스

일부 조건에 따라 현재 스레드 또는 스레드 풀에서 작업을 실행하려는 경우가 있습니다. 단일 Executor 인터페이스 를 사용 하고 구현을 전환하는 것을 선호합니다 . 현재 스레드에서 작업을 실행하는 Executor 또는 ExecutorService 의 구현을 찾는 것은 그리 어렵지 않지만 여전히 몇 가지 상용구 코드를 작성해야합니다.

다행스럽게도 Guava는 사전 정의 된 인스턴스를 제공합니다.

다음 은 동일한 스레드에서 작업 실행을 보여주는 예 입니다. 제공된 작업이 500 밀리 초 동안 휴면 상태이지만 현재 스레드를 차단 하고 실행 호출이 완료된 직후 결과를 사용할 수 있습니다 .

Executor executor = MoreExecutors.directExecutor();

AtomicBoolean executed = new AtomicBoolean();

executor.execute(() -> {
    try {
        Thread.sleep(500);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    executed.set(true);
});

assertTrue(executed.get());

directExecutor () 메서드에 의해 반환 된 인스턴스 는 실제로 정적 싱글 톤 이므로이 메서드를 사용하면 객체 생성에 대한 오버 헤드가 전혀 제공되지 않습니다.

이 메서드는 MoreExecutors.newDirectExecutorService ()보다 선호해야합니다. 그 API는 모든 호출에 대해 본격적인 실행기 서비스 구현을 생성하기 때문입니다.

4.3. 실행자 서비스 종료

또 다른 일반적인 문제는 스레드 풀이 여전히 작업을 실행하는 동안 가상 머신 을 종료 하는 것입니다. 취소 메커니즘이 있어도 작업이 제대로 작동하고 실행기 서비스가 종료 될 때 작업이 중지된다는 보장은 없습니다. 이로 인해 태스크가 계속 작업을 수행하는 동안 JVM이 무기한 정지 될 수 있습니다.

이 문제를 해결하기 위해 Guava는 기존 실행기 서비스 제품군을 도입했습니다. 이들은 JVM과 함께 종료 되는 데몬 스레드를 기반으로 합니다 .

이러한 서비스는 또한 Runtime.getRuntime (). addShutdownHook () 메서드를 사용하여 종료 후크를 추가하고 중단 된 작업을 포기하기 전에 구성된 시간 동안 VM이 종료되는 것을 방지합니다.

다음 예제에서는 무한 루프가 포함 된 작업을 제출하지만 VM 종료시 작업을 대기하기 위해 구성된 시간이 100 밀리 초인 기존 실행기 서비스를 사용합니다. exitingExecutorService 가 없으면 이 작업으로 인해 VM이 무기한 중단됩니다.

ThreadPoolExecutor executor = 
  (ThreadPoolExecutor) Executors.newFixedThreadPool(5);
ExecutorService executorService = 
  MoreExecutors.getExitingExecutorService(executor, 
    100, TimeUnit.MILLISECONDS);

executorService.submit(() -> {
    while (true) {
    }
});

4.4. 듣는 데코레이터

수신 데코레이터를 사용하면 ExecutorService 를 래핑하고 간단한 Future 인스턴스 대신 작업 제출시 ListenableFuture 인스턴스를 수신 할 수 있습니다 . ListenableFuture의 인터페이스는 확장 미래를 단일 추가 방법이 따라 addListener . 이 메서드를 사용하면 나중에 완료 될 때 호출되는 리스너를 추가 할 수 있습니다.

ListenableFuture.addListener () 메서드를 직접 사용하는 경우는 거의 없지만 Futures 유틸리티 클래스 의 대부분의 도우미 메서드에 필수적 입니다. 예를 들어 Futures.allAsList () 메서드를 사용하면 결합 된 모든 미래가 성공적으로 완료되면 완료 되는 단일 ListenableFuture 에서 여러 ListenableFuture 인스턴스를 결합 할 수 있습니다 .

ExecutorService executorService = Executors.newCachedThreadPool();
ListeningExecutorService listeningExecutorService = 
  MoreExecutors.listeningDecorator(executorService);

ListenableFuture<String> future1 = 
  listeningExecutorService.submit(() -> "Hello");
ListenableFuture<String> future2 = 
  listeningExecutorService.submit(() -> "World");

String greeting = Futures.allAsList(future1, future2).get()
  .stream()
  .collect(Collectors.joining(" "));
assertEquals("Hello World", greeting);

5. 결론

이 기사에서는 표준 Java 라이브러리와 Google의 Guava 라이브러리에서 스레드 풀 패턴과 그 구현에 대해 설명했습니다.

참고

현재글자바 스레드 풀 소개(Threadpool)

기록만이살길