ArrayBlockingQueue 위임

직접 만든 유한 큐가 put, take, 시간 제한, 즉시 반환, 종료까지 요구받기 시작하면 표준 라이브러리와 같은 문제를 다시 풀고 있는 셈입니다. ArrayBlockingQueue는 고정 용량 배열, 하나의 ReentrantLock, 생산·소비 조건을 검증된 구현으로 제공합니다. 학습용 큐에서 얻은 술어와 신호 지식은 라이브러리를 피하라는 근거가 아니라 API를 정확히 선택하기 위한 배경입니다.

위임해도 제품 결정은 남습니다. 용량을 얼마로 둘지, 생산자를 막을지, 종료를 어떤 값으로 알릴지, 인터럽트를 어디까지 전달할지는 애플리케이션 책임입니다. 표준 큐는 동시성 기계 장치를 맡고 도메인 계층은 의미 있는 메시지와 실패 정책을 정의합니다.

InterruptedException 은폐

아래 래퍼는 put 중 인터럽트를 잡고 로그만 남긴 뒤 반환합니다. 호출자는 제출이 끝났다고 판단하지만 실제 항목은 들어가지 않았을 수 있습니다. 표준 큐를 썼어도 예외 규칙을 훼손하면 조용한 유실이 다시 생깁니다. 얇은 위임 계층이 항상 안전하거나 유용한 것은 아닙니다.

bad/SwallowingQueueFacade.java

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

public final class SwallowingQueueFacade {
    private final BlockingQueue<String> queue = new ArrayBlockingQueue<>(1);

    void submit(String value) {
        try {
            queue.put(value);
        } catch (InterruptedException e) {
            System.err.println("ignored cancellation: " + value);
        }
    }

    public static void main(String[] args) throws InterruptedException {
        SwallowingQueueFacade facade = new SwallowingQueueFacade();
        facade.submit("occupy-slot");
        Thread producer = Thread.ofPlatform().start(() -> facade.submit("lost-task"));
        Thread.sleep(30);
        producer.interrupt();
        producer.join();
        System.out.println("queued=" + facade.queue.size());
        System.out.println("producerInterrupted=" + producer.isInterrupted());
    }
}

출력 큐 크기는 1이고 인터럽트 상태도 false일 수 있습니다. 예외를 잡는 과정에서 상태가 지워졌고 제출자는 실패를 알 수 없습니다. 메서드가 취소 가능하다면 InterruptedException을 선언하거나, 상위 접점에서 상태를 복구하고 명시적인 실패 결과를 반환해야 합니다. void 성공처럼 보이는 반환은 가장 피해야 할 형태입니다.

표준 큐를 선택해도 고정해야 하는 규칙

ArrayBlockingQueue 생성자는 용량과 선택적인 공정성 플래그를 받습니다. 공정 모드는 오래 기다린 작업의 접근 순서를 개선할 수 있지만 처리량 비용이 있으므로 기본으로 켜기보다 실제 기아 요구를 검토합니다. 배열 용량은 생성 후 바뀌지 않아 메모리 상한과 역압을 명시적으로 유지합니다.

put과 take는 조건이 충족될 때까지 대기하고 인터럽트에 반응하는 조합입니다.
offer와 poll은 현재 상태만 보고 즉시 결과를 반환하므로 저지연 폴백에 적합합니다.
시간 인자를 받는 오버로드는 무한 대기와 즉시 실패 사이의 제한된 양보를 표현합니다.
add와 remove는 실패를 예외로 드러내며 정상적인 과부하 분기에 남용하면 로그 소음이 커집니다.
큐는 null을 허용하지 않으므로 종료 표식도 타입으로 분리하면 데이터와 충돌하지 않습니다.
size와 remainingCapacity는 순간 관찰값이며 이후 연산 성공을 예약하지 않습니다.

메서드 호출 전에 남은 용량을 검사하고 그 결과로 put 여부를 결정하면 검사와 행동 사이 다른 생산자가 들어올 수 있습니다. 원자적 규칙은 offer 반환값 자체에 있습니다. 모니터 구현에서 배운 “조건 검사와 상태 변경을 하나로 묶는다”는 원칙이 표준 API 선택에서도 그대로 적용됩니다.

