작업자 스레드 안전하게 종료하기

앞 글의 InterruptiblePrinter는 interrupt() 한 번으로 작업자를 멈췄습니다. 하지만 실제 프린터 서비스를 끄려면 답해야 할 질문이 더 있습니다. 이미 받아 둔 문서는 모두 출력하고 끝낼까요, 아니면 지금 바로 멈출까요? 종료가 시작된 뒤에 들어오는 문서는 어떻게 할까요? 종료를 요청한 쪽은 작업자가 정말 끝났다는 것을 어떻게 알 수 있을까요?

이 질문들을 따라가면 종료는 네 단계로 나뉩니다. 종료를 요청하고, 기다리고 있는 작업자를 깨우고, 남은 작업을 정한 방식대로 처리하고, 작업자가 끝났는지 join()으로 확인하는 것입니다. 이 글에서는 먼저 흔한 실수인 '플래그만 바꾸는 종료'가 왜 작업자를 깨우지 못하는지 확인합니다. 그다음 남은 문서를 모두 출력하고 끝내는 종료와 즉시 멈추는 종료를 만들어 비교하고, 마지막으로 sleep, yield, onSpinWait가 대기와 종료에서 각각 어떤 역할을 하는지 정리합니다.

running 플래그만 바꾸면 큐에서 기다리는 작업자가 깨어나지 않는다

작업자를 멈추는 가장 흔한 첫 시도는 running 같은 플래그를 두고 반복문 조건에서 확인하는 것입니다. 다른 스레드가 바꾼 값이 작업자에게 보이도록 volatile도 붙입니다. 다음 프로그램은 이 방식으로 빈 큐의 take()에서 기다리는 작업자를 멈추려고 합니다.

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

wrong-alive-after-flag=true

코드를 순서대로 보겠습니다. main은 작업자를 시작한 뒤, 작업자가 빈 큐의 take()에서 기다리는 WAITING 상태가 될 때까지 Thread.onSpinWait()를 부르는 반복문으로 기다립니다. 그다음 control.running을 false로 바꾸고 join(20)으로 최대 20ms 동안 작업자가 끝나기를 기다립니다. 출력은 그 20ms가 지난 뒤에도 작업자가 살아 있다는 뜻입니다. take()가 돌아오는 경우는 큐에 항목이 들어올 때와 interrupt를 받을 때뿐인데, 이 시점까지 프로그램은 둘 중 어느 것도 하지 않으므로 이 결과는 실행할 때마다 같습니다.

플래그가 작업자를 멈추지 못한 이유는 volatile이 하는 일과 작업자가 있는 곳이 어긋났기 때문입니다. volatile은 main이 쓴 false를 작업자가 다음에 읽을 때 반드시 보게 해 줍니다. 하지만 작업자는 지금 플래그를 읽고 있지 않습니다. 반복문 조건을 이미 통과해 take() 안에서 기다리는 중이고, take()가 돌아오지 않는 한 조건 검사로 돌아가지 못합니다. 값을 보이게 하는 일(가시성)과 기다리는 스레드를 깨우는 일은 서로 다른 문제입니다.

마지막 두 줄의 interrupt()와 join()은 예제 프로세스를 깔끔하게 끝내기 위한 정리입니다. interrupt를 받은 take()가 InterruptedException을 던지면 작업자는 catch에서 플래그를 다시 세우고 끝납니다. 작업자를 데몬 스레드로 만든 것은, 정리 두 줄이 없더라도 main이 끝나면 프로세스가 끝나게 하려는 실습용 안전장치입니다. 데몬 설정이 종료 절차를 대신하지 못한다는 점은 뒤에서 다시 짚습니다.

작업자를 깨우는 방법은 두 가지입니다. 큐에 무언가를 넣어 take()가 돌아오게 하거나, interrupt()로 대기를 예외로 끝내는 것입니다. 어느 쪽을 쓸지는 '남은 문서를 어떻게 할 것인가'에 따라 달라집니다.

종료 방식 세 가지

