Executor 작업 경계

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

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

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

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

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

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

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

bad/ThreadPerRequest.java

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은 새 작업을 막고 이미 제출된 작업의 배출을 시작한다.
종료 대기는 애플리케이션 전체 마감 시간 안에서 제한되어야 한다.

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

src/BoundedExecutorDemo.java

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());
    }
}

처음 두 execute: 원문의 자원 배정: 새 작업자에게 첫 작업으로 직접 전달
실행·반환 경계: 대기열을 거치지 않고 실행 가능
두 작업자가 생긴 뒤의 제출: 원문의 자원 배정: 용량 4의 대기열에 등록
실행·반환 경계: 기존 작업자가 큐에서 꺼내 실행
포화 또는 종료 뒤 제출: 원문의 자원 배정: AbortPolicy가 접수를 거부
실행·반환 경계: 호출자에게 RejectedExecutionException

원문 main은 6개를 제출합니다. 작업자 2명과 큐 용량 4는 설정이며, 실제 동시 점유 수나 큐의 최대 점유를 측정한 값은 아닙니다. 추가 제출의 거부 경로는 이 main에서 실행하지 않습니다.

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

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

src/ObservedExecutor.java

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를 쓰면 예외를 결과 수집 구분으로 전달할 수 있습니다.

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

작업 성격	실행 정책	이유
CPU 계산	코어 수 근처 고정 풀	과도한 경쟁 방지
제한 외부 I/O	의존 서비스 상한 반영	동시 요청 보호
짧은 독립 I/O 다수	가상 스레드 검토	대기 비용 감소
작업을 하나씩 실행	단일 스레드 Executor	업무상 제출 순서는 별도 결정
짧은 비동기 단계 조합	CompletableFuture	단계 실행 정책과 실패 조합 확인

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

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

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

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

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

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

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

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

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

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

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

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

연습 문제

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

ConcurrentHashMap과 merge를 사용합니다.

정답과 해설

exercise/WorkerCountSolution.java

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);
    }
}

세 예제의 종료 대기와 조회 시점을 나눈다
원문 프로그램	awaitTermination 뒤 동작	조회 결과를 읽는 범위
BoundedExecutorDemo	false면 shutdownNow 후 바로 조회	completed는 변화 중 근삿값 · names 순회 중 쓰기가 남을 수 있음
ObservedExecutor	1초 대기의 boolean을 무시	failures는 0 또는 1일 수 있으며 나중에 catch가 기록할 수 있음
WorkerCountSolution	1초 대기의 boolean을 무시	total과 counts 문자열은 서로 다른 시점의 집계

BoundedExecutorDemo: awaitTermination 뒤 동작: false면 shutdownNow 후 바로 조회
조회 결과를 읽는 범위: completed는 변화 중 근삿값 · names 순회 중 쓰기가 남을 수 있음
ObservedExecutor: awaitTermination 뒤 동작: 1초 대기의 boolean을 무시
조회 결과를 읽는 범위: failures는 0 또는 1일 수 있으며 나중에 catch가 기록할 수 있음
WorkerCountSolution: awaitTermination 뒤 동작: 1초 대기의 boolean을 무시
조회 결과를 읽는 범위: total과 counts 문자열은 서로 다른 시점의 집계

shutdownNow도 작업 종료를 기다리지 않습니다. BoundedExecutorDemo에서 쓰기가 남은 synchronizedList를 별도 잠금 없이 stream으로 순회하면 예외를 포함한 비결정적 동작이 가능합니다.

한 번의 실행에서는 total=9와 count-worker-0의 1건, count-worker-1과 count-worker-2의 각 4건을 출력했습니다. 분배는 스케줄에 따라 달라집니다.

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

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

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

이 페이지의 목차