종료 메시지와 파이프라인

문자열 "STOP"을 독약으로 쓰면 실제 데이터와 충돌합니다. sealed 타입으로 작업과 종료를 구별하고 소비자 수만큼 종료 메시지를 넣습니다. 한 소비자가 종료 토큰을 가져가면 다른 소비자에게 자동 전달되지 않기 때문입니다. 제출과 소비는 인터럽트를 호출 경계까지 전파합니다.

src/BlockingQueuePipeline.java

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

public final class BlockingQueuePipeline {
    interface Message {}
    record Work(int value) implements Message {}
    enum Stop implements Message { INSTANCE }

    public static void main(String[] args) throws InterruptedException {
        int consumerCount = 3;
        BlockingQueue<Message> queue = new ArrayBlockingQueue<>(8);
        long[] partial = new long[consumerCount];
        List<Thread> consumers = new ArrayList<>();

        for (int id = 0; id < consumerCount; id++) {
            int slot = id;
            consumers.add(Thread.ofPlatform().name("consumer-" + id).start(() -> {
                try {
                    long local = 0;
                    while (true) {
                        Message message = queue.take();
                        if (message == Stop.INSTANCE) {
                            partial[slot] = local;
                            return;
                        }
                        local += ((Work) message).value();
                    }
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }));
        }

        for (int value = 1; value <= 100; value++) {
            queue.put(new Work(value));
        }
        for (int i = 0; i < consumerCount; i++) {
            queue.put(Stop.INSTANCE);
        }
        for (Thread consumer : consumers) {
            consumer.join();
        }
        long sum = 0;
        for (long value : partial) sum += value;
        System.out.println("sum=" + sum);
        System.out.println("remaining=" + queue.size());
    }
}

합계는 5,050이고 큐는 비어야 합니다. 종료 토큰은 앞선 작업 뒤에 FIFO 순서로 들어가므로 각 소비자는 이미 접수된 메시지를 건너뛰지 않습니다. 다만 생산자가 여러 명이고 종료 시점이 조정되지 않으면 한 생산자가 Stop을 넣은 뒤 다른 생산자가 Work를 넣을 수 있습니다. 제출 종료를 한 조정자가 확정하거나 큐를 감싸는 수명주기 계층을 둬야 합니다.

용량 산정 기준

큐 크기를 메모리 여유만 보고 정하면 오래 기다린 작업이 쌓일 수 있습니다. 대략적인 출발점은 초당 소비 처리량과 허용 대기 시간의 곱입니다. 초당 200건을 처리하고 큐 대기를 2초 안으로 제한하려면 정상 폭주 흡수량은 약 400건입니다. 실제 작업 크기 분포와 처리 편차를 포함해 부하 실험으로 조정합니다.

src/QueueCapacityEstimator.java

import java.time.Duration;

public final class QueueCapacityEstimator {
    static int estimate(double consumedPerSecond, Duration allowedQueueDelay, double burstFactor) {
        if (!(consumedPerSecond > 0)) throw new IllegalArgumentException("consumedPerSecond");
        if (allowedQueueDelay.isNegative() || allowedQueueDelay.isZero()) {
            throw new IllegalArgumentException("allowedQueueDelay");
        }
        if (burstFactor < 1.0) throw new IllegalArgumentException("burstFactor");
        double seconds = allowedQueueDelay.toNanos() / 1_000_000_000.0;
        long estimated = (long) Math.ceil(consumedPerSecond * seconds * burstFactor);
        if (estimated > Integer.MAX_VALUE) throw new IllegalArgumentException("capacity overflow");
        return Math.toIntExact(estimated);
    }

