유한 버퍼와 생산자-소비자 문제

웹 서비스의 로그 수집기를 생각해 보겠습니다. 요청이 몰리면 수집 스레드는 짧은 순간에 이벤트를 수백 건씩 만들어 내지만, 이벤트를 디스크에 쓰는 저장 스레드는 디스크 속도만큼만 처리할 수 있습니다. 그래서 둘 사이에 큐를 하나 두고, 수집 스레드는 큐에 넣고 저장 스레드는 큐에서 꺼내게 합니다. 이처럼 일을 만들어 내는 쪽을 생산자(producer), 일을 가져가 처리하는 쪽을 소비자(consumer), 둘이 함께 쓰는 큐를 버퍼(buffer)라고 부릅니다.

버퍼가 있으면 두 쪽이 서로의 속도에 맞춰 움직이지 않아도 됩니다. 하지만 메모리는 무한하지 않으므로 버퍼에는 보통 최대 크기를 정해 두는데, 이런 버퍼를 유한 버퍼(bounded buffer)라고 합니다. 크기를 정하는 순간 답해야 할 질문이 생깁니다. 버퍼가 가득 찼는데 생산자가 또 넣으려 하면 어떻게 할까요? 버퍼가 비었는데 소비자가 꺼내려 하면 어떻게 하고, 비어 있는 것이 생산자가 잠시 느려서인지 일이 정말 끝나서인지는 어떻게 구별할까요?

생산자와 소비자가 유한 버퍼를 함께 쓰면서도 값을 잃거나 엉뚱한 값을 읽지 않게 하는 것이 동시성 프로그래밍의 고전적인 문제인 생산자-소비자 문제(producer-consumer problem)입니다. 유한 버퍼 문제(bounded-buffer problem)라고도 부릅니다. 이 문제는 두 가지 일로 나뉩니다. 하나는 여러 스레드가 동시에 버퍼를 건드려도 내부 구조가 깨지지 않게 하는 일이고, 다른 하나는 가득 찼을 때와 비었을 때 무엇을 할지 정해서 그 결과를 호출한 쪽에 알리는 일입니다. 첫 번째는 synchronized로 해결되지만, 두 번째는 버퍼를 설계하는 사람이 직접 정해야 합니다.

이 장은 이 문제를 단계적으로 풉니다. 이 글에서는 기다리지 않는 버퍼, 즉 가득 차면 바로 거절하고 비어 있으면 '지금은 없다'고 바로 알려 주는 버퍼를 만듭니다. 다음 글부터는 생산자와 소비자를 기다리게 하는 방법을 다루는데, 락을 쥔 채 기다리면 왜 멈추는지 확인한 뒤 wait·notify와 Condition을 거쳐 표준 라이브러리의 ArrayBlockingQueue로 마무리합니다. 예제에는 19장에서 배운 synchronized와 volatile, 16장에서 배운 Queue와 ArrayDeque가 나옵니다.

가득 차면 조용히 버리는 버퍼

먼저 흔히 저지르는 실수부터 보겠습니다. 아래 SilentDropBuffer는 용량이 정해진 큐로, put과 take에 모두 synchronized를 붙였습니다. 그래서 여러 스레드가 동시에 호출해도 ArrayDeque의 내부 구조가 깨지지는 않습니다. 문제는 가득 찼을 때입니다. put은 새 값을 넣지 않고 그냥 return하는데, 반환형이 void라서 호출한 쪽은 값이 들어갔는지 알 방법이 없습니다.

이 글의 예제는 코드 위에 적힌 경로와 이름 그대로 저장합니다. 다른 파일이 필요 없는 예제는 java bad/SilentDropBuffer.java처럼 소스 파일 모드로 바로 실행할 수 있습니다.

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

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

    public SilentDropBuffer(int capacity) {
        this.capacity = capacity;
    }

    public synchronized void put(String value) {
        if (queue.size() == capacity) {
            return;
        }
        queue.add(value);
    }

    public synchronized String take() {
        return queue.poll();
    }

    public static void main(String[] args) {
        SilentDropBuffer buffer = new SilentDropBuffer(2);
        for (String event : new String[]{"login", "search", "purchase"}) {
            buffer.put(event);
            System.out.println("accepted=" + event);
        }
        System.out.println("saved=" + buffer.take());
        System.out.println("saved=" + buffer.take());
        System.out.println("saved=" + buffer.take());
    }
}

