조건 재검사

대기에서 깨어났다는 사건과 원하는 조건이 참이라는 사실은 같지 않습니다. 알림은 “공유 상태가 바뀌었을 수 있으니 다시 보라”는 힌트입니다. 여러 소비자가 동시에 깨어나면 가장 먼저 잠금을 얻은 하나가 원소를 가져가고, 뒤늦게 진입한 소비자는 다시 빈 큐를 만납니다. JVM은 명시적 알림이 없어도 대기에서 복귀할 가능성을 허용하므로 조건 술어를 재검사하는 코드는 선택 사항이 아닙니다.

if (empty) wait()는 대기 전에는 조건을 한 번 확인하지만 복귀 후에는 바로 다음 줄로 진행합니다. 반면 while (empty) wait()는 잠금을 다시 얻은 시점의 최신 상태를 검사합니다. 반복문은 바쁜 대기가 아닙니다. 조건이 거짓일 때 wait가 모니터를 내놓고 잠들므로 CPU를 계속 사용하지 않습니다.

단일 항목을 둘러싼 소비자 경쟁

다음 저장소는 소비 조건에 if를 씁니다. 두 작업자가 빈 큐에서 대기한 뒤 항목 하나를 넣고 notifyAll을 호출합니다. 첫 작업자가 값을 제거하고 끝난 다음 두 번째 작업자도 복귀하여 remove를 실행하므로 NoSuchElementException에 도달할 수 있습니다. 스케줄에 따라 결과가 바뀌므로 여러 번 실행하거나 장벽으로 대기 시점을 맞춰 관찰합니다.

intentional-runtime/IfGuardedTake.java

import java.util.ArrayDeque;
import java.util.Queue;
import java.util.concurrent.CountDownLatch;

public final class IfGuardedTake {
    private final Queue<String> values = new ArrayDeque<>();
    private final CountDownLatch entered = new CountDownLatch(2);

    synchronized String take() throws InterruptedException {
        if (values.isEmpty()) {
            entered.countDown();
            wait();
        }
        return values.remove();
    }

    synchronized void put(String value) {
        values.add(value);
        notifyAll();
    }

    public static void main(String[] args) throws InterruptedException {
        IfGuardedTake queue = new IfGuardedTake();
        Runnable consumer = () -> {
            try {
                System.out.println(Thread.currentThread().getName() + "=" + queue.take());
            } catch (Exception e) {
                System.out.println(Thread.currentThread().getName() + " failed=" + e.getClass().getSimpleName());
            }
        };
        Thread c1 = Thread.ofPlatform().name("reader-a").start(consumer);
        Thread c2 = Thread.ofPlatform().name("reader-b").start(consumer);
        queue.entered.await();
        queue.put("only-one");
        c1.join();
        c2.join(200);
        c2.interrupt();
    }
}

실패는 두 번째 소비자가 예외를 내거나, 실행 타이밍에 따라 한 스레드가 끝나지 않는 형태로 보입니다. 핵심은 특정 스레드 이름이 아닙니다. 한 번 참이 된 “항목 존재” 조건을 다른 스레드가 먼저 소비할 수 있다는 점입니다. 조건은 개인에게 예약된 허가증이 아니라 공유 상태에 대한 순간적인 관찰입니다.

조건 술어를 코드의 중심에 배치

버퍼 메서드를 읽을 때 먼저 canPut = size < capacity, canTake = !isEmpty 두 술어를 적습니다. 대기 반복문의 조건은 각각 !canPut, !canTake입니다. 상태 변경이 끝난 뒤 반대 역할이 만족할 수 있는 술어가 무엇인지 확인하고 알립니다. 이렇게 표현하면 잠금 도구가 synchronized에서 Condition으로 바뀌어도 논리는 유지됩니다.

복귀 원인은 알림, 인터럽트, 시간 만료, 허용된 임의 복귀 중 하나일 수 있으므로 상태 검사가 최종 판단자가 됩니다.
while의 본문에는 상태 변경을 넣지 않고 오직 대기만 두어 술어와 전이를 분리합니다.
큐 조작은 조건 검사를 수행한 동일한 잠금 구간에 있어야 다른 참여자가 사이에 상태를 바꾸지 못합니다.
값을 제거한 소비자는 빈자리를 만들었고, 값을 추가한 생산자는 소비 가능 항목을 만들었다는 사건을 알립니다.
대기에서 인터럽트되면 원소를 추가하거나 제거하지 않은 채 메서드를 빠져나가 상태 불변식을 보존합니다.
테스트는 두 소비자·한 값처럼 경쟁을 의도적으로 만들어 재검사 경로가 실제로 실행되는지 확인합니다.

