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

이번 실행은 queued=1, producerInterrupted=false를 출력하고 종료 코드 0으로 끝났습니다. stderr의 ignored cancellation: lost-task는 catch가 남긴 진단이며, 제출 실패를 호출자에게 전파한 것은 아닙니다.

InterruptedException이 던져질 때 인터럽트 상태가 지워지고, 이 catch는 상태를 복구하지 않습니다. submit의 정상 반환에는 수용 여부가 담기지 않습니다.

메서드가 취소 가능하다면 InterruptedException을 선언하거나, 상위 접점에서 상태를 복구하고 명시적인 실패 결과를 반환해야 합니다.

void 성공처럼 보이는 반환은 가장 피해야 할 형태입니다.

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

ArrayBlockingQueue 생성자는 용량과 선택적인 공정성 플래그를 받습니다.

공정 모드는 오래 기다린 작업의 접근 순서를 개선할 수 있지만 처리량 비용이 있으므로 기본으로 켜기보다 실제 기아 요구를 검토합니다.

배열 용량은 생성 후 바뀌지 않아 메모리 상한과 역압을 명시적으로 유지합니다.

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

메서드 호출 전에 남은 용량을 검사하고 그 결과로 put 여부를 결정하면 검사와 행동 사이 다른 생산자가 들어올 수 있습니다.

원자적 규칙은 offer 반환값 자체에 있습니다.

모니터 구현에서 배운 “조건 검사와 상태 변경을 하나로 묶는다”는 원칙이 표준 API 선택에서도 그대로 적용됩니다.

종료 메시지와 파이프라인

문자열 "STOP"을 독약으로 쓰면 실제 데이터와 충돌합니다.

원문은 일반 Message 인터페이스 아래 Work 레코드와 Stop 열거형으로 작업과 종료를 구별하고, 소비자 수만큼 종료 메시지를 넣습니다.

한 소비자가 종료 토큰을 가져가면 다른 소비자에게 자동 전달되지 않기 때문입니다.

main의 제출은 인터럽트를 상위로 전파하고, 소비자는 catch에서 인터럽트 상태를 복구한 뒤 종료합니다.

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

작업 인출과 소비자 종료·전체 집계의 경계를 나눈다
코드 구간	큐와 소비자가 하는 일	이 시점이 뜻하는 완료
main의 Work 제출	1~100을 담은 Work 100개를 순서대로 put	제출 완료 · 소비자의 합산은 진행 중일 수 있음
main의 Stop 제출	`Stop.INSTANCE` 3개 · 소비자마다 하나를 꺼내 종료	한 소비자의 종료가 다른 소비자의 처리 완료는 아님
main의 join 3회 뒤	각 소비자가 저장한 partial을 합산	모든 소비자 종료 뒤 전체 합계와 남은 큐를 조회

main의 Work 제출: 큐와 소비자가 하는 일: 1~100을 담은 Work 100개를 순서대로 put
이 시점이 뜻하는 완료: 제출 완료 · 소비자의 합산은 진행 중일 수 있음
main의 Stop 제출: 큐와 소비자가 하는 일: Stop.INSTANCE 3개 · 소비자마다 하나를 꺼내 종료
이 시점이 뜻하는 완료: 한 소비자의 종료가 다른 소비자의 처리 완료는 아님
main의 join 3회 뒤: 큐와 소비자가 하는 일: 각 소비자가 저장한 partial을 합산
이 시점이 뜻하는 완료: 모든 소비자 종료 뒤 전체 합계와 남은 큐를 조회

이번 실행은 sum=5050, remaining=0을 출력했습니다. 단일 main이 모든 Work 뒤에 Stop을 넣으며, 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);
    }
}

이번 기본 입력의 출력은 capacity=500입니다.

이 계산은 지연 상한을 보장하는 공식이 아니라 측정 계획의 출발점입니다. 또한 burstFactor의 NaN은 원문의 < 1.0 검사를 통과하므로 입력 검증의 완성본으로 보아서는 안 됩니다. 이 main은 해당 경계를 실행하지 않습니다.

생산률이 장기적으로 소비률보다 높으면 어떤 용량도 결국 차므로 생산 제한, 수평 확장, 작업 축약 같은 구조적 대응이 필요합니다.

큐 지연 백분위와 거부율을 함께 감시합니다.

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

요구	권장 출발점	근거
고정 용량 FIFO	ArrayBlockingQueue	메모리 상한과 조건 대기
노드 기반 유한 큐	LinkedBlockingQueue 생성 시 용량 지정	용량은 생성 뒤 바뀌지 않음
우선순위 소비	PriorityBlockingQueue	동일 우선순위 안정성 별도 검토
생산자와 소비자 직접 교환	SynchronousQueue	내부 저장 없이 rendezvous
지연이 만료된 항목 소비	DelayQueue	take·poll은 만료된 선두를 제거

래퍼는 도메인 타입 변환, 지표, 수명주기처럼 실제 책임이 있을 때만 둡니다.

단지 메서드 이름을 바꾸고 예외를 숨기는 계층은 표준 문서를 가리고 오류 가능성을 늘립니다.

큐 타입을 필드 인터페이스로 두면 테스트나 정책 변경이 필요할 때 구현을 교체하기도 쉽습니다.

연습 문제

감사 이벤트는 100밀리초까지만 큐 빈칸을 기다립니다.

그 안에 수용되지 않으면 스레드를 더 막지 말고 동시성 메모리 컬렉션에 이벤트와 거부 시각을 남기세요.

인터럽트는 거부로 바꾸지 말고 상위 호출자에게 전달합니다.

성공 건수와 폴백 건수를 출력합니다.

정답과 해설

시간 제한 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());
    }
}

main은 첫 호출에 100ms, 두 번째 호출에 1ms를 요청했고 이번 출력은 success=1, fallback=1입니다. 이 요청값은 전체 메서드 반환 시각의 상한이 아니며, 인터럽트 분기는 이 main에서 실행하지 않습니다.

실제 감사 데이터라면 메모리 목록도 프로세스 장애에 취약하므로 내구 저장소를 사용해야 합니다.

연습의 핵심은 시간 초과와 취소를 서로 다른 사건으로 유지하는 것입니다.

위임 뒤에도 남는 설계 책임

표준 큐는 락 반납, 조건 신호, 임의 복귀 대응 같은 저수준 정확성을 맡습니다.

애플리케이션은 수용 지연, 거부 처리, 종료 순서, 지표를 결정합니다.

두 책임을 분리하면 직접 구현 코드는 줄고 운영 정책은 오히려 더 선명해집니다.

BlockingQueue를 선택했다는 사실만으로 무손실을 보장하지 않습니다.

어떤 삽입 메서드를 호출하는지와 실패 반환을 어떻게 처리하는지가 결과를 정합니다.

API 조합을 제품 요구와 연결해 문서화하세요.