프린터를 끄는 방식은 남은 일을 어떻게 다루느냐에 따라 크게 세 가지로 나눌 수 있습니다.

어느 방식이 맞는지는 기능이 요구하는 바에 달려 있습니다. 사용자가 보낸 문서를 하나도 잃으면 안 되는 프린터라면 비우고 끝내야 하고, 프로그램을 빨리 꺼야 하는 상황이라면 즉시 멈추고 남은 문서는 저장해 두었다가 다음에 처리할 수 있습니다. 결과가 이렇게 다르므로, 한 메서드가 상황에 따라 다르게 동작하게 만들기보다 shutdown()과 stopNow()처럼 이름부터 다른 메서드로 나누는 편이 호출하는 쪽에서 헷갈리지 않습니다. 세 방식 모두 종료가 시작되면 새 문서는 받지 않는 것이 일반적입니다.

종료 표식을 넣어 큐를 비우고 끝내기

비우고 끝내기는 큐에 '여기서 끝'이라는 특별한 항목을 넣어서 구현할 수 있습니다. 이런 항목을 종료 표식, 또는 독이 든 알약이라는 뜻에서 poison pill이라고 부릅니다. 큐는 먼저 넣은 것을 먼저 꺼내는 FIFO 구조이므로, 표식보다 먼저 들어온 문서는 모두 표식보다 먼저 꺼내집니다. 작업자는 문서를 하나씩 출력하다가 표식을 꺼내는 순간 반복문을 빠져나오면 됩니다. 다음 DrainingPrinter는 이 방식으로 shutdown()을 구현하고, 문서 출력은 실제 장치 대신 콘솔 로그로 나타냅니다.

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

printed=java
printed=thread
printer-stopped

main은 문서 두 개를 넣고 곧바로 shutdown()을 호출합니다. shutdown()은 accepting을 false로 바꿔 새 문서를 막고, 큐 맨 뒤에 STOP을 넣은 다음 join()으로 작업자가 끝나기를 기다립니다. 작업자는 java와 thread를 차례로 출력한 뒤 STOP을 꺼내 끝나고, 그제야 join()이 돌아와 main이 printer-stopped를 출력합니다. 앞의 두 줄은 같은 작업자가 큐의 순서대로 출력하고 마지막 줄은 그 작업자가 끝난 뒤에 출력하므로, 세 줄의 순서는 실행할 때마다 같습니다.

종료가 시작된 뒤 submit()을 호출하면 printer stopping이라는 메시지와 함께 IllegalStateException이 납니다. 첫 번째 shutdown()이 끝난 뒤 다시 호출하면 accepting이 이미 false이므로 표식을 또 넣지 않고, 이미 끝난 작업자를 join()하므로 바로 돌아옵니다.

접수 확인과 표식 추가를 한 락 안에서 하는 이유

submit()은 synchronized 메서드이고, shutdown()의 앞부분은 synchronized (this) 블록입니다. 둘 다 같은 객체(this)의 락을 쓰므로 한 번에 하나씩만 실행됩니다. 이 락이 왜 필요한지는 락이 없을 때 생길 수 있는 순서를 따라가 보면 알 수 있습니다. submit()에서 synchronized를 빼고 accepting에 volatile만 붙였다고 해 봅시다. 큐에 java가 들어 있는 상태에서 한 스레드는 submit("late")를, 다른 스레드는 shutdown()을 호출하면 다음 순서가 가능합니다.

submit()을 호출한 쪽은 1단계에서 접수 중임을 확인했으므로 late가 받아들여졌다고 믿습니다. 하지만 late는 STOP 뒤에 들어갔고 작업자는 STOP에서 끝나므로, late는 영원히 출력되지 않습니다. 예외도 나지 않았으니 호출한 쪽은 이 사실을 알 방법이 없습니다. volatile을 붙여도 이 문제가 풀리지 않는 것은, 값이 잘 보이는 것과 별개로 '확인하고 넣는' 두 동작 사이에 다른 스레드가 끼어들 수 있기 때문입니다. 두 메서드를 같은 락으로 묶으면 submit()의 확인과 등록 사이에 shutdown()의 변경과 표식 추가가 끼어들 수 없으므로, 받아들인 문서는 모두 STOP보다 앞에 놓입니다.