put은 예외도 반환값도 없으므로, main은 put이 돌아오기만 하면 accepted= 줄을 찍습니다. 출력의 accepted= 세 줄과 saved= 세 줄을 짝지어 보세요.

accepted=login
accepted=search
accepted=purchase
saved=login
saved=search
saved=null

main은 put을 부를 때마다 accepted= 로그를 남기므로, 처음 세 줄만 보면 이벤트가 모두 들어간 것 같습니다. 하지만 용량이 2이므로 세 번째 put("purchase")가 호출될 때 큐에는 이미 login과 search가 있고, put은 아무것도 하지 않고 돌아옵니다. 그래서 세 번 꺼내면 login과 search 다음에 null이 나옵니다. take가 쓰는 poll은 큐가 비어 있으면 null을 돌려주기 때문입니다.

로그의 saved는 '큐에서 꺼냈다'는 뜻일 뿐, 파일이나 데이터베이스에 저장했다는 뜻이 아닙니다. 이 글의 예제는 모두 메모리 안의 큐에 넣고 꺼내는 동작만 다룹니다.

이 버퍼에는 데이터 경합(data race)이 없습니다. 모든 접근이 같은 락 안에서 일어나고, 크기도 용량 2를 넘은 적이 없습니다. 그런데도 이벤트 하나가 사라졌고, 더 나쁜 것은 아무도 그 사실을 모른다는 점입니다. 스레드 안전하다는 것과 데이터를 잃지 않는다는 것은 서로 다른 이야기입니다.

잘못은 버퍼가 유한하다는 데 있지 않습니다. 가득 찼을 때 새 값을 거절하는 것 자체는 충분히 타당한 정책입니다. 잘못은 '거절했다'는 사실을 호출한 쪽에 알리지 않은 데 있습니다. put이 false를 돌려주거나 예외를 던졌다면, 호출한 쪽은 다시 시도하거나, 따로 보관하거나, 버린 개수를 세는 것 가운데 하나를 고를 수 있었습니다. 거절을 숨기면 시스템이 데이터를 얼마나 잃고 있는지조차 잴 수 없게 됩니다.

take의 반환값에도 같은 문제가 있습니다. 세 번째 결과 null은 '큐가 비어 있다'는 뜻인데, 반환형 String만 봐서는 null이 나올 수 있다는 것을 알 수 없고, 호출한 쪽이 검사를 잊으면 NullPointerException이 납니다. ArrayDeque는 null을 원소로 받지 않으므로 여기서는 null이 곧 '비어 있음'이지만, null을 넣을 수 있는 큐였다면 '꺼낸 값이 null이다'와 '비어 있다'를 구별할 수조차 없습니다.

메서드 이름도 오해를 부릅니다. 자바의 큐 메서드는 이름으로 실패 방식을 알려 줍니다. 16장에서 본 것처럼 add는 넣지 못하면 예외를 던지고 offer는 false를 돌려주며, 뒤에서 배울 BlockingQueue의 put은 자리가 날 때까지 기다립니다. 그런데 SilentDropBuffer의 put은 기다리지도 알리지도 않고 버립니다.

버퍼가 지켜야 할 조건과 돌려줄 결과

SilentDropBuffer도 지킨 조건이 하나 있습니다. 크기가 언제나 0 이상이고 용량 이하라는 것입니다. 이처럼 객체를 만든 뒤 어떤 메서드를 호출하든 호출이 끝난 시점에는 항상 참이어야 하는 조건을 불변식(invariant)이라고 하고, 유한 버퍼의 불변식은 0 <= size && size <= capacity입니다. 하지만 앞의 예에서 보았듯이 불변식을 지킨다고 해서 버퍼가 제대로 동작하는 것은 아닙니다.

그래서 결과도 정해야 합니다. 넣기는 '받아들였다'와 '가득 차서 거절했다'를 구별해서 돌려주고, 꺼내기는 '값이 있다'와 '지금은 비어 있다'를 구별해서 돌려줘야 합니다. 다음 절의 버퍼는 넣기 결과로 열거형 OfferResult의 ACCEPTED와 FULL을, 꺼내기 결과로 Optional<E>를 씁니다.

