유한 버퍼 계약

생산자와 소비자는 같은 큐를 보지만 서로 다른 시간을 삽니다. 수집기는 짧은 순간에 이벤트를 몰아서 넣고, 저장기는 디스크 속도만큼만 꺼냅니다. 이 차이를 흡수하는 유한 버퍼는 단순한 Queue 포장이 아닙니다. 가득 찼을 때 누가 기다리는지, 어떤 값을 버리는지, 호출자가 실패를 어떻게 알아채는지를 결정하는 작은 운영 정책입니다.

첫 실험의 목표는 스레드를 많이 만드는 것이 아니라 손실을 관찰 가능한 값으로 바꾸는 것입니다. put이라는 이름을 붙이고도 내부에서 조용히 버리면 생산자는 저장되었다고 믿습니다. 반대로 빈 큐에서 null을 돌려주면서 실제 데이터로 null도 허용하면 소비자는 종료와 값 부재를 구별하지 못합니다. API의 반환형과 허용 값부터 정해야 이후 대기 전략을 올바르게 논의할 수 있습니다.

거짓 저장 로그 재현

아래 클래스는 임계 영역 자체는 synchronized로 보호합니다. 따라서 ArrayDeque의 내부 구조가 동시에 깨지지는 않습니다. 그러나 용량이 찼을 때 새 항목을 버리고도 호출자에게 결과를 알리지 않습니다. 실행하면 생산 측 로그에는 세 건이 모두 완료된 것처럼 보이지만 소비 측 집계는 두 건뿐입니다. 데이터 경쟁은 없는데 업무 데이터가 사라지는 전형적인 규칙 실패입니다.

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

마지막 출력은 saved=null입니다. 여기서 잘못은 큐가 유한하다는 사실이 아니라, 생산 완료와 실제 수용을 같은 사건처럼 기록한 데 있습니다. 버퍼가 넘치면 즉시 거부하는 정책도 충분히 타당할 수 있습니다. 다만 호출자가 false나 예외를 받아 재시도·별도 보관·폐기 계수 증가 중 하나를 선택할 수 있어야 합니다. 용량 초과를 숨기는 순간 시스템의 손실률은 측정할 수 없는 값이 됩니다.

용량 불변식과 결과 타입

유한 버퍼의 핵심 불변식은 언제나 0 <= size && size <= capacity입니다. 이 식만으로는 충분하지 않습니다. 삽입 결과에는 수용 여부가 있어야 하고, 꺼내기 결과에는 현재 비어 있다는 의미가 있어야 합니다. 이 예제에서는 null 저장을 금지하고 Optional<String>으로 부재를 표현합니다. 그러면 소비자는 실제 문자열과 비어 있는 순간을 혼동하지 않습니다.

생성 시 용량이 1보다 작으면 즉시 거부하여 영원히 수용할 수 없는 객체를 만들지 않습니다.
삽입은 잠금 안에서 크기를 검사하고 같은 임계 영역에서 원소를 추가해야 검사와 변경 사이 틈이 없습니다.
즉시 수용할 수 없으면 OfferResult.FULL을 반환하여 호출자가 과부하 정책을 결정하게 합니다.
제거는 값이 없을 때 Optional.empty()를 사용하고 큐 원소에는 null을 허용하지 않습니다.
진단용 스냅샷은 내부 가변 컬렉션을 노출하지 않고 복사본을 반환해야 외부 변경이 차단됩니다.
생산 개수와 소비 개수뿐 아니라 거부 개수도 따로 세어야 총량 보존식을 확인할 수 있습니다.

총량 보존식은 attempted = accepted + rejected입니다. 소비가 모두 끝난 시점에는 accepted = consumed + remaining도 만족해야 합니다. 두 식은 멀티스레드 실행 순서와 무관하게 확인할 수 있는 회귀 기준입니다. 로그 한 줄의 순서를 정답으로 삼으면 스케줄러가 달라질 때 정상 실행도 실패로 오판합니다.

즉시 거부형 버퍼의 완전 구현

다음 구현은 기다리지 않는 정책을 의도적으로 선택합니다. offer라는 이름, 열거형 결과, 불변 스냅샷이 같은 방향을 가리킵니다. 여러 생산자가 동시에 들어와도 크기 검사와 추가가 하나의 모니터 안에서 수행되므로 용량을 넘지 않습니다. 이 절의 관심사는 대기시키는 법이 아니라, 대기하지 않는 선택도 정직한 API로 만들 수 있다는 점입니다.