더 깊이: accepting에 volatile이 없어도 되는 이유와 join을 락 밖에서 부르는 이유

accepting은 volatile이 아니지만 읽기와 쓰기가 모두 같은 락 안에서만 일어납니다. 자바 메모리 모델에서 한 스레드가 락을 풀기 전에 한 쓰기는, 그 뒤에 같은 락을 얻은 스레드에게 보입니다. 그래서 shutdown()이 쓴 false는 그다음에 submit()에 들어오는 스레드에게 반드시 보입니다. 이 규칙은 가시성과 동기화를 다루는 19장에서 자세히 살펴봅니다.

join()을 synchronized 블록 밖에서 부르는 데도 이유가 있습니다. 락을 쥔 채 join()하면 작업자가 남은 문서를 모두 출력하는 동안 submit()을 호출한 다른 스레드가 락을 얻지 못해 함께 멈춥니다. 곧바로 거부 예외를 받아야 할 호출자가 종료가 끝날 때까지 붙잡혀 있게 되는 것입니다. 작업자가 진행 상황을 기록하려고 같은 락을 써야 하는 구조였다면 더 나쁩니다. 작업자는 락을 기다리고 shutdown()은 작업자를 기다리는 교착 상태(deadlock)가 되어 둘 다 영원히 멈춥니다.

종료 표식이 문서와 헷갈리지 않게 하기

STOP을 "STOP" 리터럴이 아니라 new String("STOP")으로 만들고, equals가 아니라 ==로 비교한 데에는 이유가 있습니다. 사용자가 보낸 문서의 내용이 우연히 STOP일 수도 있기 때문입니다. 다음 프로그램은 앞의 DrainingPrinter.java와 같은 폴더(기본 패키지)에 두고, 두 파일을 함께 컴파일해 실행합니다.

public final class StopMarkerPayload {
    public static void main(String[] args) throws InterruptedException {
        DrainingPrinter printer = new DrainingPrinter();
        printer.start();
        printer.submit("STOP");
        printer.shutdown();
        System.out.println("payload-complete");
    }
}

printed=STOP
payload-complete

문서 "STOP"은 내용이 표식과 같지만 다른 객체이므로 job == STOP은 false이고, 작업자는 이 문서를 평범하게 출력합니다. 그리고 shutdown()이 넣은 진짜 표식을 꺼냈을 때 끝납니다. 비교를 job.equals(STOP)으로 바꾸면 출력은 payload-complete 한 줄만 남습니다. 작업자가 내용만 보고 사용자의 문서를 종료 신호로 착각해, 출력하지 않은 채 끝나 버리기 때문입니다.

표식을 new String("STOP") 대신 "STOP" 리터럴로 만들어도 같은 일이 일어납니다. 문자열 풀에서 본 것처럼 내용이 같은 리터럴은 클래스가 달라도 같은 객체를 가리키므로, StopMarkerPayload의 "STOP"과 표식이 ==로도 같아집니다. 실제로 실행하면 이때도 payload-complete만 출력됩니다. 고유한 객체를 표식으로 쓰면 이런 충돌은 피할 수 있지만, 문자열 하나에 '문서'와 '제어 신호'라는 두 가지 뜻을 싣는다는 점은 그대로입니다. 그래서 실무에서는 문서와 종료 명령을 서로 다른 타입으로 표현하는 편이 명확합니다. 뒤에 나오는 BatchedPrinterWorker가 그렇게 만든 예입니다.

작업자가 먼저 끝났거나 아직 시작하지 않았을 때

