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

메인은 작업 완료나 실패를 수집하지 않고 수천 플랫폼 스레드를 동시에 만듭니다. 환경에 따라 자원 부족이나 긴 스케줄 지연이 발생할 수 있습니다. 가상 스레드는 생성 비용을 낮추지만 외부 서비스 동시성 상한과 종료 관리는 여전히 필요합니다.

실행기 설계의 세 자원

작업자는 동시에 실행할 수 있는 작업 수를 제한한다.
큐는 아직 작업자를 얻지 못한 제출을 보관한다.
거부 정책은 작업자와 큐가 모두 찼을 때 호출자 행동을 정한다.
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());
    }
}

완료 수는 6이고 서로 다른 작업자 이름은 최대 2개입니다. 큐 네 칸과 실행 두 칸이 정확히 여섯 제출을 받습니다. 용량을 넘는 추가 제출은 거부 정책으로 이동합니다.

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

실패가 작업자 교체에만 맡겨지지 않고 업무 이름과 지표로 남습니다. Callable과 Future를 쓰면 예외를 결과 수집 구분으로 전달할 수 있습니다.

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

작업 성격	실행 정책	이유
CPU 계산	코어 수 근처 고정 풀	과도한 경쟁 방지
제한 외부 I/O	의존 서비스 상한 반영	동시 요청 보호
짧은 독립 I/O 다수	가상 스레드 검토	대기 비용 감소
반드시 순차 실행	단일 스레드 Executor	순서 단순화

호출부가 ExecutorService의 구체 타입을 직접 알면 풀 변경과 업무 변경이 함께 일어납니다. 제출 경계에는 execute(Runnable) 정도의 작은 포트를 두고, 스레드 이름·큐 크기·예외 관찰·종료 책임은 애플리케이션 조립 지점에 둡니다. 그러면 단위 작업은 현재 스레드에서 직접 실행해 확인할 수 있고 운영 구성만 유한 풀로 교체할 수 있습니다.

소유권도 분명해야 합니다. 실행기를 만든 구성 요소만 종료하고, 메서드 인자로 받은 실행기를 임의로 닫지 않습니다. 공유 풀을 한 기능이 종료하면 다른 기능의 제출이 갑자기 거부되기 때문입니다. 반대로 짧은 프로그램에서 만든 풀을 닫지 않으면 비데몬 작업자 때문에 프로세스가 끝나지 않습니다. 생성자와 close의 위치가 일치하는지 코드 리뷰에서 확인하세요.

작업 이름에는 요청 ID와 업무 종류를 넣되 개인정보는 넣지 않습니다. 작업 시작·성공·실패·소요 시간을 같은 ID로 기록하면 큐 대기와 실제 실행 시간을 구분할 수 있습니다. 이 구분이 있어야 스레드 부족인지 작업 자체의 지연인지 판별할 수 있습니다.

연습 문제

작업자 3개, 큐 6개인 실행기에 아홉 작업을 넣고 모두 끝난 뒤 작업자별 처리 수를 출력하세요. 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);
    }
}

합계는 9이며 분배는 스케줄에 따라 달라집니다. 특정 작업자가 세 건씩 처리한다고 단정하지 않습니다.

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

Executor 도입의 핵심은 Thread 철자를 줄이는 것이 아니라 실행 정책을 한곳에 모으는 것입니다. 작업자, 큐, 거부, 종료를 함께 검토하고 제출 코드에는 업무 단위만 남깁니다.