notify와 잘못된 깨움

객체 모니터에는 조건별 대기실이 여러 개 생기지 않습니다.

같은 객체에서 wait()를 호출한 생산자와 소비자는 하나의 대기 집합에 섞입니다.

큐에 값이 생겨 소비자를 깨워야 하는 순간 notify()를 호출해도 JVM은 소비자를 골라 준다고 약속하지 않습니다.

이미 가득 찬 큐에서 기다리는 생산자를 깨우면 그 스레드는 잠금을 다시 얻고 조건을 확인한 뒤 아무 일도 못 한 채 되돌아갑니다.

잘못된 역할이 한 번 깨어나는 것은 성능 낭비로 끝날 수 있습니다.

그러나 이어질 상태 변경자가 더 없고 필요한 역할이 계속 잠들어 있다면 전체 작업이 멈춥니다.

notifyAll()은 현재 대기자 모두에게 재검사 기회를 주어, 단일 알림이 필요한 역할을 선택하지 않는 문제를 피합니다.

대신 여러 스레드가 같은 잠금을 두고 경쟁하고 다시 대기할 수 있어 불필요한 각성 비용이 생길 수 있습니다.

신호 한 번에 역할 선택을 기대하는 취약한 코드

다음 큐는 생산과 소비 양쪽에서 notify()만 사용합니다.

코드만 보면 상태가 바뀔 때마다 한 명을 깨우므로 효율적으로 보입니다.

하지만 대기 집합에서 어떤 역할이 선택되는지는 구현 의존입니다.

실행기는 각 작업자에 차례로 join(300)을 호출한 뒤 살아 있는 작업자에게 interrupt를 보냅니다. 최종 join에는 제한이 없으며, 모두 종료한 뒤 성공 반환 개수를 출력합니다. 전체 작업의 300ms 기한은 아닙니다.

bad/SingleNotifyMixedWaiters.java

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

public final class SingleNotifyMixedWaiters {
    private final Queue<Integer> queue = new ArrayDeque<>();
    private final int capacity = 1;

    synchronized void put(int value) throws InterruptedException {
        while (queue.size() == capacity) wait();
        queue.add(value);
        notify();
    }

    synchronized int take() throws InterruptedException {
        while (queue.isEmpty()) wait();
        int value = queue.remove();
        notify();
        return value;
    }

    public static void main(String[] args) throws InterruptedException {
        SingleNotifyMixedWaiters buffer = new SingleNotifyMixedWaiters();
        boolean[] completed = new boolean[8];
        List<Thread> workers = new ArrayList<>();
        for (int i = 0; i < 4; i++) {
            int value = i;
            int producerSlot = i * 2;
            int consumerSlot = producerSlot + 1;
            workers.add(Thread.ofPlatform().start(() -> {
                try { buffer.put(value); completed[producerSlot] = true; }
                catch (InterruptedException e) { Thread.currentThread().interrupt(); }
            }));
            workers.add(Thread.ofPlatform().start(() -> {
                try { buffer.take(); completed[consumerSlot] = true; }
                catch (InterruptedException e) { Thread.currentThread().interrupt(); }
            }));
        }
        for (Thread worker : workers) {
            worker.join(300);
        }
        for (Thread worker : workers) {
            if (worker.isAlive()) worker.interrupt();
        }
        for (Thread worker : workers) worker.join();
        int completedCount = 0;
        for (boolean done : completed) if (done) completedCount++;
        System.out.println("completed=" + completedCount + "/" + workers.size());
    }
}

이번 실행은 completed=8/8이므로 작업 정지를 재현하지 않았습니다. 이 결과가 notify의 역할 선택을 보장하지는 않습니다.

완료 수가 8보다 작은 실행을 관측하더라도 그것만으로 어떤 역할이 깨어났는지는 알 수 없습니다. 이 코드는 알림 대상과 정지 원인을 직접 계측하지 않습니다.

notifyAll과 조건 재검사

전체 알림을 받으면 생산자와 소비자가 모두 대기 집합에서 나옵니다.

현재 큐가 비었다면 생산자가 잠금을 얻어 진행할 수 있고, 소비자는 자신의 while 조건을 확인한 뒤 다시 대기합니다.

큐가 가득 찬 상황에서는 반대 역할이 움직입니다.