shutdown()의 join()이 돌아왔다고 해서 문서가 모두 출력됐다는 뜻은 아닙니다. 작업자가 출력 도중 예외로 먼저 끝나 버리면, 그 뒤의 문서는 큐에 남은 채로 join()이 바로 돌아옵니다. 실제로 첫 문서에서 예외를 던지게 바꿔 실행하면, 작업자 스레드의 예외 메시지가 출력된 뒤에도 shutdown()은 정상적으로 돌아오고 큐에는 남은 문서와 STOP이 그대로 있습니다. 그러니 제품 코드라면 작업자가 표식까지 정상적으로 도달했는지, 남은 문서가 있는지를 결과로 알려 주어야 합니다.

이 클래스는 start()를 한 번 호출한 뒤에 submit()과 shutdown()을 쓴다고 가정합니다. 아직 시작하지 않은 스레드에 join()을 호출하면 기다리지 않고 바로 돌아오므로, start() 전에 shutdown()을 부르면 문서를 하나도 출력하지 않았는데도 shutdown()이 끝나 버립니다. 또 한 번 시작한 Thread는 끝난 뒤에도 다시 start()할 수 없고, 호출하면 IllegalThreadStateException이 납니다. 호출 순서를 코드로 강제하려면 시작 여부와 접수 여부를 함께 관리해야 합니다.

interrupt로 즉시 멈추기

이번에는 남은 문서를 기다리지 않고 바로 멈추는 방식입니다. 다음 ImmediatePrinterStop의 stopNow()는 작업자에게 interrupt()를 걸고 join()한 뒤 큐에 남은 문서 수를 돌려줍니다. 작업자는 출력에 시간이 걸리는 것을 흉내 내려고 문서를 꺼낼 때마다 5ms 동안 sleep합니다. take()와 sleep() 둘 다 interrupt에 반응하므로, 작업자는 어디서 기다리고 있든 바로 깨어납니다.

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

이 프로그램은 실행할 때마다 결과가 달라질 수 있습니다. 여러 번 실행하면 대부분 다음과 같이 나오지만, 가끔 pending-after-stop=1이 나오기도 합니다.

pending-after-stop=2

가장 흔한 결과가 2인 것은 대개 main이 더 빠르기 때문입니다. start()가 돌아온 뒤 새 스레드가 실제로 실행을 시작하기까지는 시간이 조금 걸리는데, 그사이 main은 문서 두 개를 넣고 interrupt()까지 마칩니다. 작업자가 뒤늦게 첫 take()를 호출할 때는 플래그가 이미 세워져 있으므로, 앞 글에서 본 대로 take()는 문서가 있어도 꺼내지 않고 바로 예외를 던집니다. 그래서 아무것도 출력되지 않고 두 문서가 모두 큐에 남습니다.

interrupt가 늦게 도착할수록 결과가 바뀝니다. stopNow() 앞에 Thread.sleep을 넣고 시간을 몇 ms씩 늘려 가며 실행해 보면 다음 다섯 경우를 모두 볼 수 있습니다.

눈여겨볼 줄은 사라진 문서가 있는 두 줄입니다. take()로 꺼낸 문서는 더 이상 큐에 없고, sleep 도중에 중단되면 출력되지도 않습니다. 그래서 pending-after-stop이 0이라고 모든 문서가 출력된 것은 아니고, 1이라고 출력되지 못한 문서가 하나뿐인 것도 아닙니다. 남은 큐의 크기는 '아직 꺼내지 않은 문서 수'일 뿐입니다. 중단된 문서를 다시 처리하려면 큐에 있는 문서, 꺼내서 처리 중인 문서, 완료한 문서를 따로 기록해야 합니다. 남은 문서를 영속 큐로 옮길 수 있도록 stopNow()가 개수 대신 문서 목록을 돌려주게 바꾸는 것도 방법입니다.

아래 그림은 비우고 끝내기와 즉시 멈추기를 나란히 놓고, 새 문서 접수와 남은 문서 처리, 마지막 회수가 어떻게 다른지 단계별로 비교합니다.