Optional<E>는 값이 하나 있거나 없음을 나타내는 표준 클래스입니다. Optional.ofNullable(x)는 x가 null이면 빈 Optional을, 아니면 x를 담은 Optional을 만들고, 받은 쪽은 isPresent()나 isEmpty()로 값이 있는지 확인합니다. 반환형에 '없을 수 있다'는 사실이 드러나므로 호출한 쪽이 빈 경우를 잊기 어렵습니다. Optional은 33장에서 자세히 배우고, 여기서는 이 정도만 알면 됩니다.

이 밖에 다음 규칙을 함께 정합니다. 앞의 네 가지는 잘못된 상태를 미리 막기 위한 것이고, 마지막 하나는 버퍼가 제대로 동작하는지 확인하기 위한 것입니다.

• 용량이 1보다 작으면 생성자에서 바로 예외를 던집니다. 용량 0인 버퍼는 아무것도 받아들일 수 없으므로, 모든 offer가 FULL을 돌려주는 이유를 나중에 찾는 것보다 만드는 순간 거절하는 편이 낫습니다.
• 크기 검사와 추가는 하나의 synchronized 메서드 안에서 함께 합니다. 둘 사이에 틈이 있으면 다른 스레드가 끼어들 수 있는데, 이 내용은 아래에서 따로 확인합니다.
• null은 원소로 받지 않습니다. 그래야 꺼내기 결과에서 '값'과 '없음'이 겹치지 않습니다.
• 내부 상태를 보여 주는 snapshot()은 내부 큐가 아니라 복사본을 돌려줍니다. 바깥에서 그 목록을 고쳐도 버퍼에는 영향이 없지만, 원소 객체 자체까지 복사하지는 않습니다.
• 받아들인 개수와 꺼낸 개수뿐 아니라 거절한 개수도 셉니다.

마지막 규칙이 필요한 이유는 개수끼리 맞춰 볼 수 있기 때문입니다. offer는 호출될 때마다 ACCEPTED와 FULL 가운데 정확히 하나를 돌려주므로, 시도한 횟수는 받아들인 수와 거절한 수의 합, 즉 attempted = accepted + rejected여야 합니다. 또 받아들인 원소는 꺼냈거나 아직 버퍼에 남아 있거나 둘 중 하나이므로, 소비가 끝난 시점에는 accepted = consumed + remaining도 성립해야 합니다. 이 글에서는 이 두 식을 보존식이라고 부르겠습니다.

보존식이 쓸모 있는 이유는 스레드가 어떤 순서로 실행되든 성립해야 하기 때문입니다. 여러 스레드가 섞여 실행되면 로그 줄의 순서는 실행할 때마다 달라지므로, 특정 순서를 정답으로 정해 두면 정상적인 실행도 실패로 판정하게 됩니다. 반면 개수의 합은 실행 순서와 상관없이 맞아야 하므로 테스트에서 확인하기에 알맞습니다.

가득 차면 FULL을 돌려주는 버퍼

다음 ObservableBoundedBuffer는 앞의 규칙을 그대로 코드로 옮긴 것입니다. 기다리지 않는 정책을 일부러 고른 버퍼라서, 메서드 이름도 put과 take 대신 '실패하면 결과로 알려 준다'는 뜻의 offer와 poll을 씁니다. main은 용량 2인 버퍼에 이벤트 세 개를 넣어 보고, 네 가지 개수를 한 줄로 출력합니다.

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

public final class ObservableBoundedBuffer<E> {
    public enum OfferResult { ACCEPTED, FULL }

    private final Queue<E> elements = new ArrayDeque<>();
    private final int capacity;

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

    public synchronized OfferResult offer(E element) {
        Objects.requireNonNull(element, "element");
        if (elements.size() >= capacity) {
            return OfferResult.FULL;
        }
        elements.add(element);
        return OfferResult.ACCEPTED;
    }

    public synchronized Optional<E> poll() {
        return Optional.ofNullable(elements.poll());
    }

    public synchronized int size() {
        return elements.size();
    }

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

    public static void main(String[] args) {
        ObservableBoundedBuffer<String> buffer = new ObservableBoundedBuffer<>(2);
        int accepted = 0;
        int rejected = 0;
        for (String event : List.of("login", "search", "purchase")) {
            if (buffer.offer(event) == OfferResult.ACCEPTED) accepted++;
            else rejected++;
        }
        int consumed = 0;
        while (buffer.poll().isPresent()) consumed++;
        System.out.printf("accepted=%d rejected=%d consumed=%d remaining=%d%n",
                accepted, rejected, consumed, buffer.size());
    }
}

