정상 종료와 yield

백그라운드 작업자 종료는 요청, 대기 해제, 진행 중 작업 처리, 최종 회수의 네 단계로 봐야 합니다. volatile 플래그만 바꾸면 큐에서 무기한 기다리는 스레드를 깨우지 못하고, interrupt만 보내면 남은 항목 처리 정책이 없습니다. printer 사례로 두 신호를 하나의 생명 주기로 묶습니다.

running 플래그와 take 대기

아래 daemon 작업자는 빈 큐의 take()에서 WAITING입니다. main이 volatile 플래그를 false로 써도 대기가 풀리지 않아 실제로 wrong-alive-after-flag=true가 출력됩니다. 마지막 interrupt는 예제 프로세스가 스레드를 남기지 않기 위한 정리입니다.

lab/FlagOnlyShutdownBug.java

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

public final class FlagOnlyShutdownBug {
    public static void main(String[] args) throws InterruptedException {
        BlockingQueue<String> queue = new LinkedBlockingQueue<>();
        Control control = new Control();
        Thread worker =
                new Thread(
                        () -> {
                            try {
                                while (control.running) {
                                    System.out.println(queue.take());
                                }
                            } catch (InterruptedException e) {
                                Thread.currentThread().interrupt();
                            }
                        },
                        "flag-only-worker");
        worker.setDaemon(true);
        worker.start();
        while (worker.getState() != Thread.State.WAITING) {
            Thread.onSpinWait();
        }
        control.running = false;
        worker.join(20);
        System.out.println("wrong-alive-after-flag=" + worker.isAlive());
        worker.interrupt();
        worker.join();
    }

    private static final class Control {
        volatile boolean running = true;
    }
}

가시성은 해결됐지만 wake-up이 없습니다. 반대로 interrupt만 써도 현재 항목을 버릴지 완료할지 모호합니다. 종료 API가 큐 acceptance와 작업자 상태를 함께 바꿔야 합니다.

프린터 종료 방식

즉시 취소는 현재 출력을 중단하고 큐를 보존·폐기합니다. 비우기 shutdown은 새 요청을 거부하고 이미 받은 항목을 모두 처리한 뒤 끝납니다. 기한 shutdown은 일정 시간 비우기한 뒤 interrupt로 전환합니다. 요구가 다르므로 메서드 이름과 결과도 달라야 합니다.

다음 구현은 poison 명령을 큐에 넣어 비우기 shutdown을 수행합니다. 큐는 FIFO이므로 종료 표식 앞의 문서가 먼저 처리됩니다. stop 이후 submit을 거부하고 join으로 작업자 종료를 기다립니다.

src/DrainingPrinter.java

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

public final class DrainingPrinter {
    private static final String STOP = new String("STOP");
    private final BlockingQueue<String> queue = new LinkedBlockingQueue<>();
    private boolean accepting = true;
    private final Thread worker = new Thread(this::loop, "draining-printer");

    public void start() {
        worker.start();
    }

    public synchronized void submit(String document) {
        if (!accepting) throw new IllegalStateException("printer stopping");
        queue.add(document);
    }

    public void shutdown() throws InterruptedException {
        synchronized (this) {
            if (accepting) {
                accepting = false;
                queue.add(STOP);
            }
        }
        worker.join();
    }

    private void loop() {
        try {
            for (; ; ) {
                String job = queue.take();
                if (job == STOP) break;
                System.out.println("printed=" + job);
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            System.out.println("printer interrupted, pending=" + queue.size());
        }
    }

    public static void main(String[] args) throws InterruptedException {
        DrainingPrinter p = new DrainingPrinter();
        p.start();
        p.submit("java");
        p.submit("thread");
        p.shutdown();
        System.out.println("printer-stopped");
    }
}

marker는 외부 입력과 식별 정보가 겹치지 않도록 private 객체가 더 적합하지만 큐 타입을 간단히 보이려고 고유 String 인스턴스를 사용했습니다. 제품 구현에서는 sealed 명령이나 별도 closeable 큐 추상화로 표현합니다. 작업자가 예외로 먼저 종료될 때 shutdown join이 돌아오더라도 남은 항목 처리 실패를 결과로 보고해야 합니다.

즉시 종료의 catch 지점

즉시 취소에서는 interrupt()가 take·sleep을 깨우고 반복문이 반환합니다. 진행 중인 printer 드라이버가 중단 가능하지 않다면 취소가 지연될 수 있습니다. device API의 시간 제한과 초기화 capability를 확인해야 합니다.

src/ImmediatePrinterStop.java

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

public final class ImmediatePrinterStop {
    private final BlockingQueue<String> queue = new LinkedBlockingQueue<>();
    private final Thread worker = new Thread(this::run, "immediate-printer");

    void start() {
        worker.start();
    }

    void submit(String value) {
        queue.add(value);
    }

    int stopNow() throws InterruptedException {
        worker.interrupt();
        worker.join();
        return queue.size();
    }

