notify와 잘못된 깨움

앞 글의 버퍼는 큐를 바꿀 때마다 notifyAll()로 기다리는 스레드를 모두 깨웠습니다. 그런데 값 하나가 들어왔을 때 그 값을 가져갈 수 있는 소비자는 한 명뿐입니다. 한 명이면 충분한데 모두를 깨우는 것은 낭비처럼 보이고, 마침 기다리는 스레드 하나만 깨우는 notify()도 있습니다. 하지만 생산자와 소비자가 섞여 기다리는 버퍼에서 notify()만 쓰면, 조건 검사까지 제대로 한 코드가 값 몇 개를 주고받다가 그대로 멈춥니다.

이 글에서는 먼저 그 정지를 실행해 보고, 스레드들이 어떤 순서로 모두 잠드는지 한 단계씩 따라갑니다. 그다음 notifyAll()이 이 문제를 어떻게 피하는지, notify()를 써도 되는 경우는 언제인지, 그리고 notifyAll()이 그 대가로 무엇을 치르는지 살펴봅니다.

한 객체에는 대기 집합이 하나뿐입니다

wait()를 호출한 스레드는 그 객체의 대기 집합(wait set)에 들어가 알림을 기다립니다. 대기 집합은 객체마다 하나뿐이어서, 기다리는 이유가 달라도 같은 객체에서 wait()를 부른 스레드는 모두 한곳에 모입니다. 유한 버퍼에서 생산자는 빈칸이 생기기를, 소비자는 값이 들어오기를 기다립니다. 기다리는 조건은 다르지만 둘 다 this.wait()를 부르므로 같은 대기 집합에 섞여 있습니다.

notify()는 이 대기 집합에서 스레드 하나를 골라 깨웁니다. 누구를 고르는지에 대해 Object.notify의 API 문서는 "선택은 임의적이며 구현의 재량에 달려 있다(The choice is arbitrary and occurs at the discretion of the implementation)"고만 적고 있습니다. 즉 notify()는 깨울 스레드가 어떤 조건을 기다리는지 보지 않습니다. 소비자를 깨워야 할 순간에 생산자를 깨울 수도 있고, 그 반대일 수도 있습니다. 반면 notifyAll()은 대기 집합에 있는 스레드를 모두 깨웁니다.

notify만 쓰는 버퍼가 멈추는 모습

다음 버퍼는 용량이 1이고, 값을 넣은 뒤와 꺼낸 뒤에 notify()만 호출합니다. 조건은 앞 글에서 배운 대로 while로 다시 검사하므로, 잘못 깨어난 스레드가 빈 큐에서 값을 꺼내는 일은 없습니다. 생산자 둘은 1,200개씩 값을 넣고 소비자 둘은 1,200개씩 꺼내므로, 모두 끝나면 2,400개를 주고받게 됩니다. main은 스레드마다 최대 0.5초씩 기다린 뒤, 그래도 살아 있는 스레드가 있으면 이름과 상태를 출력하고 interrupt()로 깨워 프로그램을 끝냅니다. 마지막 줄에는 실제로 꺼낸 값의 수를 출력합니다.

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

public final class SingleNotifyMixedWaiters {
    private static final int PRODUCERS = 2;
    private static final int CONSUMERS = 2;
    private static final int TOTAL = 2_400; // 주고받을 값의 총수