조건을 메서드 이름과 맞추는 것도 중요합니다. take는 값이 생길 때까지 기다리는 동작이고 poll은 현재 없으면 즉시 빈 결과를 주는 동작입니다. 같은 구현에서 두 규칙을 뒤섞으면 호출자가 지연 특성을 예측할 수 없습니다. 메서드 이름, 예외 선언, 시간 제한을 통해 대기 성격을 드러냅니다.

경쟁 후에도 안전한 다중 소비 버퍼

완성 버전은 여러 생산자와 소비자를 받아들입니다. put과 take의 대기 조건을 반복 검사하고 상태 전이 뒤 전체 알림을 보냅니다. ArrayDeque는 null을 허용하지 않으므로 빈 상태 표현이 값과 겹치지 않습니다. 스냅샷 메서드는 확인용으로만 두고 실제 업무 흐름은 대기 메서드를 사용합니다.

src/WhileGuardedBuffer.java

import java.util.ArrayDeque;
import java.util.List;
import java.util.Objects;
import java.util.Queue;

public final class WhileGuardedBuffer<E> {
    private final Queue<E> queue = new ArrayDeque<>();
    private final int capacity;

    public WhileGuardedBuffer(int capacity) {
        if (capacity <= 0) throw new IllegalArgumentException("capacity=" + capacity);
        this.capacity = capacity;
    }

    public synchronized void put(E value) throws InterruptedException {
        Objects.requireNonNull(value, "value");
        while (queue.size() >= capacity) {
            wait();
        }
        queue.add(value);
        notifyAll();
    }

    public synchronized E take() throws InterruptedException {
        while (queue.isEmpty()) {
            wait();
        }
        E value = queue.remove();
        notifyAll();
        return value;
    }

    public synchronized List<E> snapshot() {
        return List.copyOf(queue);
    }

    public static void main(String[] args) throws InterruptedException {
        WhileGuardedBuffer<String> buffer = new WhileGuardedBuffer<>(1);
        Thread first = consumer("first", buffer);
        Thread second = consumer("second", buffer);
        Thread.sleep(30);
        buffer.put("alpha");
        buffer.put("beta");
        first.join();
        second.join();
        System.out.println("remaining=" + buffer.snapshot().size());
    }

    private static Thread consumer(String name, WhileGuardedBuffer<String> buffer) {
        return Thread.ofPlatform().name(name).start(() -> {
            try {
                System.out.println(name + " consumed=" + buffer.take());
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });
    }
}

어느 소비자가 alpha를 얻는지는 보장하지 않습니다. 대신 두 값이 한 번씩만 출력되고 잔여 크기가 0이며 모든 작업이 종료되어야 합니다. 동시성 테스트의 단언은 스케줄 순서보다 보존 성질을 겨냥해야 합니다. 특정 이름의 스레드가 첫 값을 받는다고 가정하면 정상적인 스케줄 변화에도 테스트가 흔들립니다.

술어를 이름 붙여 복잡한 상태를 다루기

현실의 작업 큐에는 일시 중지, 종료 요청, 우선순위 같은 조건이 더 붙습니다. 긴 논리식을 반복문에 직접 넣으면 어떤 상태에서 깨어나야 하는지 읽기 어렵습니다. 아래 예제는 canTake와 isDrained를 분리합니다. 닫힌 큐에 남은 항목은 계속 처리하고, 닫힌 뒤 완전히 비면 전용 결과로 종료합니다.

src/CloseableWorkQueue.java

import java.util.ArrayDeque;
import java.util.Objects;
import java.util.Optional;
import java.util.Queue;

public final class CloseableWorkQueue<E> {
    private final Queue<E> jobs = new ArrayDeque<>();
    private boolean closed;

    public synchronized void submit(E job) {
        Objects.requireNonNull(job, "job");
        if (closed) throw new IllegalStateException("queue closed");
        jobs.add(job);
        notifyAll();
    }

    public synchronized Optional<E> takeUntilClosed() throws InterruptedException {
        while (!canTake() && !isDrained()) {
            wait();
        }
        if (isDrained()) return Optional.empty();
        return Optional.of(jobs.remove());
    }

    public synchronized void close() {
        closed = true;
        notifyAll();
    }

    private boolean canTake() {
        return !jobs.isEmpty();
    }

    private boolean isDrained() {
        return closed && jobs.isEmpty();
    }