    private void run() {
        try {
            while (true) {
                String value = queue.take();
                Thread.sleep(5);
                System.out.println("printed=" + value);
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }

    public static void main(String[] args) throws InterruptedException {
        ImmediatePrinterStop p = new ImmediatePrinterStop();
        p.start();
        p.submit("one");
        p.submit("two");
        int pending = p.stopNow();
        System.out.println("pending-after-stop=" + pending);
    }
}

대기 중 수는 interrupt 시점에 따라 달라질 수 있으므로 기능 성공 기준으로 고정하지 않습니다. 중요한 것은 stopNow 반환 시 작업자가 TERMINATED라는 점입니다. 미처리 항목은 호출자가 영속성 큐로 넘기도록 반환 타입을 확장할 수 있습니다.

Empty·sleep·yield의 차이

반복문 본문이 비어 있는 while(condition){}는 바쁜 대기 spin입니다. current 핵심을 계속 사용하며 조건 가시성까지 보장되지 않으면 끝나지 않을 수 있습니다. 매우 짧은 락-free 대기에서만 Thread.onSpinWait()와 함께 제한적으로 사용하고 일반 큐 대기는 블로킹 기본형을 씁니다.

Thread.sleep(duration)은 현재 스레드를 TIMED_WAITING으로 보내 최소 시간 전에는 실행되지 않게 합니다. 정확한 재개 시각과 다른 스레드 완료는 보장하지 않습니다. 중단 가능하므로 shutdown 통로가 됩니다.

Thread.yield()는 current 실행 가능 스레드가 스케줄러에 양보 의사를 알리는 힌트입니다. 무시될 수 있고 어떤 스레드가 다음인지 정하지 않습니다. 정확성, 공정성, producer-소비자 조정을 yield에 의존시키지 않습니다.

yield와 큐 프로토콜

작업을 여러 개 연속 처리한 뒤 yield()를 호출하면 다른 실행 가능 스레드가 기회를 얻을 수도 있습니다. 그러나 큐의 take()와 락이 이미 명확한 대기·소유권 규칙을 제공하며 yield()는 성능 실험 대상일 뿐입니다. 처리량이 떨어지면 제거할 수 있어야 합니다.

app/BatchedPrinterWorker.java

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

public final class BatchedPrinterWorker {
    private sealed interface Command permits Print, Stop {}

    private record Print(String value) implements Command {}

    private record Stop() implements Command {}

    public static void main(String[] args) throws InterruptedException {
        BlockingQueue<Command> queue = new LinkedBlockingQueue<>();
        Thread worker = new Thread(() -> consume(queue), "batch-printer");
        worker.start();
        queue.add(new Print("A"));
        queue.add(new Print("B"));
        queue.add(new Print("C"));
        queue.add(new Stop());
        worker.join();
        System.out.println("batch-complete");
    }

    private static void consume(BlockingQueue<Command> queue) {
        try {
            int batch = 0;
            for (; ; ) {
                Command command = queue.take();
                if (command instanceof Stop) return;
                Print print = (Print) command;
                System.out.println("printed=" + print.value());
                if (++batch % 2 == 0) Thread.yield();
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}

출력 A, B, C 순서는 FIFO 큐 때문에 고정되지 yield 때문이 아닙니다. Stop도 같은 channel을 타므로 앞선 출력이 비우기됩니다.

shutdown API의 검증 목록

새 submit이 어느 시점부터 거부되는지, 큐의 대기 항목을 비울지, 진행 중인 항목이 원자적인지, interrupt가 블로킹 호출을 깨우는지, 소유자가 join하는지 확인합니다. 프로세스 종료 훅에서 호출할 때는 훅 자체의 시간 제한과 재진입도 고려합니다.

daemon 지정은 shutdown 구현을 대체하지 않습니다. 사용자 스레드가 프로세스를 붙잡는 문제와 안전하게 상태를 닫는 문제를 따로 해결합니다. 종료 로그는 요청 시각과 실제 작업자 종료 시각을 구분합니다.

연습 문제

큐에 세 작업을 넣고 poison marker로 비우기를 요청하되 100ms 안에 작업자가 끝나지 않으면 interrupt하세요. 메서드 반환 시 작업자는 살아 있지 않아야 합니다.

정답과 두 단계 중단

먼저 정상 비우기 기회를 주고 시간 제한 join 뒤 생존하면 interrupt와 무기한 join으로 정리를 끝냅니다.

exercise/DeadlineDrainSolution.java

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

public final class DeadlineDrainSolution {
    private static final int STOP = -1;

    public static void main(String[] args) throws InterruptedException {
        BlockingQueue<Integer> queue = new LinkedBlockingQueue<>();
        Thread worker =
                new Thread(
                        () -> {
                            try {
                                for (; ; ) {
                                    int value = queue.take();
                                    if (value == STOP) return;
                                    Thread.sleep(10);
                                    System.out.println("done=" + value);
                                }
                            } catch (InterruptedException e) {
                                Thread.currentThread().interrupt();
                            }
                        },
                        "deadline-drain");
        worker.start();
        queue.add(1);
        queue.add(2);
        queue.add(3);
        queue.add(STOP);
        worker.join(100);
        if (worker.isAlive()) {
            worker.interrupt();
            worker.join();
        }
        System.out.println("stopped=" + !worker.isAlive() + ", remaining=" + queue.size());
    }
}

현재 작업량은 정상적으로 비워지지만 느린 환경에서는 interrupt 경로로 끝날 수 있습니다. 두 경우 모두 stopped=true가 규칙입니다.

graceful shutdown은 플래그 하나나 yield 힌트가 아닙니다. acceptance 차단, 큐 정책, interrupt 가능한 대기, 시간 제한 전환, 최종 join을 한 소유자가 책임질 때 종료 상태를 예측할 수 있습니다.