    private final Queue<Integer> queue = new ArrayDeque<>();
    private final int capacity = 1;
    private int taken; // 지금까지 꺼낸 값의 수

    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();
        taken++;
        notify(); // 기다리는 스레드 하나만 깨운다
        return value;
    }

    synchronized int takenCount() {
        return taken;
    }

    public static void main(String[] args) throws InterruptedException {
        SingleNotifyMixedWaiters buffer = new SingleNotifyMixedWaiters();
        List<Thread> workers = new ArrayList<>();
        for (int id = 1; id <= PRODUCERS; id++) {
            workers.add(Thread.ofPlatform().name("producer-" + id).start(() -> {
                try {
                    for (int i = 0; i < TOTAL / PRODUCERS; i++) buffer.put(i);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }));
        }
        for (int id = 1; id <= CONSUMERS; id++) {
            workers.add(Thread.ofPlatform().name("consumer-" + id).start(() -> {
                try {
                    for (int i = 0; i < TOTAL / CONSUMERS; i++) buffer.take();
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }));
        }

        for (Thread worker : workers) {
            worker.join(500); // 스레드마다 최대 0.5초씩 기다린다
        }
        for (Thread worker : workers) {
            if (worker.isAlive()) {
                System.out.println(worker.getName() + " " + worker.getState());
                worker.interrupt(); // wait에서 빠져나오게 해 프로그램을 끝낸다
            }
        }
        for (Thread worker : workers) {
            worker.join();
        }
        System.out.println("taken=" + buffer.takenCount() + "/" + TOTAL);
    }
}

한 번 실행한 결과는 다음과 같습니다. 0.5초를 기다린 뒤에도 살아 있는 스레드만 이름과 상태가 찍히므로, 모두 제때 끝났다면 마지막 줄 하나만 나와야 합니다.

producer-1 WAITING
producer-2 WAITING
consumer-1 WAITING
consumer-2 WAITING
taken=3/2400

2,400개 가운데 3개만 오간 뒤 네 스레드가 모두 WAITING 상태로 남았습니다. WAITING은 스레드가 wait()처럼 시간 제한 없는 대기에 들어가 있다는 뜻입니다. 네 스레드가 모두 대기 집합에서 알림을 기다리고 있으니, 알림을 보내 줄 스레드는 하나도 남지 않은 것입니다. 마지막 줄은 interrupt()로 스레드들을 끝낸 뒤에 출력한 값입니다.

이 프로그램을 20번 실행했더니 20번 모두 이렇게 멈췄고, 멈추기 전에 오간 값의 수는 2개에서 63개 사이로 실행할 때마다 달랐습니다. 상태를 출력하는 줄의 순서는 스레드를 workers 목록에 넣은 순서라서 늘 같습니다. 여러분의 환경에서는 숫자가 다르게 나올 수 있고, notify()가 매번 운 좋게 알맞은 스레드를 고르면 끝까지 진행될 수도 있습니다. 하지만 한 번 끝까지 갔다고 해서 코드가 맞는 것은 아닙니다.

모두가 잠드는 과정을 한 단계씩 따라가기

출력만으로는 실제 실행에서 스레드들이 어떤 순서로 움직였는지 알 수 없습니다. 그래도 멈출 수 있는 순서 하나를 직접 따라가 보면 원인이 드러납니다. 생산자를 P1, P2, 소비자를 C1, C2라고 하고, 처음에는 큐가 비어 있다고 하겠습니다. 표의 대기 집합 열에는 그 단계가 끝났을 때 대기 집합에 있는 스레드를 기다리기 시작한 순서대로 적었습니다.

3단계에서 깨어난 C1은 곧바로 실행되지 않습니다. 대기 집합에서 나온 뒤 모니터를 다시 얻어야 하는데, 깨어난 스레드라고 해서 우선권이 있지는 않습니다. 그래서 put()에서 돌아와 다음 값을 넣으려는 P1이 먼저 모니터를 얻을 수 있고, 4단계처럼 큐가 가득 찬 것을 보고 기다리게 됩니다.

문제는 6단계입니다. C1이 값을 꺼내 빈칸을 만들었으니 이 변화를 쓸 수 있는 쪽은 생산자인데, notify()는 같은 소비자인 C2를 골랐습니다. C2는 8단계에서 while로 조건을 다시 확인하고, 큐가 비어 있으니 다시 기다립니다. 여기까지는 while이 제 할 일을 한 것입니다. 하지만 C2는 큐를 바꾸지 않았으므로 notify()를 부르지 않고, 6단계의 알림은 아무도 진행시키지 못한 채 사라집니다. 8단계가 끝나면 네 스레드가 모두 대기 집합에 있으므로 이제 notify()를 호출할 스레드가 없습니다.

이 순서에서 notify()는 두 번 모두 가장 오래 기다린 스레드를 골랐습니다. 공평하게 순서대로 골라도 멈출 수 있다는 뜻입니다. 원인은 선택이 무작위라는 데 있지 않습니다. 서로 다른 조건을 기다리는 스레드가 한 대기 집합에 섞여 있는데, notify()가 그 차이를 모른다는 데 있습니다.

정리하면 while 재검사는 잘못 깨어난 스레드가 공유 상태를 망가뜨리지 않게 해 줍니다. 이처럼 나쁜 일이 일어나지 않는 성질을 안전성(safety)이라고 합니다. 하지만 필요한 스레드가 결국 깨어나 프로그램이 앞으로 나아가는 성질, 즉 생존성(liveness)까지 지켜 주지는 못합니다. 이 버퍼는 틀린 값을 내놓지는 않지만 영원히 멈출 수 있습니다.

더 깊이: 이 정지는 교착 상태 탐지에 잡히지 않습니다

프로그램이 멈추면 교착 상태(deadlock)를 먼저 떠올리기 쉽습니다. 교착 상태는 스레드들이 서로 상대가 쥔 락을 기다리며 원을 이룬 상태입니다. 위의 정지는 그렇지 않습니다. 네 스레드는 모두 모니터를 놓고 대기 집합에서 알림을 기다릴 뿐, 어느 스레드도 락을 쥐고 있지 않습니다.

그래서 JDK의 교착 상태 탐지로는 이 정지를 찾을 수 없습니다. 위 프로그램에서 스레드들을 인터럽트하기 직전에 ManagementFactory.getThreadMXBean().findDeadlockedThreads()를 호출해 보면, 교착 상태인 스레드가 없다는 뜻인 null이 돌아옵니다. 스레드 덤프에서도 모든 스레드가 WAITING으로 보일 뿐입니다. 이런 정지는 각 스레드가 어떤 조건을 기다리는지, 그 조건을 참으로 만들어 줄 스레드가 남아 있는지를 따져 봐야 찾을 수 있습니다. 스레드 상태와 스레드 덤프를 읽는 방법은 같은 장의 뒤쪽 글에서 다룹니다.

notifyAll로 바꾸면 끝까지 진행합니다

같은 프로그램에서 notify() 두 곳만 notifyAll()로 바꿔 보겠습니다. 나머지 코드는 그대로입니다.

    synchronized void put(int value) throws InterruptedException {
        while (queue.size() == capacity) wait();
        queue.add(value);
        notifyAll(); // 기다리는 스레드를 모두 깨운다
    }

    synchronized int take() throws InterruptedException {
        while (queue.isEmpty()) wait();
        int value = queue.remove();
        taken++;
        notifyAll(); // 기다리는 스레드를 모두 깨운다
        return value;
    }

실행 결과는 한 줄입니다. 앞의 결과와 달리 이름과 상태를 찍은 줄이 없고, 꺼낸 값의 수만 남았습니다.

taken=2400/2400

이번에는 20번 실행해 20번 모두 2,400개를 다 주고받았고, 매번 1초도 걸리지 않았습니다. 모든 스레드가 join(500)이 기다리는 동안 끝났으므로 살아 있는 스레드를 출력하는 줄은 나오지 않습니다.

앞의 6단계에 대입해 보면 이유를 알 수 있습니다. C1이 값을 꺼내고 notifyAll()을 부르면 C2, P1, P2가 모두 깨어납니다. 세 스레드는 차례로 모니터를 얻어 조건을 다시 확인하는데, C2는 큐가 비어 있으니 다시 기다리지만 P1과 P2 가운데 먼저 모니터를 얻은 쪽은 값을 넣고 다시 notifyAll()을 부릅니다. 이렇게 큐가 바뀔 때마다 기다리던 스레드가 모두 조건을 다시 확인할 기회를 얻으므로, 그 변화를 쓸 수 있는 스레드가 있다면 그 스레드도 깨어난 스레드 가운데 들어 있습니다. 알림이 엉뚱한 스레드에게 가서 사라지는 일이 생기지 않습니다.

이 방법은 조건을 while로 다시 검사할 때만 안전합니다. 모두를 깨우면 조건이 맞지 않는 스레드도 함께 깨어나기 때문입니다. 실제로 위 프로그램의 두 while을 if로 바꾸고 notifyAll()을 쓰게 해서 10번 실행했더니, 7번은 소비자 하나가 빈 큐에서 값을 꺼내다 NoSuchElementException으로 죽었습니다. 먼저 모니터를 얻은 소비자가 값을 가져간 뒤, 함께 깨어난 다른 소비자가 조건을 다시 보지 않고 remove()를 실행했기 때문입니다.

notify를 써도 되는 경우

notify()가 늘 틀린 것은 아닙니다. 위 프로그램에서 PRODUCERS와 CONSUMERS를 모두 1로 바꾸면 notify()만으로도 20번 모두 끝까지 진행했습니다. 생산자와 소비자가 하나씩이면, 한쪽이 기다리는 동안 큐를 바꿀 수 있는 스레드는 다른 한쪽뿐입니다. 그 스레드가 큐를 바꾸고 notify()를 부를 때 대기 집합에 있는 스레드는 기다리던 한쪽뿐이므로, 알림은 언제나 알맞은 상대에게 갑니다.

반면 생산자 하나에 소비자 둘, 또는 생산자 둘에 소비자 하나로 바꾸자 20번 모두 멈췄습니다. 총수를 2,400으로 정한 것은 1부터 4까지 어느 수로도 나누어떨어져서 이렇게 스레드 수를 바꿔 보기 좋기 때문입니다. 일반적으로 notify()로 충분하려면 다음 두 조건이 함께 맞아야 합니다. 『Java Concurrency in Practice』(Brian Goetz 외)가 정리한 기준입니다.

• 대기 집합의 스레드가 모두 같은 조건을 기다리고, 깨어난 뒤 하는 일도 같아야 합니다. 그래야 누구를 깨워도 알림이 헛되지 않습니다.
• 상태 변화 한 번으로 진행할 수 있는 스레드가 많아야 하나여야 합니다. 여럿이 진행해야 하는데 하나만 깨우면 나머지는 계속 기다립니다.

생산자와 소비자가 섞여 기다리는 버퍼는 첫째 조건을 어깁니다. 둘째 조건을 어기는 예로는 설정이 바뀌기를 기다리는 구독자 모두에게 새 설정을 알리는 경우가 있는데, 연습 문제 1에서 직접 확인합니다.

두 조건은 지금 맞더라도 코드를 고치다 보면 쉽게 깨집니다. 생산자와 소비자가 하나씩이라 안전하던 코드도 누군가 소비자를 하나 더 붙이는 순간 멈출 수 있게 되고, 이런 버그는 테스트에서 잘 드러나지 않습니다. 그래서 두 조건이 구조적으로 늘 맞는다고 확신할 수 없다면 notifyAll()을 기본으로 쓰는 편이 안전합니다. 지금까지의 내용을 표로 정리하면 다음과 같습니다.

notifyAll이 치르는 비용

notifyAll()의 대가는 헛되이 깨어나는 스레드입니다. 깨어난 스레드는 모니터를 차례로 얻어 조건을 확인하고, 조건이 맞지 않으면 다시 잠듭니다. 깨우고 다시 재울 때마다 문맥 전환(context switch)과 모니터 경쟁이 따릅니다. 그 횟수를 직접 세어 보려고 앞 글의 버퍼에 필드 두 개를 붙였습니다. waits는 wait()를 호출하기 직전에, wakeups는 wait()가 정상적으로 돌아온 직후에 1씩 늘어나며, 두 필드 모두 synchronized 메서드 안에서만 바꾸고 읽습니다.

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

main은 소비자 셋을 먼저 시작하고 30ms 동안 잠들어, 소비자들이 빈 큐에서 기다릴 시간을 줍니다. 그다음 값 11, 22, 33을 차례로 넣습니다. 용량이 2이므로, 소비자들이 값을 가져가기 전에 세 번째 값을 넣으려 하면 main도 기다려야 합니다. 한 번 실행한 결과입니다.

value=11
value=22
value=33
waits=5 wakeups=5

세 값은 각각 한 번씩 출력되지만, 어느 소비자가 어느 값을 받는지와 출력 순서는 실행할 때마다 달라집니다. waits는 소비자 셋이 처음 한 번씩 기다린 3에, 그 뒤에 생긴 대기가 더해진 값입니다. 뒤에 생긴 대기는 큐가 가득 차 main이 기다린 경우이거나, notifyAll()로 깨어난 소비자가 다른 소비자가 이미 값을 가져간 빈 큐를 보고 다시 기다린 경우입니다. 그래서 이 숫자도 실행마다 달라서, 여러 번 실행해 보면 대부분 4가 나오고 가끔 3이나 5, 6이 나오기도 합니다. 모든 wait()가 예외 없이 돌아왔으므로 프로그램이 끝난 뒤에는 두 숫자가 같고, 대기 중에 인터럽트되어 InterruptedException이 나면 그 호출의 wakeups는 늘지 않습니다.

도표는 같은 큐 상태에서 put과 take가 서로 다른 조건을 검사한다는 점을 보여 줍니다. 크기가 0이면 take만 기다리고, 용량만큼 차 있으면 put만 기다리며, 그 사이에서는 둘 다 진행할 수 있습니다. notifyAll()로 함께 깨어난 스레드들이 모니터를 얻는 순서는 정해져 있지 않으므로, 각자 모니터를 얻은 시점의 크기로 조건을 다시 판단합니다.

오래 주고받을 때의 대기 횟수

스레드와 값이 많아지면 숫자도 커집니다. 다음 프로그램은 생산자 넷이 50개씩 넣고 소비자 넷이 50개씩 꺼내도록 용량 4인 같은 버퍼를 씁니다. 넣는 개수와 꺼내는 개수를 맞춰 두었으므로 종료 표시나 강제 종료 없이도 모든 스레드가 스스로 끝납니다. 끝나면 소비자들이 받은 값의 합과 extraWakeups를 출력합니다. 앞의 NotifyAllBoundedQueue.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));
    }
}

