Executor 작업 경계

업무 코드는 “이 작업을 실행한다”만 말하고 몇 개의 스레드를 언제 만들지는 실행 정책이 결정해야 합니다.

요청마다 new Thread를 호출하면 생성 비용, 동시 실행 상한, 종료, 예외 관리를 모든 호출 지점이 떠안습니다.

Executor는 작업 제출과 실행 자원 관리를 분리합니다.

스레드 풀은 작업자를 재사용하고 큐로 순간 폭주를 흡수합니다.

그러나 무제한 큐나 무제한 스레드는 문제를 미룰 뿐입니다.

풀 크기, 큐 용량, 거부 정책을 한 세트로 설계합니다.

이 장의 완전한 클래스는 코드 표제의 파일명과 같은 .java 파일에 각각 저장해 실행합니다. Thread.ofPlatform을 쓰는 예제는 Java 21 이상이 필요합니다. 먼저 Runnable 람다, synchronizedList·ConcurrentHashMap의 역할, 앞 장의 인터럽트와 완료 대기를 익혀 두세요. 아래 5,000개 스레드 반례는 생성 비용을 설명하는 코드이므로 자원이 작은 환경에서 무작정 실행하기보다 먼저 읽고, 풀을 쓰는 BoundedExecutorDemo부터 확인할 수 있습니다. 생성 수를 줄인 실험은 원 5,000개 조건을 실행한 결과와 구별합니다.

요청별 스레드 생성과 완료 대기 누락

import java.util.ArrayList;
import java.util.List;

public final class ThreadPerRequest {
    public static void main(String[] args) {
        List<Thread> threads = new ArrayList<>();
        for (int request = 0; request < 5_000; request++) {
            int id = request;
            threads.add(Thread.ofPlatform().start(() -> {
                try { Thread.sleep(100); }
                catch (InterruptedException e) { Thread.currentThread().interrupt(); }
                if (id == -1) System.out.println(id);
            }));
        }
        System.out.println("created=" + threads.size());
    }
}

메인은 플랫폼 스레드 시작을 5,000회 시도하며, 모두 시작하면 created=5000을 출력합니다. 생성 도중 앞선 작업이 끝날 수 있으므로 이 값이 동시에 살아 있는 스레드 수는 아닙니다. 명시적 join은 없지만 시작한 비데몬 작업자가 남으면 JVM은 main 반환 뒤에도 종료되지 않습니다.

환경에 따라 스레드 시작이 자원 부족으로 실패하거나 스케줄이 지연될 수 있습니다. 시작 루프가 실패하면 created 출력까지 도달하지 못할 수 있습니다.

가상 스레드는 생성 비용을 낮추지만 외부 서비스 동시성 상한과 종료 관리는 여전히 필요합니다.

실행기 설계의 세 자원

• 작업자는 동시에 실행할 수 있는 작업 수를 제한한다.
• 큐는 아직 작업자를 얻지 못한 제출을 보관한다.
• 거부 정책은 포화되거나 종료한 실행기에 제출했을 때 호출자 행동을 정한다.
• ThreadFactory는 이름과 미처리 예외 정책을 일관되게 부여한다.
• shutdown은 새 작업을 막고 이미 제출된 작업의 배출을 시작한다.
• 종료 대기는 애플리케이션 전체 마감 시간 안에서 제한되어야 한다.

유한 실행기로 작업 재사용 확인

큐 용량은 대기 중인 작업 수입니다. 아래 설정에서 작업자 두 명이 모두 실행 중이고 네 작업이 큐에 남아 있는 동안에는 일곱 번째 제출을 받아들일 공간이 없습니다. AbortPolicy는 호출자에게 RejectedExecutionException을 던집니다. execute가 빈자리가 생길 때까지 기다려 주는 것은 아닙니다. 작업이 빨리 끝나면 공간을 다시 쓸 수 있으므로, 연속으로 제출한 총개수만 보고 거부 여부를 정할 수는 없습니다.

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public final class BoundedExecutorDemo {
    public static void main(String[] args) throws InterruptedException {
        ThreadPoolExecutor executor = new ThreadPoolExecutor(
                2, 2, 0, TimeUnit.MILLISECONDS,
                new ArrayBlockingQueue<>(4),
                Thread.ofPlatform().name("report-worker-", 0).factory(),
                new ThreadPoolExecutor.AbortPolicy());
        List<String> names = java.util.Collections.synchronizedList(new ArrayList<>());
        for (int task = 0; task < 6; task++) {
            executor.execute(() -> names.add(Thread.currentThread().getName()));
        }
        executor.shutdown();
        if (!executor.awaitTermination(2, TimeUnit.SECONDS)) executor.shutdownNow();
        System.out.println("completed=" + executor.getCompletedTaskCount());
        System.out.println("workerNames=" + names.stream().distinct().sorted().toList());
    }
}