    public static void main(String[] args) {
        int capacity = estimate(200.0, Duration.ofSeconds(2), 1.25);
        System.out.println("capacity=" + capacity);
    }
}

예상값은 500입니다. 이 계산은 확정 공식이 아니라 측정 계획을 세우는 출발점입니다. 생산률이 장기적으로 소비률보다 높으면 어떤 용량도 결국 차므로 생산 제한, 수평 확장, 작업 축약 같은 구조적 대응이 필요합니다. 큐 지연 백분위와 거부율을 함께 감시합니다.

직접 구현과 표준 구현 사이의 판단

요구	권장 출발점	근거
고정 용량 FIFO	ArrayBlockingQueue	메모리 상한과 조건 대기
거의 무제한 연결 큐	LinkedBlockingQueue 용량 지정	기본 무제한 사용 주의
우선순위 소비	PriorityBlockingQueue	동일 우선순위 안정성 별도 검토
생산자와 소비자 직접 교환	SynchronousQueue	내부 저장 없이 rendezvous

래퍼는 도메인 타입 변환, 지표, 수명주기처럼 실제 책임이 있을 때만 둡니다. 단지 메서드 이름을 바꾸고 예외를 숨기는 계층은 표준 문서를 가리고 오류 가능성을 늘립니다. 큐 타입을 필드 인터페이스로 두면 테스트나 정책 변경이 필요할 때 구현을 교체하기도 쉽습니다.

연습 문제

감사 이벤트는 100밀리초까지만 큐 빈칸을 기다립니다. 그 안에 수용되지 않으면 스레드를 더 막지 말고 List 기반 실패 저장소에 이벤트와 거부 시각을 남기세요. 인터럽트는 거부로 바꾸지 말고 상위 호출자에게 전달합니다. 성공 건수와 폴백 건수를 출력합니다.

정답과 해설

시간 제한 offer가 false를 반환할 때만 폴백합니다. 실패 저장소에는 여러 생산자가 접근할 수 있으므로 동시성 컬렉션이나 명시적 잠금이 필요합니다. 예제는 ConcurrentLinkedQueue를 사용해 거부 기록을 보존합니다.

exercise/TimedAuditSubmissionSolution.java

import java.time.Duration;
import java.time.Instant;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.TimeUnit;

public final class TimedAuditSubmissionSolution {
    record AuditEvent(String action) {}
    record Rejected(AuditEvent event, Instant at) {}

    private final BlockingQueue<AuditEvent> accepted = new ArrayBlockingQueue<>(1);
    private final ConcurrentLinkedQueue<Rejected> fallback = new ConcurrentLinkedQueue<>();

    boolean submit(AuditEvent event, Duration timeout) throws InterruptedException {
        boolean stored = accepted.offer(event, timeout.toNanos(), TimeUnit.NANOSECONDS);
        if (!stored) fallback.add(new Rejected(event, Instant.now()));
        return stored;
    }

    public static void main(String[] args) throws InterruptedException {
        TimedAuditSubmissionSolution service = new TimedAuditSubmissionSolution();
        int success = 0;
        if (service.submit(new AuditEvent("login"), Duration.ofMillis(100))) success++;
        if (service.submit(new AuditEvent("logout"), Duration.ofMillis(1))) success++;
        System.out.println("success=" + success);
        System.out.println("fallback=" + service.fallback.size());
    }
}

첫 이벤트는 수용되고 두 번째는 매우 짧은 제한 뒤 폴백되어 success=1, fallback=1이 됩니다. 실제 감사 데이터라면 메모리 목록도 프로세스 장애에 취약하므로 내구 저장소를 사용해야 합니다. 연습의 핵심은 시간 초과와 취소를 서로 다른 사건으로 유지하는 것입니다.

위임 뒤에도 남는 설계 책임

표준 큐는 락 반납, 조건 신호, 임의 복귀 대응 같은 저수준 정확성을 맡습니다. 애플리케이션은 수용 지연, 거부 처리, 종료 순서, 지표를 결정합니다. 두 책임을 분리하면 직접 구현 코드는 줄고 운영 정책은 오히려 더 선명해집니다.

BlockingQueue를 선택했다는 사실만으로 무손실을 보장하지 않습니다. 어떤 삽입 메서드를 호출하는지와 실패 반환을 어떻게 처리하는지가 결과를 정합니다. API 조합을 제품 요구와 연결해 문서화하세요.