네 숫자는 차례로 넣기에 성공한 수, 거절된 수, 꺼낸 수, 버퍼에 남은 수입니다. 시도한 이벤트 세 개가 이 숫자들 가운데 어디에 나타나는지 따라가 보세요.

accepted=2 rejected=1 consumed=2 remaining=0

login과 search는 자리가 있어 ACCEPTED가 되고, 세 번째 purchase는 크기가 이미 용량 2와 같으므로 FULL이 됩니다. 이번에는 거절이 반환값으로 드러나므로 main이 rejected를 셀 수 있습니다. 이어지는 while (buffer.poll().isPresent())는 값이 나오는 동안 꺼내기를 계속하므로 두 번 꺼낸 뒤 멈추고, 남은 원소는 0개입니다. 시도 3번은 수용 2와 거부 1의 합이고, 수용 2는 소비 2와 잔여 0의 합이므로 두 보존식이 모두 맞습니다.

이 main은 스레드 하나로 실행하므로 동시성을 시험하지는 않습니다. 개수를 세고 맞춰 보는 방법을 먼저 보여 주는 예제이고, 여러 스레드는 아래의 PollingDrainDemo에서 씁니다. 출력의 consumed도 값이 나온 poll 호출의 횟수일 뿐 저장을 마친 개수가 아닙니다.

코드에서 눈여겨볼 곳은 세 군데입니다. offer는 첫 줄의 Objects.requireNonNull로 null을 거절하는데, 예외 메시지 element로 무엇이 잘못됐는지 바로 알 수 있습니다. poll은 elements.poll()이 비었을 때 돌려주는 null을 Optional.ofNullable로 빈 Optional로 바꿉니다. size()와 snapshot()에도 synchronized를 붙였는데, 같은 락 안에서 읽어야 다른 스레드가 바꾼 최신 상태를 볼 수 있기 때문입니다. snapshot()이 쓰는 List.copyOf는 수정할 수 없는 복사본을 만듭니다.

FULL을 받은 뒤 무엇을 할지는 버퍼가 아니라 호출한 쪽이 정합니다. 수집이 늦어지면 안 되는 쪽은 거절된 이벤트를 디스크의 임시 파일에 따로 모아 둘 수 있고, 최신 상태만 중요하다면 오래된 값을 새 값으로 바꾸는 다른 버퍼를 쓸 수 있습니다. 버퍼가 할 일은 그 결정을 내릴 수 있도록 결과를 정확히 알려 주는 것입니다.

크기 검사와 추가를 나누면 안 되는 이유

offer가 크기 검사와 추가를 한 메서드 안에서 하는 데는 이유가 있습니다. 다음처럼 둘을 각각 synchronized 메서드로 나누면, 메서드 하나하나는 락을 잡지만 두 호출 사이에는 락이 없습니다.

// 잘못된 예: 검사와 추가가 서로 다른 임계 영역에 있습니다
public OfferResult offer(E element) {
    if (size() >= capacity) {    // 락을 잡고 크기를 읽은 뒤 놓는다
        return OfferResult.FULL;
    }
    add(element);                // 락을 다시 잡고 추가한다
    return OfferResult.ACCEPTED;
}

용량이 1이고 버퍼가 빈 상태에서 두 생산자가 거의 동시에 이 offer를 호출하면 다음 순서가 가능합니다.

두 생산자 모두 '자리가 있다'는 확인을 통과했지만, 확인한 뒤 추가하기 전에 상대가 끼어들었기 때문에 크기가 용량을 넘었습니다. 19장의 출금 예제에서 잔액 확인과 차감을 한 임계 영역에 묶어야 했던 것과 같은 이유입니다. 이런 끼어들기는 드물게 일어납니다. 이 환경에서 용량 1인 버퍼에 두 스레드가 동시에 넣는 실험을 2만 번 반복했더니 크기가 2가 된 경우는 18번이었고, 횟수는 실행할 때마다 다릅니다. 드물다는 것은 테스트를 몇 번 돌려서는 발견하기 어렵다는 뜻이기도 합니다.

빈 버퍼를 보고 바로 끝내면 안 되는 이유