한 번의 실행에서는 completed=6과 workerNames=[report-worker-0, report-worker-1]을 출력했습니다. awaitTermination의 반환값은 출력하지 않으므로 이 결과만으로 true를 반환했다고 단정하지 않습니다.

처음 두 제출은 핵심 작업자를 만들고 그 작업자에게 바로 배정됩니다. 다음 제출은 큐에 들어갈 수 있지만, 앞 작업자가 이미 꺼내 처리했다면 큐의 점유는 줄어 있습니다. 따라서 “작업자 2 + 큐 용량 4 = 여섯 자리”는 아직 작업이 끝나지 않은 포화 상태의 수용량 설명이며, 이번 실행에서 항상 네 작업이 대기했다는 관측은 아닙니다. ThreadPoolExecutor의 공식 실행 규칙은 핵심 미만의 작업자 생성, 큐 적재, 최댓값 확장 순서를 구분합니다.

Runnable 예외 관찰을 위한 작업 래퍼

작업 실패와 제출 실패는 발생 위치가 다릅니다. 아래 newFixedThreadPool(2)는 대기 큐가 무한인 관찰용 예제이므로 유입량을 제한하는 운영 구성으로 그대로 쓰지 않습니다. 래퍼는 실제 실행된 작업의 RuntimeException만 집계하며, Error를 잡지 않습니다. 이미 종료한 실행기에 제출하다가 호출자 쪽에서 발생한 거부는 래퍼 안으로 들어가지 않으므로 failures에 추가되지 않습니다.

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

public final class ObservedExecutor {
    private final ExecutorService delegate = Executors.newFixedThreadPool(2);
    private final AtomicInteger failures = new AtomicInteger();

    void execute(String name, Runnable task) {
        delegate.execute(() -> {
            try { task.run(); }
            catch (RuntimeException e) {
                failures.incrementAndGet();
                System.err.println("task=" + name + " failed=" + e.getClass().getSimpleName());
            }
        });
    }

    public static void main(String[] args) throws InterruptedException {
        ObservedExecutor service = new ObservedExecutor();
        service.execute("ok", () -> {});
        service.execute("broken", () -> {
            throw new IllegalStateException("boom");
        });
        service.delegate.shutdown();
        service.delegate.awaitTermination(1, java.util.concurrent.TimeUnit.SECONDS);
        System.out.println("failures=" + service.failures.get());
    }
}

이 래퍼는 RuntimeException을 잡아 업무 이름을 stderr에 남기고 failures를 증가시킵니다. Error는 잡지 않습니다. 한 번의 실행에서는 failures=1과 task=broken failed=IllegalStateException을 기록했으며, 두 출력 스트림 사이의 순서는 측정하지 않았습니다.

Callable과 Future를 쓰면 예외를 결과 수집 경계로 전달할 수 있습니다.

작업 수명과 실행기 소유권 결정

종료 요청 뒤에도 작업이 남을 수 있습니다. shutdown은 새 접수를 막고 기존 작업의 완료를 기다릴 수 있게 합니다. awaitTermination의 false는 제한 시간에 종료를 확인하지 못했다는 뜻입니다. 그 뒤 shutdownNow를 호출해도 실행 중인 작업의 종료까지 기다리지는 않으며, 반환 목록은 큐에서 꺼낸 미시작 작업입니다. 위 예제의 시간 초과 경로에서 작업자가 아직 names를 갱신한다면 synchronizedList 자체가 stream 전체 순회를 보호하지 않습니다. 목록을 외부에서 synchronized (names)로 잠그고 복사해야 안전한 시점별 목록을 얻으며, 최종 결과로 삼으려면 작업 종료까지 확인해야 합니다.

실행 범위를 업무 코드에서 분리하는 법