이 구현에도 앞 글의 CancellableDocumentConsumer와 같은 빈틈이 있습니다. submit()에 접수 상태 확인이 없어서 stopNow() 뒤에도 문서를 받지만, 이미 끝난 작업자는 그 문서를 처리하지 않습니다. 또한 즉시 멈추기는 작업자가 기다리는 호출이 interrupt에 반응할 때만 '즉시'입니다. 실제 프린터 드라이버 호출이 interrupt에 반응하지 않는다면 그 호출이 끝날 때까지 취소가 늦어지므로, 장치 API가 시간 제한이나 취소 기능을 제공하는지 확인해야 합니다.

세 번째 방식인 기한을 두고 비우기는 앞의 두 방식을 이어 붙이면 됩니다. 종료 표식을 넣어 비우기를 시작하고, join(timeout)으로 정해진 시간만큼 기다린 뒤에도 작업자가 살아 있으면 interrupt()로 즉시 멈추기로 넘어갑니다. 이 방식은 연습 문제에서 직접 만들어 봅니다.

sleep, yield, onSpinWait, 빈 반복문의 차이

종료를 기다리거나 다른 스레드에게 차례를 넘기는 코드를 쓰다 보면, 조건이 맞을 때까지 반복문을 돌리거나 그 안에서 sleep이나 yield를 부르고 싶어집니다. 네 가지는 비슷해 보이지만 하는 일이 다르고, 어느 것도 다른 스레드가 바꾼 값을 보이게 해 주지는 않습니다.

본문이 빈 반복문 while (!done) {}를 바쁜 대기(busy waiting)라고 합니다. 조건이 바뀔 때까지 CPU를 계속 쓰면서 같은 값을 읽습니다. 게다가 done이 volatile이 아니고 동기화도 없다면, 다른 스레드가 값을 바꿔도 이 스레드가 끝내 보지 못해 반복이 끝나지 않을 수 있습니다. 반복문 안에서 Thread.onSpinWait()를 부르면 '지금 바쁜 대기 중'이라는 힌트를 JVM에 주어, JVM이 그 반복을 CPU에 더 알맞게 실행할 수 있게 합니다. 하지만 이 메서드도 가시성을 만들어 주지 않으므로 조건 변수에는 여전히 volatile이나 동기화가 필요합니다. 이 장의 예제가 onSpinWait()를 쓴 곳은 실습 순서를 맞추기 위한 아주 짧은 대기뿐이고, 다른 스레드의 일을 기다리는 일반적인 대기는 take() 같은 블로킹 메서드에 맡겨야 합니다.

Thread.sleep(millis)는 현재 스레드를 정해진 시간 동안 쉬게 합니다. 실제로 쉬는 시간은 시스템 타이머와 스케줄러의 정밀도에 따라 조금씩 달라지고, 시간이 지난 뒤 곧바로 실행된다는 보장도 없습니다. 쉬는 동안 interrupt를 받으면 일찍 깨어나 InterruptedException을 던지므로, 종료 요청이 작업자에게 닿는 통로가 됩니다. 주의할 점은 synchronized 블록 안에서 sleep해도 쥐고 있던 락을 놓지 않는다는 것입니다. 락을 쥔 채 잠든 스레드가 있으면 같은 락에 들어가려는 다른 스레드는 그동안 BLOCKED 상태로 기다립니다. 그러니 다른 스레드가 그 락을 얻어 조건을 바꿔 줘야 하는 상황이라면, sleep을 넣는 것으로는 문제가 풀리지 않습니다.

Thread.yield()는 '지금 CPU를 다른 스레드에게 양보해도 좋다'는 뜻을 스케줄러에 전하는 힌트입니다. 스케줄러는 이 힌트를 무시해도 되고, 받아들이더라도 어느 스레드가 다음에 실행될지는 정하지 않습니다. 그래서 프로그램이 올바르게 동작하는지, 스레드들이 공평하게 실행되는지, 생산자와 소비자가 차례를 맞추는지를 yield에 맡겨서는 안 됩니다. 자바 API 문서도 이 메서드를 적절하게 쓸 일은 드물고, 경쟁 조건 때문에 생기는 버그를 재현하는 디버깅이나 테스트에 쓸모가 있을 수 있다고 설명합니다.