sum=19900
extraWakeups=0

생산자들이 넣는 값은 0부터 199까지 200개이므로, 모두 한 번씩 소비됐다면 합은 (0 + 199) × 200 ÷ 2 = 19,900입니다. 10번 실행해 10번 모두 이 합이 나왔습니다. 값이 하나라도 사라지거나 두 번 소비되면 합이 대개 달라지므로 간단한 검사로 쓸 만하지만, 우연히 같은 합이 나오는 잘못도 있습니다. 연습 문제 3에서 그런 예를 다룹니다.

extraWakeups는 wakeups에서 값의 개수 200을 뺀 값이고, 음수면 0으로 출력합니다. 위 출력처럼 0이 나오는 경우가 많지만 실행에 따라 0보다 큰 값도 나옵니다. 이 숫자를 헛되이 깨어난 횟수로 읽기 쉽지만 그렇지 않습니다. 시험 삼아 wakeups++ 바로 다음에서 조건이 여전히 거짓인지, 즉 깨어났는데 곧 다시 기다려야 하는지를 따로 세어 보았습니다. 25번 실행 가운데 20번은 extraWakeups가 0이었는데, 그 20번에서도 헛되이 깨어난 횟수는 실행마다 18번에서 93번 사이였습니다. 깨어난 횟수의 합이 값의 개수보다 적어도 그 안에 아무 일도 못 하고 다시 잠든 깨움이 섞여 있는 것입니다. 게다가 이런 카운터는 wait()가 돌아온 횟수만 셀 뿐, 스레드가 모니터를 얻으려고 기다린 시간이나 운영체제 안에서 일어난 깨움은 보여 주지 않습니다.