올바른 역할을 직접 지정하지 못하는 단일 대기 집합에서 안전성을 확보하는 방식입니다.

알림 대상 선택을 JVM의 우연한 대기 순서에 맡겨야 한다면 notifyAll이 보수적인 기본값입니다.
모든 복귀 경로는 while 술어로 보호되어 잘못 깨어난 역할이 공유 상태를 훼손하지 않게 합니다.
상태 변경 없이 알림만 보내면 깨어난 참여자가 전부 같은 거짓 조건을 보고 다시 잠들므로 효과가 없습니다.
각성 횟수, 재대기 횟수, 잠금 경쟁 시간은 처리량이 커질 때 별도 성능 지표가 됩니다.
대기 역할이 많아 비용이 커지면 알림을 다시 notify로 줄이기보다 조건 대기실을 분리합니다.
표준 동시성 큐가 요구를 충족한다면 직접 신호 정책을 유지보수하지 않는 편이 검증 비용이 낮습니다.

notifyAll을 정확성 장치로 채택한 뒤 실제 측정에서 병목이 확인되면 ReentrantLock의 여러 Condition으로 이동할 수 있습니다.

생산자는 소비자 조건만, 소비자는 생산자 조건만 신호하도록 역할을 지정합니다.

최적화 순서는 “잘못된 역할도 안전하게 재대기”를 먼저 확보하고 “불필요한 역할은 깨우지 않기”를 나중에 적용하는 방식이 좋습니다.

전체 알림으로 생존성을 우선한 유한 큐

아래 구현은 단일 모니터의 한계를 숨기지 않습니다.

상태 전이마다 모든 대기자를 깨우고, 깨어난 작업자는 자신의 조건을 다시 검사합니다.

통계 필드는 임계 영역 안에서 갱신하여 대기 진입과 복귀 수를 관찰합니다.

src/NotifyAllBoundedQueue.java

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

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

    public NotifyAllBoundedQueue(int capacity) {
        if (capacity < 1) throw new IllegalArgumentException("capacity");
        this.capacity = capacity;
    }

    public synchronized void put(E value) throws InterruptedException {
        if (value == null) throw new NullPointerException("value");
        while (queue.size() == capacity) {
            waits++;
            wait();
            wakeups++;
        }
        queue.add(value);
        notifyAll();
    }

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

    public synchronized int waitCount() { return waits; }
    public synchronized int wakeupCount() { return wakeups; }

    public static void main(String[] args) throws InterruptedException {
        NotifyAllBoundedQueue<Integer> buffer = new NotifyAllBoundedQueue<>(2);
        Thread[] consumers = new Thread[3];
        for (int i = 0; i < consumers.length; i++) {
            consumers[i] = Thread.ofPlatform().name("consumer-" + i).start(() -> {
                try { System.out.println("value=" + buffer.take()); }
                catch (InterruptedException e) { Thread.currentThread().interrupt(); }
            });
        }
        Thread.sleep(30);
        for (int value : new int[]{11, 22, 33}) {
            buffer.put(value);
        }
        for (Thread consumer : consumers) {
            consumer.join();
        }
        System.out.printf("waits=%d wakeups=%d%n", buffer.waitCount(), buffer.wakeupCount());
    }
}

같은 큐 상태라도 생산자와 소비자의 대기 조건은 다르다
다시 검사한 큐 크기	put의 조건과 동작	take의 조건과 동작
`size = 0`	가득 차지 않음 · 추가 가능	비어 있음 · wait
`0 < size < capacity`	가득 차지 않음 · 추가 가능	비어 있지 않음 · 제거 가능
`size = capacity`	가득 참 · wait	비어 있지 않음 · 제거 가능

size = 0: put의 조건과 동작: 가득 차지 않음 · 추가 가능
take의 조건과 동작: 비어 있음 · wait
0 < size < capacity: put의 조건과 동작: 가득 차지 않음 · 추가 가능
take의 조건과 동작: 비어 있지 않음 · 제거 가능
size = capacity: put의 조건과 동작: 가득 참 · wait
take의 조건과 동작: 비어 있지 않음 · 제거 가능

깨어난 작업자는 모니터를 다시 얻은 뒤 이 조건을 검사합니다. 여러 작업자가 순서대로 진행할 수도 있으며, notifyAll은 획득 순서나 공정성을 보장하지 않습니다.