마지막으로 sleep과 yield는 메모리 가시성에 아무 영향도 주지 않습니다. JLS 17.3절은 두 메서드에 동기화 의미가 전혀 없다고 명시합니다. 컴파일러는 sleep이나 yield를 호출하기 전에 레지스터에 담아 둔 값을 메모리에 쓸 필요가 없고, 호출한 뒤에 다시 읽을 필요도 없습니다. 반복문에 sleep을 넣었더니 다른 스레드의 변경이 보이기 시작했다면 그것은 우연일 뿐 보장된 동작이 아닙니다. 네 가지와 블로킹 메서드를 비교하면 다음과 같습니다.

yield를 넣어도 순서는 큐가 정한다

문서를 몇 개 처리할 때마다 yield()를 호출하면 다른 스레드가 CPU를 쓸 기회를 얻을 수도 있습니다. 하지만 기다림과 차례는 이미 큐의 take()가 분명하게 정하고 있으므로, 이런 yield()는 성능을 실험해 보는 정도의 의미밖에 없습니다. 다음 BatchedPrinterWorker는 문서 두 개를 출력할 때마다 yield()를 호출합니다. 앞에서 말한 대로 문서와 종료 명령을 서로 다른 타입으로 표현한 예이기도 합니다.

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

printed=A
printed=B
printed=C
batch-complete

Command는 큐에 들어갈 명령의 공통 타입입니다. sealed interface Command permits Print, Stop은 이 인터페이스를 구현할 수 있는 타입을 Print와 Stop 둘로 제한하는데, sealed는 sealed 계층을 다루는 36장에서 자세히 배웁니다. 문서는 Print 레코드에 담고 종료 명령은 Stop 레코드로 보내므로, 내용이 STOP인 문서가 들어와도 종료 명령과 헷갈릴 일이 없습니다. 작업자는 꺼낸 명령이 Stop이면 끝나고, 아니면 Print로 바꿔 출력합니다.

출력이 A, B, C 순서인 것은 yield() 덕분이 아니라, 큐가 넣은 순서대로 꺼내 주고 꺼내는 작업자가 하나뿐이기 때문입니다. 실제로 yield() 호출을 지우고 실행해도 출력은 똑같습니다. Stop도 같은 큐로 전달되므로 앞에 넣은 문서 세 개를 모두 출력한 뒤에 꺼내지고, main은 join()으로 작업자가 끝나기를 기다린 뒤 batch-complete를 출력합니다. 이 yield()가 처리량을 떨어뜨린다면 언제든 지울 수 있어야 하고, 지웠더니 동작이 바뀐다면 그 코드는 처음부터 잘못 만든 것입니다.

종료 기능을 만들 때 확인할 것

지금까지 만든 예제를 바탕으로, 작업자 스레드를 두는 클래스에 종료 기능을 만들 때 확인할 항목을 정리하면 다음과 같습니다.

• 언제부터 새 작업을 거부하는가? 접수 확인과 등록이 종료 요청과 엇갈리지 않는가?
• 큐에 남은 작업은 모두 처리하는가, 처리하지 않고 남겨서 돌려주는가?
• 처리 중인 작업은 끝까지 하는가, 중간에 멈춰도 되는가? 중간에 멈췄다면 그 작업은 어디에 기록되는가?
• 작업자가 기다리는 모든 호출이 종료 표식이나 interrupt로 깨어나는가?
• 종료를 요청한 쪽이 join()으로 작업자가 끝났는지 확인하는가? 작업자가 예외로 먼저 끝난 경우도 결과로 알 수 있는가?