    public static void main(String[] args) throws InterruptedException {
        CloseableWorkQueue<Integer> queue = new CloseableWorkQueue<>();
        queue.submit(10);
        queue.submit(20);
        queue.close();
        int sum = 0;
        Optional<Integer> next;
        while ((next = queue.takeUntilClosed()).isPresent()) sum += next.orElseThrow();
        System.out.println("sum=" + sum);
    }
}

결과는 sum=30입니다. 종료 플래그만 보고 즉시 빠져나가면 이미 접수된 작업이 남습니다. 반대로 빈 상태만 기다리면 더 이상 생산자가 없는 닫힌 큐에서 영원히 잠듭니다. 두 술어를 구별하면 “남은 작업 배출 후 종료”라는 제품 규칙이 코드에 직접 나타납니다.

조건 대기의 `if`·`while` 선택 기준

검토 항목	위험 신호	수정 방향
깨어난 뒤 조건을 다시 읽는가	`wait` 다음 줄이 `remove`	`while` 술어로 감싸기
다른 스레드가 자원을 선점할 수 있는가	대기자가 둘 이상	재경쟁을 전제로 설계
종료 상태가 별도로 있는가	빈 상태와 end 혼용	종료 술어 분리
테스트가 순서를 단정하는가	스레드 이름별 값 고정	총량과 종료를 검증

if가 허용되는 드문 경우를 찾는 것보다 조건 대기의 기본 형식으로 while을 고정하는 편이 안전합니다. 대기자가 하나라고 지금 확신해도 유지보수 중 역할 수가 늘 수 있습니다. 표준 라이브러리의 문서와 관례도 반복 검사를 전제로 하므로 팀 코드 리뷰에서 빠르게 확인할 수 있습니다.

연습 문제

작업 큐에 pause, resume, close 기능을 추가하세요. 소비자는 일시 중지 중에는 항목이 있어도 가져가면 안 됩니다. 닫을 때는 남은 작업을 모두 처리한 뒤 빈 결과로 종료하되, 닫힌 상태에서 일시 중지가 계속되어 종료를 막아서는 안 됩니다. 조건 술어를 메서드로 이름 붙이세요.

정답과 해설

소비 가능 조건은 “항목이 있고, 일시 중지가 아니거나 이미 닫힘”입니다. 배출 완료는 “닫혔고 비어 있음”입니다. 닫기는 더 이상의 제출을 금지하고, 남은 작업이 있다면 pause를 무시해 종료가 진행되도록 합니다. 모든 상태 변경 뒤 대기자를 깨워 새 술어를 평가하게 합니다.

exercise/PausableBatchQueueSolution.java

import java.util.ArrayDeque;
import java.util.Optional;
import java.util.Queue;

public final class PausableBatchQueueSolution<E> {
    private final Queue<E> queue = new ArrayDeque<>();
    private boolean paused;
    private boolean closed;

    synchronized void submit(E value) {
        if (closed) throw new IllegalStateException("closed");
        if (value == null) throw new NullPointerException("value");
        queue.add(value);
        notifyAll();
    }

    synchronized void pause() {
        if (!closed) paused = true;
    }

    synchronized void resume() {
        paused = false;
        notifyAll();
    }

    synchronized void close() {
        closed = true;
        paused = false;
        notifyAll();
    }

    synchronized Optional<E> take() throws InterruptedException {
        while (!mayConsume() && !drained()) wait();
        if (drained()) return Optional.empty();
        return Optional.of(queue.remove());
    }

    private boolean mayConsume() {
        return !queue.isEmpty() && !paused;
    }

    private boolean drained() {
        return closed && queue.isEmpty();
    }

    public static void main(String[] args) throws InterruptedException {
        PausableBatchQueueSolution<String> jobs = new PausableBatchQueueSolution<>();
        jobs.pause();
        jobs.submit("index");
        Thread worker = Thread.ofPlatform().start(() -> {
            try {
                System.out.println(jobs.take().orElseThrow());
                System.out.println("ended=" + jobs.take().isEmpty());
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });
        jobs.close();
        worker.join();
    }
}

close가 pause를 해제하므로 작업자는 index를 처리하고 다음 호출에서 종료를 확인합니다. 제품 요구가 “일시 중지 상태에서는 닫아도 처리하지 않고 반환”이라면 술어가 달라져야 합니다. 핵심은 숨은 우선순위를 만들지 않고 상태 전이 표와 코드가 같은 정책을 말하게 하는 것입니다.

알림 이후의 책임 정리

깨어난 작업자는 행동할 권리를 받은 것이 아니라 조건을 다시 검사할 기회를 얻었습니다. 잠금을 획득하고 현재 술어가 참임을 확인한 뒤에만 큐를 바꿉니다. 이 원칙이 있으면 임의 복귀와 다른 작업자의 선점을 같은 코드 경로로 안전하게 처리할 수 있습니다.

조건이 늘어나면 긴 표현을 반복하지 말고 의미 있는 이름으로 분리합니다. 대기 반복문을 읽는 사람이 “무엇을 기다리고 언제 끝나는지” 바로 알 수 있어야 합니다. 반복 검사는 동시성 기법인 동시에 상태 모델을 문서화하는 장치입니다.