값 11, 22, 33은 각각 한 번 출력되지만 소비자 이름과 값의 조합은 정해지지 않습니다.

생산자와 소비자의 정상 wait 복귀를 함께 세므로 이 값을 소비 건수나 실패한 재경쟁 횟수와 같게 해석하지 않습니다. 대기 중 인터럽트로 예외가 나면 해당 호출의 wakeups 증가는 건너뜁니다.

교환 총량과 대기 통계를 확인하는 부하 탐침

다음 코드는 여러 생산자와 소비자가 정해진 총량을 교환한 뒤 합계와 대기 통계를 검사합니다.

독약 값이나 강제 종료 대신 역할별 반복 횟수를 같게 맞춰 모든 스레드가 자연스럽게 끝나게 합니다.

src/NotifyAllLoadProbe.java

import java.util.ArrayList;
import java.util.List;

public final class NotifyAllLoadProbe {
    public static void main(String[] args) throws InterruptedException {
        NotifyAllBoundedQueue<Integer> queue = new NotifyAllBoundedQueue<>(4);
        List<Thread> threads = new ArrayList<>();
        long[] partial = new long[4];

        for (int producerId = 0; producerId < 4; producerId++) {
            int start = producerId * 50;
            threads.add(Thread.ofPlatform().start(() -> {
                try {
                    for (int value = start; value < start + 50; value++) {
                        queue.put(value);
                    }
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }));
        }
        for (int consumerId = 0; consumerId < 4; consumerId++) {
            int slot = consumerId;
            threads.add(Thread.ofPlatform().start(() -> {
                try {
                    long local = 0;
                    for (int count = 0; count < 50; count++) {
                        local += queue.take();
                    }
                    partial[slot] = local;
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }));
        }
        for (Thread thread : threads) {
            thread.join();
        }
        long sum = 0;
        for (long value : partial) sum += value;
        System.out.println("sum=" + sum);
        System.out.println("extraWakeups=" + Math.max(0, queue.wakeupCount() - 200));
    }
}

wait 복귀 통계와 불필요한 각성 계측은 구별한다
출력 지표	원문의 증가 또는 계산	이번 실행
`waits`	put과 take에서 wait 호출 직전 증가	큐 예제: `5`
`wakeups`	같은 wait가 정상 반환한 직후 증가	큐 예제: `5`
`sum`	부하 탐침에서 소비한 0~199의 합	부하 탐침: `19900`
`extraWakeups`	부하 탐침의 `max(0, wakeupCount - 200)`	부하 탐침: `0`

waits: 원문의 증가 또는 계산: put과 take에서 wait 호출 직전 증가
이번 실행: 큐 예제: 5
wakeups: 원문의 증가 또는 계산: 같은 wait가 정상 반환한 직후 증가
이번 실행: 큐 예제: 5
sum: 원문의 증가 또는 계산: 부하 탐침에서 소비한 0~199의 합
이번 실행: 부하 탐침: 19900
extraWakeups: 원문의 증가 또는 계산: 부하 탐침의 max(0, wakeupCount - 200)
이번 실행: 부하 탐침: 0

두 대기 카운터에는 생산자와 소비자가 모두 포함됩니다. 각성 원인·실패한 재경쟁·OS 깨움·경과 시간은 계측하지 않으므로, extraWakeups=0도 불필요한 각성이 없었다는 뜻은 아닙니다.

이번 실행의 합계와 종료는 이 입력에서의 교환 결과를 확인합니다. 합계 하나로 모든 값이 정확히 한 번씩 전달됐거나 모든 스케줄이 완료함을 증명하지는 않습니다.

최적화를 비교하려면 같은 부하에서 재검사 후 다시 대기한 횟수와 처리 시간을 별도로 계측해야 합니다. 이 main은 두 Condition 구현과의 비교를 실행하지 않습니다.

작은 큐에서 수치 차이가 업무 지연으로 이어지지 않는다면 단순한 표준 구현을 유지하는 선택도 합리적입니다.

단일 알림과 전체 알림의 선택 기준

대기 구조	안전한 신호	이유
한 알림으로 한 대기자만 진행하면 됨	`notify` 검토 가능	어느 대기자를 골라도 같은 술어
생산·소비가 같은 집합 공유	`notifyAll` 우선	현재 대기자 모두에게 재검사 기회
역할별 Condition 분리	반대 조건 신호	대상 대기실을 지정 가능
라이브러리 큐 사용	내부 구현에 위임	검증된 정책과 취소 지원

대기자 수가 하나라는 가정은 코드 주석으로 끝내지 말고 생성 경로에서 강제해야 합니다.

유지보수자가 두 번째 작업자를 추가하는 순간 notify의 전제가 깨질 수 있습니다.

구조적으로 단일 대기임을 보장할 수 없다면 전체 알림이나 조건 분리를 선택하는 편이 안전합니다.

연습 문제

여러 작업자가 현재 설정 버전을 기억하고 다음 버전이 게시될 때까지 기다립니다.

게시자는 버전을 1씩 올리고 설정 문자열을 교체합니다.

한 번의 게시가 현재 대기 중인 모든 구독자에게 보여야 하므로 notify가 아니라 notifyAll을 사용하세요.

작업자는 자신이 본 버전보다 큰지 반복 검사해야 합니다.

정답과 해설

구독자가 기다리는 술어는 publishedVersion <= observedVersion입니다.

새 설정과 버전은 같은 모니터 안에서 바꾼 뒤 전체 알림을 보냅니다.

반환값에 버전을 포함해야 작업자가 다음 호출에서 정확한 기준을 전달할 수 있습니다.

exercise/VersionedConfigSignalSolution.java

import java.util.ArrayList;
import java.util.List;

public final class VersionedConfigSignalSolution {
    record Snapshot(long version, String value) {}

    private long version;
    private String value = "initial";

    synchronized Snapshot awaitNewerThan(long observed) throws InterruptedException {
        while (version <= observed) wait();
        return new Snapshot(version, value);
    }

    synchronized void publish(String next) {
        if (next == null || next.isBlank()) throw new IllegalArgumentException("next");
        value = next;
        version++;
        notifyAll();
    }

    public static void main(String[] args) throws InterruptedException {
        VersionedConfigSignalSolution source = new VersionedConfigSignalSolution();
        List<Snapshot> received = java.util.Collections.synchronizedList(new ArrayList<>());
        Thread a = subscriber(source, received, "a");
        Thread b = subscriber(source, received, "b");
        while (a.getState() != Thread.State.WAITING || b.getState() != Thread.State.WAITING) {
            Thread.onSpinWait();
        }
        source.publish("dark-mode-on");
        a.join();
        b.join();
        System.out.println("subscribers=" + received.size());
    }

    private static Thread subscriber(VersionedConfigSignalSolution source,
                                     List<Snapshot> received,
                                     String name) {
        return Thread.ofPlatform().name(name).start(() -> {
            try { received.add(source.awaitNewerThan(0)); }
            catch (InterruptedException e) { Thread.currentThread().interrupt(); }
        });
    }
}

이번 출력은 subscribers=2입니다. 게시가 한 번뿐이므로 두 반환값은 코드상 버전 1과 dark-mode-on이지만 그 필드는 출력하지 않습니다. 대기 상태를 읽는 반복과 join에는 자체 제한 시간이 없습니다.

이 패턴은 이벤트를 모두 보존하는 큐가 아닙니다.

게시가 연속으로 두 번 일어나기 전에 구독자가 실행되지 않으면 최신 버전만 관찰할 수 있으므로 변경 이력 전달이 필요할 때는 별도 로그 구조를 사용해야 합니다.

생존성·효율 판단 단계

단일 대기 집합에서 notifyAll은 현재 대기자 모두를 깨우지만, 모니터를 다시 얻는 순서나 공정성을 보장하지 않습니다.

불필요한 각성이 있다는 이유로 이를 notify로 바꾸면 측정 가능한 비용을 명세되지 않은 운에 교환하는 셈입니다.

먼저 신호 대상 선택 때문에 진행 가능한 역할만 계속 잠드는 구조를 피하고, 이후 계측으로 각성 비용이 실제 병목인지 확인합니다.

역할 선택이 필요해졌다면 다음 단계는 신호 한 번을 영리하게 만드는 것이 아니라 대기 공간 자체를 나누는 것입니다.

두 Condition은 생산 가능과 소비 가능을 타입 대신 구조로 분리하여 신호 대상을 명시합니다.