이 종료 기능을 JVM 종료 훅(shutdown hook)에서 호출할 계획이라면 두 가지를 더 생각해야 합니다. 먼저 훅은 빨리 끝나야 합니다. 사용자가 로그아웃하거나 운영체제가 꺼질 때는 프로세스가 정리할 시간이 정해져 있을 수 있으므로, 남은 작업을 시간 제한 없이 비우는 종료보다 기한을 둔 종료가 어울립니다. 또 훅 안에서 System.exit()를 다시 호출하면 그 호출은 돌아오지 않고 영원히 멈춥니다. JVM이 이미 종료 과정에 들어가 있기 때문입니다.

작업자를 데몬 스레드로 만드는 것은 종료 기능을 대신하지 못합니다. JVM은 사용자 스레드가 모두 끝나면 데몬 스레드를 기다리지 않고 종료하므로, 하던 작업이 중간에 끊기고 finally 블록도 실행되지 않을 수 있습니다. 데몬 스레드는 '이 스레드 때문에 프로세스가 끝나지 않는 문제'를 막아 줄 뿐이고, 상태를 안전하게 닫는 일은 이 글에서 만든 종료 절차가 해야 합니다. 종료 기록을 남길 때도 종료를 요청한 시각과 작업자가 실제로 끝난 시각을 따로 남겨야, 종료에 얼마나 걸렸고 어디서 지체됐는지 알 수 있습니다.

정리

• volatile 플래그는 값을 보이게 할 뿐 take()처럼 기다리는 스레드를 깨우지 못합니다. 기다리는 작업자는 종료 표식을 넣거나 interrupt()를 걸어 깨웁니다.
• 비우고 끝내기는 접수를 막고 종료 표식을 큐 맨 뒤에 넣어, 그 앞의 문서를 모두 처리한 뒤 끝냅니다. 접수 확인과 등록, 접수 중단과 표식 추가를 같은 락 안에서 해야 표식 뒤로 문서가 끼어들지 않습니다.
• 종료 표식은 문서와 헷갈리지 않아야 합니다. 고유한 객체를 ==로 비교하거나, 더 낫게는 문서와 명령을 서로 다른 타입으로 표현합니다.
• 즉시 멈추기에서 남은 큐의 크기는 미완료 문서 수가 아닙니다. 꺼낸 뒤 중단된 문서는 큐에도 출력에도 남지 않습니다.
• sleep과 yield에는 동기화 의미가 없고, yield는 무시될 수 있는 힌트입니다. 대기와 순서는 큐나 락 같은 동기화 도구로 정합니다.
• 데몬 설정은 종료 절차를 대신하지 않습니다. 종료를 요청한 쪽이 join()으로 작업자가 끝났는지 확인합니다.

연습 문제

1. 큐에 작업 세 개(1, 2, 3)를 넣고 종료 표식으로 비우기를 요청하되, 100ms 안에 작업자가 끝나지 않으면 interrupt()로 멈추세요. 작업자는 작업 하나마다 10ms 동안 sleep한 뒤 done=번호를 출력합니다. main이 마지막 줄을 출력할 때 작업자는 살아 있지 않아야 하고, 그때 큐에 남은 항목 수도 함께 출력합니다.
2. 1번 풀이에서 작업자가 1을 꺼내 sleep하던 중에 interrupt를 받아 끝났다고 합시다. 큐에는 2, 3, 종료 표식이 남아 있습니다. 큐에 남은 항목 수와 출력하지 못한 작업 수를 각각 세고, 두 숫자가 같더라도 같은 것을 센 것이 아닌 이유를 설명하세요.
3. 아직 start()하지 않은 DrainingPrinter에 "A"를 submit()하고 shutdown()을 호출했습니다. shutdown()이 돌아온 것을 보고 A가 출력됐다고 판단할 수 있을까요? 그 뒤에 start()를 호출하면 어떻게 되며, 같은 객체에 start()를 한 번 더 호출하면 어떻게 될까요?

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

done=1
done=2
done=3
stopped=true, remaining=0

다음 글

19장의 첫 글에서는 이 장에서 여러 번 등장한 volatile과 synchronized가 정확히 무엇을 보장하는지, 여러 스레드가 같은 값을 고칠 때 왜 값이 사라지는지를 실행으로 확인합니다.