이제 생산자와 소비자를 서로 다른 스레드로 돌려 보겠습니다. 소비자 쪽에서 먼저 정해야 할 것은 언제 멈출지입니다. poll()이 빈 결과를 돌려줬다고 해서 일이 끝난 것은 아닙니다. 생산자가 잠시 느려서 아직 다음 값을 넣지 못했을 수도 있기 때문입니다. 그래서 소비자에게는 '생산자가 더 넣을 것이 없다'는 신호가 따로 필요합니다.

아래 PollingDrainDemo는 그 신호로 volatile 필드 productionFinished를 씁니다. 생산자는 A부터 D까지 네 값을 넣은 뒤 이 필드를 true로 바꾸고, 소비자는 '생산이 끝났고 버퍼도 비었다'가 될 때까지 꺼내기를 반복합니다. 스레드는 JDK 21에 추가된 Thread.ofPlatform().start(...)로 만드는데, new Thread(...).start()와 같은 일을 합니다. 이 예제는 앞의 ObservableBoundedBuffer를 사용하므로, 두 파일을 같은 src 폴더에 두고 함께 컴파일해 실행합니다.

javac -d out src/ObservableBoundedBuffer.java src/PollingDrainDemo.java
java -cp out PollingDrainDemo

import java.util.Optional;

public final class PollingDrainDemo {
    private static final class RunState {
        private volatile boolean productionFinished;
        private int handled;
    }

    public static void main(String[] args) throws InterruptedException {
        ObservableBoundedBuffer<String> buffer = new ObservableBoundedBuffer<>(4);
        RunState state = new RunState();

        Thread producer = Thread.ofPlatform().start(() -> {
            try {
                for (String value : new String[]{"A", "B", "C", "D"}) {
                    ObservableBoundedBuffer.OfferResult result = buffer.offer(value);
                    if (result != ObservableBoundedBuffer.OfferResult.ACCEPTED) {
                        throw new IllegalStateException("unexpected overflow: " + value);
                    }
                }
            } finally {
                state.productionFinished = true;
            }
        });

        Thread consumer = Thread.ofPlatform().start(() -> {
            while (!state.productionFinished || buffer.size() > 0) {
                Optional<String> next = buffer.poll();
                if (next.isPresent()) {
                    state.handled++;
                } else {
                    Thread.onSpinWait();
                }
            }
        });

        producer.join();
        consumer.join();
        System.out.println("handled=" + state.handled);
    }
}

생산자는 네 값을 넣고, 소비자가 꺼낸 개수는 state.handled에 쌓입니다. main은 두 스레드를 모두 join()한 뒤에 이 값을 읽으므로, 넣은 값이 하나도 빠지지 않았다면 4가 나와야 합니다.

handled=4

두 스레드가 번갈아 실행되는 순서는 실행할 때마다 다릅니다. 소비자가 A를 꺼낸 뒤에 생산자가 B를 넣을 수도 있고, 생산자가 넷을 모두 넣은 뒤에야 소비자가 시작될 수도 있습니다. 하지만 출력은 언제나 handled=4입니다. 소비자는 productionFinished가 true인 것을 본 뒤에도 버퍼가 빌 때까지 반복을 계속하므로, 넣은 값을 하나도 빠뜨리지 않고 꺼내기 때문입니다. 이 환경에서 30번 실행했을 때도 모두 같은 결과가 나왔습니다.

나머지 코드에도 각각 이유가 있습니다. productionFinished는 생산자가 쓰고 소비자가 락 없이 읽는 필드이므로, 다른 스레드가 쓴 값이 보이도록 volatile을 붙였습니다. 반면 handled는 소비자 스레드만 바꾸고 main은 consumer.join()이 끝난 뒤에 읽습니다. join이 돌아온 뒤에는 그 스레드가 한 일이 모두 보이므로 handled에는 volatile이 필요 없습니다.

생산자의 finally는 예외가 나도 종료 신호를 보내기 위한 것입니다. 용량이 4이고 값도 넷이라 이 프로그램에서는 FULL이 나올 수 없지만, 용량을 줄이는 식으로 코드를 바꾸면 생산자가 IllegalStateException으로 끝날 수 있습니다. 그때 finally가 없다면 productionFinished는 영원히 false로 남고, 소비자는 빈 버퍼를 끝없이 확인하며 프로그램도 끝나지 않습니다. 종료 신호는 정상으로 끝날 때뿐 아니라 실패로 끝날 때도 보내야 합니다.