호출부가 ExecutorService의 구체 타입을 직접 알면 풀 변경과 업무 변경이 함께 일어납니다.

제출 경계에는 execute(Runnable) 정도의 작은 포트를 두고, 스레드 이름·큐 크기·예외 관찰·종료 책임은 애플리케이션 조립 지점에 둡니다.

그러면 단위 작업은 현재 스레드에서 직접 실행해 확인할 수 있고 운영 구성만 유한 풀로 교체할 수 있습니다.

소유권도 분명해야 합니다.

실행기를 만든 구성 요소만 종료하고, 메서드 인자로 받은 실행기를 임의로 닫지 않습니다.

공유 풀을 한 기능이 종료하면 다른 기능의 제출이 갑자기 거부되기 때문입니다.

반대로 짧은 프로그램에서 만든 풀을 닫지 않으면 비데몬 작업자 때문에 프로세스가 끝나지 않습니다.

생성자와 close의 위치가 일치하는지 코드 리뷰에서 확인하세요.

작업 이름에는 요청 ID와 업무 종류를 넣되 개인정보는 넣지 않습니다.

제출·시작·성공·실패 시각을 같은 ID로 기록하면 큐 대기와 실제 실행 시간을 구분할 수 있습니다.

이 구분이 있어야 스레드 부족인지 작업 자체의 지연인지 판별할 수 있습니다.

연습 문제

작업자 3개, 큐 용량 6개인 실행기에 아홉 작업을 넣고 종료를 요청하세요. 최대 1초의 종료 대기 뒤 그 시점의 작업자별 처리 수를 출력하고, 종료가 확인되지 않은 집계를 최종값으로 단정하지 마세요.

ConcurrentHashMap과 merge를 사용합니다.

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public final class WorkerCountSolution {
    public static void main(String[] args) throws InterruptedException {
        var counts = new ConcurrentHashMap<String, Integer>();
        var executor = new ThreadPoolExecutor(3, 3, 0, TimeUnit.MILLISECONDS,
                new ArrayBlockingQueue<>(6),
                Thread.ofPlatform().name("count-worker-", 0).factory());
        for (int i = 0; i < 9; i++) {
            executor.execute(() -> counts.merge(Thread.currentThread().getName(), 1, Integer::sum));
        }
        executor.shutdown();
        executor.awaitTermination(1, TimeUnit.SECONDS);
        System.out.println("total=" + counts.values().stream().mapToInt(Integer::intValue).sum());
        System.out.println(counts);
    }
}

제출과 실행 분리를 완료하는 조건

Executor 도입의 핵심은 Thread 철자를 줄이는 것이 아니라 실행 정책을 한곳에 모으는 것입니다.

작업자, 큐, 거부, 종료를 함께 검토하고 제출 코드에는 업무 단위만 남깁니다.

접수·실행·종료 경계를 구별하는 추가 연습

1. 작업자 수와 최대 작업자 수가 모두 2이고 큐 용량이 4이며 AbortPolicy를 쓰는 실행기가 있다. 첫 두 작업은 모두 시작했고, 이후 제출하는 작업은 큐에서 대기한다. 이 동안 작업이 하나도 끝나지 않고 취소나 큐 제거도 없다. 일곱 번 execute를 호출하면 접수·실행 중·대기·거부 작업 수는 각각 얼마인가? 일곱 번째 호출은 공간이 빌 때까지 기다리는가?
2. 1번에서 접수한 여섯 작업이 그대로 남아 있다. 실행 중인 두 작업은 인터럽트를 받아도 별도 종료 신호 전까지 계속 실행한다고 하자. shutdown 뒤 제한 시간까지 종료 신호가 오지 않아 awaitTermination이 false를 반환했다. 이어 shutdownNow를 호출한 직후 반환 목록의 크기는 얼마이며, 실행기 전체 종료와 완료 작업 여섯 개를 보장하는가?
3. 위 ObservedExecutor에 정상 작업 하나와 IllegalStateException을 던지는 작업 하나를 제출했다. 둘 다 끝나고 실행기의 종료를 확인한 뒤 failures를 읽는다. 이후 같은 실행기에 정상 작업을 한 번 더 제출하면 어떤 예외가 어느 스레드에서 발생하며, 실패 계수는 늘어나는가?