src/ObservableBoundedBuffer.java

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입니다. 합계가 맞으므로 구매 이벤트가 저장되지 않았다는 사실도 운영 지표에 남습니다. 이 정책에서 재시도는 버퍼 내부 책임이 아닙니다. 수집 지연을 허용할 수 없는 호출자는 거부 항목을 디스크 스풀로 넘길 수 있고, 최신 상태만 중요하다면 오래된 값을 제거하는 별도 버퍼를 선택할 수 있습니다.

종료 토큰을 데이터와 분리하는 소비 루프

빈 큐는 작업 종료를 뜻하지 않습니다. 생산자가 잠시 느린 것일 수 있기 때문입니다. 기다리지 않는 버퍼를 폴링하는 소비자는 poll()의 빈 결과를 보고 즉시 종료해서는 안 됩니다. 아래 예제는 생산 완료 플래그와 버퍼의 빈 상태를 함께 확인합니다. volatile 플래그가 생산 스레드가 쓴 값을 소비 스레드에 공개하고, 처리 건수는 소비 스레드 하나만 변경합니다.

src/PollingDrainDemo.java

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

이 구현은 짧은 실험에는 맞지만 실제 서비스에서 빈 큐를 계속 확인하면 CPU를 낭비합니다. 뒤 절에서 wait, Condition, BlockingQueue로 잠들게 만드는 이유가 여기에 있습니다. 먼저 종료 조건과 손실 규칙을 고정했기 때문에 대기 기술을 바꾸더라도 사용자 관점의 보존식은 그대로 재사용할 수 있습니다.

즉시 반환 정책을 선택할 때 묻는 질문

운영 질문	관측 지표	적합한 대응
한 건도 잃으면 안 되는가	거부 건수와 재처리 성공률	호출자를 막거나 내구 스풀 사용
최신 값만 의미가 있는가	같은 키의 갱신 빈도	오래된 항목 교체 정책 검토
요청 지연 상한이 짧은가	`offer` 지연 백분위	즉시 실패와 명시적 폴백
생산 폭주가 일시적인가	버퍼 점유율 지속 시간	제한된 시간 대기 고려

capacity를 크게 잡는 것만으로 정책 결정이 사라지지 않습니다. 큰 큐는 손실 시점을 늦추는 대신 오래된 작업의 지연과 메모리 점유를 늘립니다. 처리율이 장기적으로 생산률보다 낮다면 어떤 크기도 결국 찹니다. 용량은 정상적인 순간 폭주를 흡수할 만큼만 두고, 과부하 때의 반환값과 호출자 행동을 별도로 설계해야 합니다.

연습 문제

온도 센서는 같은 장치의 측정치를 빠르게 보냅니다. 용량이 찼을 때 무조건 새 값을 거부하지 말고, 동일한 센서 ID가 이미 있으면 그 항목만 최신 측정으로 교체하세요. 다른 ID만 가득 차 있다면 FULL을 반환해야 합니다. 교체와 신규 수용을 구별할 수 있는 결과 타입도 설계합니다.

정답과 해설

배열이나 덱만 사용하면 ID 검색이 선형 시간이지만 작은 상태 버퍼에서는 정책의 명료성이 먼저입니다. 잠금 안에서 기존 위치 탐색, 교체, 용량 검사를 모두 수행해야 두 생산자가 같은 ID를 동시에 추가하는 일을 막을 수 있습니다. 결과를 세 종류로 나누면 운영 지표에서 신규 수용과 덮어쓰기를 분리할 수 있습니다.

exercise/LatestReadingBufferSolution.java

import java.util.LinkedHashMap;
import java.util.Map;
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이고 마지막에 A의 20.8도 측정이 나옵니다. 교체가 삽입 순서를 바꿔야 하는지는 별도의 제품 결정입니다. 현재 LinkedHashMap의 단순 put은 기존 키의 순서를 유지하므로 “처음 들어온 센서부터 처리”라는 해석을 택했습니다.

이 절에서 남겨야 할 판단 기준

스레드 안전하다는 말과 업무적으로 손실이 없다는 말은 다릅니다. 모니터는 컬렉션의 동시 변경을 직렬화하지만, 가득 찬 상황에서 무엇을 할지는 API 설계자가 정해야 합니다. 시도·수용·거부·소비·잔여 수를 서로 맞춰 볼 수 있을 때 비로소 유한 버퍼의 동작을 확인할 수 있습니다.

다음 대기 구현으로 넘어가기 전에 FULL이 정상 분기인지 장애인지 결정하세요. 정상 분기라면 반환값과 지표를 준비하고, 장애라면 예외나 내구 저장소를 사용합니다. 이 선택을 미룬 채 void put만 노출하면 가장 중요한 과부하 정보가 사라집니다.