도표는 네 숫자가 코드의 어디에서 늘어나거나 계산되는지 정리한 것입니다. notifyAll()과 다른 방식의 비용을 비교하려면 같은 부하에서 헛되이 깨어난 횟수와 처리 시간을 함께 재야 하는데, 이 프로그램은 그런 비교까지 하지는 않습니다. 큐가 작고 스레드가 적어서 이 비용이 실제 지연으로 이어지지 않는다면, 단순한 notifyAll() 구현을 그대로 두는 것도 합리적인 선택입니다.

더 깊이: 대기자가 많을수록 비용이 커지는 이유

notifyAll() 한 번은 기다리던 스레드 n개를 모두 깨웁니다. 상태 변화 한 번으로 진행할 수 있는 스레드가 하나뿐이라면, 나머지 n−1개는 모니터를 차례로 얻어 조건을 확인한 뒤 다시 잠듭니다. 그래서 변화 한 번마다 n에 비례하는 깨움과 모니터 경쟁이 생기고, 이런 변화가 n번 이어지면 깨움의 총수는 n²에 비례할 수 있습니다. 하나씩만 깨우면 n번으로 끝날 일입니다.

그렇다고 notify()로 되돌아가면 앞에서 본 것처럼 정확성을 잃습니다. 깨움 비용이 실제로 병목이라면 대기실 자체를 나누는 것이 다음 단계입니다. 생산자는 '빈칸 생김'을, 소비자는 '값 들어옴'을 따로 기다리게 하면, 값을 넣은 생산자는 소비자 쪽만 깨울 수 있습니다. 내장 모니터로는 객체 하나에 대기 집합을 하나밖에 둘 수 없으므로, 이때는 다음 글에서 다루는 Condition을 씁니다. 순서는 먼저 notifyAll()로 정확하게 만들고, 측정으로 병목을 확인한 뒤에 나누는 것입니다.