버퍼가 비어 있을 때 부르는 Thread.onSpinWait()는 '다른 스레드를 기다리며 바쁘게 반복하는 중'이라는 힌트를 런타임에 줄 뿐, 스레드를 재우지는 않습니다. 그래서 버퍼가 비어 있는 동안 소비자는 CPU 코어 하나를 계속 차지합니다. 이렇게 조건이 바뀔 때까지 반복해서 확인하는 방식을 바쁜 대기(busy waiting) 또는 폴링(polling)이라고 하는데, 짧은 실험에는 괜찮지만 실제 서비스에서는 CPU를 낭비합니다. 다음 글부터 소비자를 재워 두었다가 값이 들어오면 깨우는 방법을 찾는 이유가 여기에 있습니다.

더 깊이: 반복 조건에서 두 검사의 순서

소비자의 반복 조건 !state.productionFinished || buffer.size() > 0에서 두 검사의 순서는 바꾸면 안 됩니다. 순서를 뒤집어 buffer.size() > 0 || !state.productionFinished로 쓰면 다음 실행이 가능합니다.

소비자가 크기를 읽은 순간에는 버퍼가 비어 있었고, 플래그를 읽은 순간에는 생산이 끝나 있었습니다. 서로 다른 순간에 읽은 두 값을 합쳐 '비었고 끝났다'고 판단했기 때문에, 네 값은 버퍼에 남은 채 버려집니다.

원래 순서에서는 이런 일이 생기지 않습니다. volatile 필드에 쓴 값을 다른 스레드가 읽으면 쓴 스레드가 그전에 한 일도 모두 보이므로(19장의 happens-before 규칙), 플래그에서 true를 읽었다면 네 번의 offer도 이미 끝난 뒤입니다. 그 뒤에 읽는 크기에는 남은 값이 모두 반영되므로, 반복은 남은 값을 다 꺼낸 다음에만 끝납니다. 이 환경에서 소비자를 먼저 시작하는 실험 프로그램으로 두 순서를 2만 번씩 실행했더니, 뒤집은 순서는 776번 값을 남긴 채 끝났고 원래 순서는 한 번도 그러지 않았습니다.

가득 찼을 때 어떤 정책을 고를까

ObservableBoundedBuffer는 가득 차면 즉시 거절하는 정책을 골랐지만, 언제나 그것이 정답은 아닙니다. 어떤 정책이 맞는지는 데이터를 잃었을 때 얼마나 곤란한지, 호출한 쪽이 얼마나 오래 기다릴 수 있는지에 따라 달라집니다. 다음 표는 자주 만나는 상황과 그때 살펴볼 숫자, 알맞은 대응을 정리한 것입니다.

'최신 값만 의미가 있다'는 경우는 아래 연습 문제에서 직접 만들어 봅니다. 생산자를 기다리게 하는 방법과 정해진 시간만 기다리는 방법은 이 장의 뒤쪽 글에서 다룹니다.

용량을 크게 잡으면 이런 고민이 사라질 것 같지만 그렇지 않습니다. 생산자가 초당 1,000건을 넣고 소비자가 초당 800건을 꺼낸다면 버퍼는 초당 200건씩 늘어나므로, 용량이 10,000이어도 50초면 가득 찹니다. 큰 버퍼는 가득 차는 시점을 늦출 뿐이고, 그동안 쌓인 항목이 메모리를 차지하며 처리도 늦어집니다. 10,000건이 쌓인 상태라면 새로 들어온 항목은 앞의 10,000건이 처리될 때까지, 즉 12.5초를 기다려야 합니다. 그래서 용량은 정상적인 순간 폭주를 흡수할 만큼만 두고, 소비가 오랫동안 생산을 따라가지 못하는 상황은 반환값과 호출한 쪽의 대응으로 따로 처리해야 합니다.

정리