정리

• 객체마다 대기 집합은 하나뿐이라서, 생산자와 소비자가 같은 객체에서 wait()를 부르면 한곳에 섞여 기다립니다.
• notify()는 깨울 스레드가 무엇을 기다리는지 보지 않습니다. 같은 역할을 깨우면 그 스레드는 다시 잠들고 알림은 사라지므로, 모든 스레드가 잠든 채 멈출 수 있습니다.
• while 재검사는 안전성을 지켜 줄 뿐, 필요한 스레드가 깨어나는 생존성까지 지켜 주지는 않습니다.
• notifyAll()은 기다리던 스레드 모두에게 조건을 다시 확인할 기회를 주어 이 정지를 막습니다. 대신 헛되이 깨어났다가 다시 잠드는 스레드가 생기고, 대기자가 많을수록 그 비용이 커집니다.
• notify()는 모든 대기자가 같은 조건을 기다리고 변화 한 번에 한 스레드만 진행할 수 있을 때만 씁니다. 확신할 수 없으면 notifyAll()을 씁니다.

연습 문제

1. 여러 작업자가 설정이 바뀌기를 기다립니다. 게시자는 설정 문자열을 바꾸고 버전을 1 올립니다. 작업자는 자신이 마지막으로 본 버전을 넘기고, 그보다 새 버전이 게시될 때까지 기다렸다가 버전과 설정을 함께 받습니다. 한 번의 게시가 그때 기다리던 작업자 모두에게 전달되도록 awaitNewerThan(long observed)와 publish(String next)를 구현하세요. 그리고 notifyAll() 대신 notify()를 쓰면 어떻게 되는지 설명하세요.
2. 생산자와 소비자가 하나씩일 때는 notify()만 써도 멈추지 않는 이유를 대기 집합에 누가 있을 수 있는지로 설명하세요. 그리고 소비자를 하나 더 늘렸을 때 모든 스레드가 잠드는 순서를 하나 적어 보세요.
3. 입력 값이 0, 1, 2, 3인데 소비 기록이 0, 0, 3, 3이었다고 합시다. 개수 검사와 합계 검사를 통과하는지 계산하고, 어떤 값이 빠지거나 중복됐는지 찾으세요. 이런 잘못까지 잡으려면 무엇을 비교해야 하나요?

import java.util.ArrayList;
import java.util.Collections;
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 = 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

다음 글

ReentrantLock과 Condition에서는 같은 버퍼를 synchronized와 wait()·notifyAll() 대신 명시적 락과 Condition으로 만듭니다. Condition은 락 하나에 대기실을 여러 개 둘 수 있게 해 주며, 그다음 글에서는 이를 이용해 생산자와 소비자가 기다리는 곳을 나눕니다.