• 생산자-소비자 문제는 속도가 다른 생산자와 소비자가 유한 버퍼를 함께 쓰면서도 값을 잃거나 잘못 읽지 않게 하는 문제입니다. 가득 찼을 때, 비었을 때, 일이 끝났을 때 무엇을 할지 정해야 합니다.
• synchronized는 버퍼의 내부 구조를 지켜 주지만 가득 찼을 때 무엇을 할지는 정해 주지 않습니다. 스레드 안전한 버퍼도 값을 조용히 잃을 수 있습니다.
• 넣기 결과는 ACCEPTED와 FULL처럼, 꺼내기 결과는 Optional처럼 호출한 쪽이 구별할 수 있는 형태로 돌려줍니다. 크기 검사와 추가는 한 임계 영역 안에서 합니다.
• attempted = accepted + rejected와 accepted = consumed + remaining은 실행 순서와 상관없이 성립해야 하므로 동시성 코드를 확인하는 기준으로 쓸 수 있습니다.
• 빈 버퍼는 일이 끝났다는 뜻이 아닙니다. 소비자는 종료 신호와 빈 상태를 함께 확인해야 하고, 종료 신호는 실패할 때도 보내야 합니다.
• 버퍼를 계속 확인하는 바쁜 대기는 CPU를 낭비합니다. 용량을 키워도 소비가 생산을 따라가지 못하면 버퍼는 결국 가득 찹니다.

연습 문제

1. PollingDrainDemo의 반복 조건에서 || buffer.size() > 0을 지우고 while (!state.productionFinished)만 남기면 출력이 어떻게 달라질지 예상해 보세요. 실행할 때마다 결과가 같을까요?
2. PollingDrainDemo의 버퍼 용량을 4에서 2로 줄이면 어떤 결과들이 나올 수 있나요? 생산자의 finally가 없었다면 어떻게 되었을지도 설명하세요.
3. 온도 센서들이 측정값을 빠르게 보냅니다. 같은 센서의 값은 가장 최신 것만 의미가 있으므로, 버퍼가 가득 찼더라도 같은 센서 ID의 값이 이미 있으면 그 항목을 새 측정값으로 바꾸는 버퍼를 만드세요. 다른 센서의 값만으로 가득 차 있다면 FULL을 돌려줍니다. 새로 넣은 경우와 바꾼 경우를 구별할 수 있는 결과 타입도 설계하세요.

import java.util.LinkedHashMap;
import java.util.Objects;
import java.util.Optional;

public final class LatestReadingBufferSolution {
    record Reading(String sensorId, double celsius) {
        Reading {
            if (sensorId == null || sensorId.isBlank()) throw new IllegalArgumentException("sensorId");
        }
    }

    enum StoreResult { INSERTED, REPLACED, FULL }

    private final int capacity;
    private final LinkedHashMap<String, Reading> readings = new LinkedHashMap<>();

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

    synchronized StoreResult store(Reading reading) {
        Objects.requireNonNull(reading, "reading");
        if (readings.containsKey(reading.sensorId())) {
            readings.put(reading.sensorId(), reading);
            return StoreResult.REPLACED;
        }
        if (readings.size() == capacity) return StoreResult.FULL;
        readings.put(reading.sensorId(), reading);
        return StoreResult.INSERTED;
    }

    synchronized Optional<Reading> takeOldest() {
        var iterator = readings.entrySet().iterator();
        if (!iterator.hasNext()) return Optional.empty();
        Reading oldest = iterator.next().getValue();
        iterator.remove();
        return Optional.of(oldest);
    }

    public static void main(String[] args) {
        LatestReadingBufferSolution buffer = new LatestReadingBufferSolution(2);
        System.out.println(buffer.store(new Reading("A", 20.1)));
        System.out.println(buffer.store(new Reading("B", 21.0)));
        System.out.println(buffer.store(new Reading("A", 20.8)));
        System.out.println(buffer.store(new Reading("C", 19.7)));
        System.out.println(buffer.takeOldest().orElseThrow());
    }
}

INSERTED
INSERTED
REPLACED
FULL
Reading[sensorId=A, celsius=20.8]

다음 글

synchronized 안에서 sleep하면 멈추는 이유에서는 가득 찬 버퍼에서 생산자를 기다리게 해 봅니다. 가장 먼저 떠오르는 방법은 synchronized 메서드 안에서 Thread.sleep으로 잠깐씩 쉬며 빈자리를 기다리는 것인데, 이 방법은 프로그램을 영원히 멈추게 합니다. 왜 그런지 실행 결과와 스레드 덤프로 확인하고, 락을 놓고 기다리는 wait로 고칩니다.